ConsumerRecordWriter implements Writer<ConsumerRecord<Void, Record>> { @VisibleForTesting Writer<Record> newRecordWriter(Schema schema) { String key = keySupplier.get(); AbortableOutputStream outputStream; try { outputStream = outputStreamFactory.create(key); } catch (IOException e) { throw new RuntimeException("Error creating output.", e); } LongConsumer consumer = x -> byteCounter.getAndAdd(x); ConsumerCountOutputStream output = new ConsumerCountOutputStream(outputStream, consumer); try { return recordWriterFactory.create(schema, output); } catch (IOException e) { throw new RuntimeException("Error creating writer.", e); } } @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(expected = RuntimeException.class) public void newRecordWriter_newOutputStreamException() throws IOException { doThrow(IOException.class).when(outputStreamFactory).create(LOCATION); underTest.newRecordWriter(schema1); }
@Test(expected = RuntimeException.class) public void newRecordWriter_newRecordWriterException() throws IOException { when(outputStreamFactory.create(LOCATION)).thenReturn(abortableOutputStream); doThrow(IOException.class).when(recordWriterFactory).create(eq(schema1), any(OutputStream.class)); underTest.newRecordWriter(schema1); }