@Test
public void write_NoFlush() throws IOException {

  when(outputStreamFactory.create(LOCATION)).thenReturn(abortableOutputStream);
  ArgumentCaptor<OutputStream> captor = ArgumentCaptor.forClass(OutputStream.class);
  when(recordWriterFactory.create(eq(schema1), captor.capture())).thenReturn(recordWriter);
  ConsumerRecord<Void, Record> record = record(schema1, "foo", 1, 10);
  underTest.write(record);
  verify(recordWriter).write(record.value());
  assertThat(underTest.getRecordCounter().get(), is(1L));
  verify(metrics).consumedBytes(10);
  verify(metrics).offsetHighwaterMark(0, 1);
  assertThat(writers.size(), is(1));
}