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, "");
    }
  }
Run batch with given consumer for each pipe.

This method will also set the logger appropriately and calculate the runner specific metrics.