RoundRobinBlockingQueueRecordWriter implements RecordWriter<P> { @Override public void writeRecords(Batch<P> batch) throws Exception { for (Record<P> record : batch) { BlockingQueue<Record<P>> queue = queues.get(nextQueue++ % queuesNumber); queue.put(record); } } RoundRobinBlockingQueueRecordWriter(List<BlockingQueue<Record<P>>> queues); @Override void writeRecords(Batch<P> batch);  }
@Test public void recordsShouldBeWrittenToQueuesInRoundRobinFashion() throws Exception { roundRobinQueueRecordWriter.writeRecords(new Batch(record1, record2, record3, record4)); assertThat(queue1).containsExactly(record1, record3); assertThat(queue2).containsExactly(record2, record4); }