ConsumerRecordWriter implements Writer<ConsumerRecord<Void, Record>> { @Override public void write(ConsumerRecord<Void, Record> record) throws IOException { Schema schema = record.value().getSchema(); writers.computeIfAbsent(schema, this::newRecordWriter).write(record.value()); recordCounter.getAndIncrement(); metrics.consumedBytes(record.serializedValueSize()); metrics.offsetHighwaterMark(record.partition(), record.offset()); if (byteCounter.getAndAdd(ZERO) >= flushBytesThreshold) { flush(); } } @Autowired  ConsumerRecordWriter(
      Supplier<String> keySupplier,
      RecordWriter.Factory recordWriterFactory,
      AbortableOutputStreamFactory outputStreamFactory,
      @Value("${writer.flushBytesThreshold:134217728}") long flushBytesThreshold,
      Metrics metrics); @Override void write(ConsumerRecord<Void, Record> record); @Override void close(); @Override void flush();  }
@Test public void write_Flush() throws IOException { when(outputStreamFactory.create(LOCATION)).thenReturn(abortableOutputStream); ArgumentCaptor<OutputStream> captor = ArgumentCaptor.forClass(OutputStream.class); when(recordWriterFactory.create(eq(schema1), captor.capture())).thenReturn(recordWriter); underTest.getByteCounter().getAndAdd(3L); ConsumerRecord<Void, Record> record = record(schema1, "foo", 1, 10); underTest.write(record); verify(recordWriter).write(record.value()); assertThat(underTest.getRecordCounter().get(), is(0L)); verify(metrics).consumedBytes(10); verify(metrics).offsetHighwaterMark(0, 1); verify(metrics).uploadedBytes(3L); verify(metrics).uploadedEvents(1L); assertThat(writers.size(), is(0)); }
@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)); }