JmsRecordWriter implements RecordWriter<Message> { @Override public void writeRecords(Batch<Message> batch) throws Exception { for (Record<Message> record : batch) { messageProducer.send(record.getPayload()); } } JmsRecordWriter(final ConnectionFactory connectionFactory, final Destination destination); @Override void open(); @Override void writeRecords(Batch<Message> batch); @Override void close();  }
@Test public void testProcessRecord() throws Exception { jmsRecordWriter.writeRecords(new Batch<>(record)); verify(messageProducer).send(message); }
@Test(expected = Exception.class) public void testRecordProcessingWithError() throws Exception { doThrow(jmsException).when(messageProducer).send(message); jmsRecordWriter.writeRecords(new Batch<>(record)); }