@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();
}