CompositePipelineListener implements PipelineListener { @Override public <P> void afterRecordProcessing(Record<P> inputRecord, Record<P> outputRecord) { for (ListIterator<PipelineListener> iterator = listeners.listIterator(listeners.size()); iterator.hasPrevious();) { iterator.previous().afterRecordProcessing(inputRecord, outputRecord); } } 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 testAfterRecordProcessing() { compositePipelineListener.afterRecordProcessing(inputRecord, outputRecord); InOrder inOrder = inOrder(pipelineListener1, pipelineListener2); inOrder.verify(pipelineListener2).afterRecordProcessing(inputRecord, outputRecord); inOrder.verify(pipelineListener1).afterRecordProcessing(inputRecord, outputRecord); }