@Test
public void replenishBuffer() throws Exception {

  Payload<JsonNode> payload = new Payload<JsonNode>((byte) 0, 1, mapper.createObjectNode());
  Record record = new Record(0, 1L, 3L, payload);
  List<Record> records = singletonList(record);
  doReturn(records).when(consumer).poll();
  doReturn(records).when(metrics).record(eq(POLL), argThat(new ArgMatcher<Supplier<Iterable<Record>>>() {}));
  underTest.replenishBuffer();
  @SuppressWarnings("unchecked")
  ArgumentCaptor<Supplier<Iterable<Record>>> recordsCaptor = ArgumentCaptor.forClass(Supplier.class);
  verify(metrics).record(eq(POLL), recordsCaptor.capture());
  assertThat(recordsCaptor.getValue().get(), is(records));
  ArgumentCaptor<Runnable> bufferCaptor = ArgumentCaptor.forClass(Runnable.class);
  verify(metrics).record(eq(BUFFER), bufferCaptor.capture());
  bufferCaptor.getValue().run();
  assertThat(underTest.getBuffer().peek(), is(record));
}