CompositePipelineListener implements PipelineListener { @Override public <P> void onRecordProcessingException(Record<P> record, Throwable throwable) { for (ListIterator<PipelineListener> iterator = listeners.listIterator(listeners.size()); iterator.hasPrevious();) { iterator.previous().onRecordProcessingException(record, throwable); } } CompositePipelineListener();  CompositePipelineListener(List<PipelineListener> listeners); @Override Record<P> beforeRecordProcessing(Record<P> record); @Override void afterRecordProcessing(Record<P> inputRecord, Record<P> outputRecord); @Override void onRecordProcessingException(Record<P> record, Throwable throwable); void addPipelineListener(PipelineListener pipelineListener);  }
@Test public void testOnRecordProcessingException() { compositePipelineListener.onRecordProcessingException(inputRecord, exception); InOrder inOrder = inOrder(pipelineListener1, pipelineListener2); inOrder.verify(pipelineListener2).onRecordProcessingException(inputRecord, exception); inOrder.verify(pipelineListener1).onRecordProcessingException(inputRecord, exception); }