@Test
public void run() throws Exception {

  Request request = new Request(1L);
  JsonNode value = mapper.createObjectNode();
  Payload<JsonNode> payload = new Payload<JsonNode>((byte) 0, 2, value);
  Record record = new Record(0, 1L, 3L, payload);
  doNothing().when(underTest).sendEvent(any());
  doNothing().when(underTest).sendRebalance(any());
  doAnswer(i -> {
    underTest.getBuffer().offer(record);
    return null;
  }).when(underTest).replenishBuffer();
  try {
    Mono.fromRunnable(underTest).subscribeOn(single()).subscribe();
    underTest.getEvents().offer(request);
    Awaitility.await().atMost(500, MILLISECONDS).pollInterval(10, MILLISECONDS).until(() -> {
      verify(underTest).handleIncomingEvent(request);
    });
    Awaitility.await().atMost(500, MILLISECONDS).pollInterval(10, MILLISECONDS).until(() -> {
      verify(underTest).sendMessage(record);
    });
  } finally {
    underTest.close();
  }
  verify(underTest).sendEvent(new Connection(podName));
  ArgumentCaptor<RebalanceListener> captor = ArgumentCaptor.forClass(RebalanceListener.class);
  verify(consumer).init(eq(1L), captor.capture());
  Set<Integer> assignment = singleton(0);
  captor.getValue().onRebalance(assignment);
  verify(underTest).sendRebalance(assignment);
}