private void acceptConsumer ( ThrowingConsumer < Pipe > consumer , Pipe p ) throws PipelineRuntimeException , StageException { try { // Process pipe consumer . accept ( p ) ; } catch ( Throwable t ) { String instanceName = p . getStage ( ) . getInfo ( ) . getInstanceName ( ) ; LOG . error ( "Failed executing stage '{}': {}" , instanceName , t . toString ( ) , t ) ; Throwables . propagateIfInstanceOf ( t , PipelineRuntimeException . class ) ; Throwables . propagateIfInstanceOf ( t , StageException . class ) ; Throwables . propagate ( t ) ; } }
Accept given consumer and proper log context of any exception .