container/src/main/java/com/streamsets/datacollector/runner/PipeRunner.java
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, "");
    }
  }
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, "");
    }
  }
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.

This method will also set the logger appropriately and calculate the runner specific metrics.
Run batch with given consumer for each pipe .