Commits { public static Flux<Map<Integer, Long>> fromMessages(@NonNull Flux<Message<?>> messages, @NonNull Duration interval) { if (interval.toMillis() < 0L) { throw new IllegalArgumentException("Must not be a negative interval."); } return messages .window(interval) .<Map<Integer, Long>> flatMap( f -> f.collect(HashMap::new, (c, m) -> c.merge(m.getPartition(), m.getOffset() + 1L, Math::max))) .filter(x -> !x.isEmpty()); }  static Flux<Map<Integer, Long>> fromMessages(@NonNull Flux<Message<?>> messages, @NonNull Duration interval);  }
@Test(timeout = 50000) public void commit() throws Exception { Message<String> message1 = new Message<>(0, 1L, 1, 1L, "a"); Message<String> message2 = new Message<>(1, 2L, 1, 1L, "b"); Message<String> message3 = new Message<>(1, 3L, 1, 1L, "c"); Flux<Message<?>> messages = Flux.just(message1, message2, message3); Publisher<Map<Integer, Long>> batch = Commits.fromMessages(messages, Duration.ofSeconds(1)); StepVerifier.create(batch).assertNext(offsets -> { assertThat(offsets.size(), is(2)); assertThat(offsets.get(0), is(2L)); assertThat(offsets.get(1), is(4L)); }).verifyComplete(); }