RandomBlockingQueueRecordWriter implements RecordWriter<P> { @Override public void writeRecords(Batch<P> batch) throws Exception { for (Record<P> record : batch) { BlockingQueue<Record<P>> queue = queues.get(random.nextInt(queuesNumber)); queue.put(record); } } RandomBlockingQueueRecordWriter(List<BlockingQueue<Record<P>>> queues); @Override void writeRecords(Batch<P> batch);  }
@Test public void recordsShouldBeWrittenRandomlyToOneOfTheQueues() throws Exception { randomQueueRecordWriter.writeRecords(new Batch(record)); if (queue1.isEmpty()) { assertThat(queue2).containsOnly(record); } else { assertThat(queue2).isEmpty(); assertThat(queue1).containsOnly(record); } }