BroadcastJmsQueueRecordWriter implements RecordWriter<P> { @Override public void writeRecords(Batch<P> batch) throws Exception { for (Record<P> record : batch) { for (QueueSender queue : queues) { queue.send((Message) record.getPayload()); } } } BroadcastJmsQueueRecordWriter(List<QueueSender> queues); @Override void writeRecords(Batch<P> batch);  }
@Test public void testBroadcastJmsRecord() throws Exception { broadcastJmsQueueRecordWriter.writeRecords(new Batch(jmsRecord)); verify(queue1).send(message); verify(queue2).send(message); }