BlockingQueueRecordWriter implements RecordWriter<P> { @Override public void writeRecords(Batch<P> batch) throws Exception { for (Record<P> record : batch) { blockingQueue.offer(record, timeout, TimeUnit.MILLISECONDS); } } BlockingQueueRecordWriter(final BlockingQueue<Record<P>> blockingQueue);  BlockingQueueRecordWriter(final BlockingQueue<Record<P>> blockingQueue, final long timeout); @Override void writeRecords(Batch<P> batch); static final long DEFAULT_TIMEOUT; }
@Test public void testWriteRecords() throws Exception { writer.writeRecords(new Batch<>(record1, record2)); assertThat(queue).containsExactly(record1, record2); }