public void executeBatch ( String offsetKey , String offsetValue , long batchStartTime , ThrowingConsumer < Pipe > consumer ) throws PipelineRuntimeException , StageException { MDC . put ( LogConstants . RUNNER , String . valueOf ( runnerId ) ) ; // Persist static information for the batch (this won't change as the batch progresses) this . runtimeMetricGauge . put ( METRIC_BATCH_START_TIME , batchStartTime ) ; this . runtimeMetricGauge . put ( METRIC_OFFSET_KEY , Optional . ofNullable ( offsetKey ) . orElse ( "" ) ) ; this . runtimeMetricGauge . put ( METRIC_OFFSET_VALUE , Optional . ofNullable ( offsetValue ) . orElse ( "" ) ) ; this . runtimeMetricGauge . put ( METRIC_STAGE_START_TIME , System . currentTimeMillis ( ) ) ; try { // Run one pipe at a time for ( Pipe p : pipes ) { String instanceName = p . getStage ( ) . getInfo ( ) . getInstanceName ( ) ; this . runtimeMetricGauge . put ( METRIC_CURRENT_STAGE , instanceName ) ; MDC . put ( LogConstants . STAGE , instanceName ) ; if ( p instanceof StagePipe ) { this . runtimeMetricGauge . put ( METRIC_STAGE_START_TIME , System . currentTimeMillis ( ) ) ; } acceptConsumer ( consumer , p ) ; } // We've successfully finished batch this . runtimeMetricGauge . computeIfPresent ( METRIC_BATCH_COUNT , ( key , value ) -> ( ( long ) value ) + 1 ) ; } finally { resetBatchSpecificMetrics ( ) ; MDC . put ( LogConstants . RUNNER , "" ) ; MDC . put ( LogConstants . STAGE , "" ) ; } }
Run batch with given consumer for each pipe .