container/src/main/java/com/streamsets/datacollector/execution/runner/common/ProductionPipelineRunner.java
@Override
  public void destroy(
    SourcePipe originPipe,
    List<PipeRunner> pipeRunners,
    BadRecordsHandler badRecordsHandler,
    StatsAggregationHandler statsAggregationHandler
  ) throws StageException, PipelineRuntimeException {
    // We're no longer running
    running = false;

    // There are two ways a runner can be in use - used by real runner (e.g. when origin produced data) or when it's
    // processing "empty" batch when the runner was idle for too long. The first case is guarded by the framework - this
    // method won't be called until the execution successfully finished. However the second way with idle drivers is run
    // from a separate thread and hence we have to ensure here that it's "done" before moving on with the destroy phase.
    try {
      destroyLock.lock();

      // Firstly destroy the runner, to make sure that any potential run away thread from origin will be denied
      // further processing.
      if (runnerPool != null) {
        runnerPool.destroy();
      }

      int batchSize = configuration.get(Constants.MAX_BATCH_SIZE_KEY, Constants.MAX_BATCH_SIZE_DEFAULT);
      long lastBatchTime = offsetTracker.getLastBatchTime();
      long start = System.currentTimeMillis();
      FullPipeBatch pipeBatch;

      // Destroy origin pipe
      pipeBatch = new FullPipeBatch(null, null, batchSize, false);
      try {
        LOG.trace("Destroying origin pipe");
        pipeBatch.skipStage(originPipe);
        originPipe.destroy(pipeBatch);
      } catch (RuntimeException e) {
        LOG.warn("Exception throw while destroying pipe", e);
      }

      // Now destroy the pipe runners
      //
      // We're destroying them in reverser order to make sure that the last runner to destroy is the one with id '0'
      // that holds reference to all class loaders. Runners with id >0 do not own their class loaders and hence needs to
      // be destroyed before the runner with id '0'.
      for (PipeRunner pipeRunner : Lists.reverse(pipeRunners)) {
        final FullPipeBatch finalPipeBatch = pipeBatch;

        pipeRunner.executeBatch(null, null, start, pipe -> {
          // Set the last batch time in the stage context of each pipe
          ((StageContext) pipe.getStage().getContext()).setLastBatchTime(lastBatchTime);
          String instanceName = pipe.getStage().getConfiguration().getInstanceName();

          if (pipe instanceof StagePipe) {
            // Stage pipes are processed only if they are in event path
            if (pipe.getStage().getConfiguration().isInEventPath()) {
              LOG.trace("Stage pipe {} is in event path, running last process", instanceName);
              pipe.process(finalPipeBatch);
            } else {
              LOG.trace("Stage pipe {} is in data path, skipping it's processing.", instanceName);
              finalPipeBatch.skipStage(pipe);
            }
          } else {
            // Non stage pipes are executed always
            LOG.trace("Non stage pipe {}, running last process", instanceName);
            pipe.process(finalPipeBatch);
          }

          // And finally destroy the pipe
          try {
            LOG.trace("Running destroy for {}", instanceName);
            pipe.destroy(finalPipeBatch);
          } catch (RuntimeException e) {
            LOG.warn("Exception throw while destroying pipe", e);
          }
        });

        badRecordsHandler.handle(null, null, pipeBatch.getErrorSink(), pipeBatch.getSourceResponseSink());

        // Next iteration should have new and empty PipeBatch
        pipeBatch = new FullPipeBatch(null, null, batchSize, false);
        pipeBatch.skipStage(originPipe);
      }
      if (isStatsAggregationEnabled()) {
        List<Record> stats = new ArrayList<>();
        statsAggregatorRequests.drainTo(stats);
        Object timeSeriesString = pipelineConfigBean.constants.get(MetricsEventRunnable.TIME_SERIES_ANALYSIS);
        boolean timeSeriesAnalysis = (timeSeriesString != null) ? (Boolean) timeSeriesString : true;
        String metricRegistryStr;
        try {
          metricRegistryStr = ObjectMapperFactory.get().writer().writeValueAsString(metrics);
        } catch (Exception e) {
          throw new RuntimeException(Utils.format("Error converting metric json to string: {}", e), e);
        }
        LOG.info("Queueing last batch of record to be sent to stats aggregator");
        stats.add(AggregatorUtil.createMetricJsonRecord(
            runtimeInfo.getId(),
            runtimeInfo.getMasterSDCId(),
            pipelineConfiguration.getMetadata(),
            false,
            timeSeriesAnalysis,
            true,
            metricRegistryStr
        ));

        statsAggregationHandler.handle(null, null, stats);
      }
    } finally {
        destroyLock.unlock();
    }
  }
@Override
  public void destroy(
    SourcePipe originPipe,
    List<PipeRunner> pipeRunners,
    BadRecordsHandler badRecordsHandler,
    StatsAggregationHandler statsAggregationHandler
  ) throws StageException, PipelineRuntimeException {
    // We're no longer running
    running = false;

    // There are two ways a runner can be in use - used by real runner (e.g. when origin produced data) or when it's
    // processing "empty" batch when the runner was idle for too long. The first case is guarded by the framework - this
    // method won't be called until the execution successfully finished. However the second way with idle drivers is run
    // from a separate thread and hence we have to ensure here that it's "done" before moving on with the destroy phase.
    try {
      destroyLock.lock();

      // Firstly destroy the runner, to make sure that any potential run away thread from origin will be denied
      // further processing.
      if (runnerPool != null) {
        runnerPool.destroy();
      }

      int batchSize = configuration.get(Constants.MAX_BATCH_SIZE_KEY, Constants.MAX_BATCH_SIZE_DEFAULT);
      long lastBatchTime = offsetTracker.getLastBatchTime();
      long start = System.currentTimeMillis();
      FullPipeBatch pipeBatch;

      // Destroy origin pipe
      pipeBatch = new FullPipeBatch(null, null, batchSize, false);
      try {
        LOG.trace("Destroying origin pipe");
        pipeBatch.skipStage(originPipe);
        originPipe.destroy(pipeBatch);
      } catch (RuntimeException e) {
        LOG.warn("Exception throw while destroying pipe", e);
      }

      // Now destroy the pipe runners
      //
      // We're destroying them in reverser order to make sure that the last runner to destroy is the one with id '0'
      // that holds reference to all class loaders. Runners with id >0 do not own their class loaders and hence needs to
      // be destroyed before the runner with id '0'.
      for (PipeRunner pipeRunner : Lists.reverse(pipeRunners)) {
        final FullPipeBatch finalPipeBatch = pipeBatch;

        pipeRunner.executeBatch(null, null, start, pipe -> {
          // Set the last batch time in the stage context of each pipe
          ((StageContext) pipe.getStage().getContext()).setLastBatchTime(lastBatchTime);
          String instanceName = pipe.getStage().getConfiguration().getInstanceName();

          if (pipe instanceof StagePipe) {
            // Stage pipes are processed only if they are in event path
            if (pipe.getStage().getConfiguration().isInEventPath()) {
              LOG.trace("Stage pipe {} is in event path, running last process", instanceName);
              pipe.process(finalPipeBatch);
            } else {
              LOG.trace("Stage pipe {} is in data path, skipping it's processing.", instanceName);
              finalPipeBatch.skipStage(pipe);
            }
          } else {
            // Non stage pipes are executed always
            LOG.trace("Non stage pipe {}, running last process", instanceName);
            pipe.process(finalPipeBatch);
          }

          // And finally destroy the pipe
          try {
            LOG.trace("Running destroy for {}", instanceName);
            pipe.destroy(finalPipeBatch);
          } catch (RuntimeException e) {
            LOG.warn("Exception throw while destroying pipe", e);
          }
        });

        badRecordsHandler.handle(null, null, pipeBatch.getErrorSink(), pipeBatch.getSourceResponseSink());

        // Next iteration should have new and empty PipeBatch
        pipeBatch = new FullPipeBatch(null, null, batchSize, false);
        pipeBatch.skipStage(originPipe);
      }
      if (isStatsAggregationEnabled()) {
        List<Record> stats = new ArrayList<>();
        statsAggregatorRequests.drainTo(stats);
        Object timeSeriesString = pipelineConfigBean.constants.get(MetricsEventRunnable.TIME_SERIES_ANALYSIS);
        boolean timeSeriesAnalysis = (timeSeriesString != null) ? (Boolean) timeSeriesString : true;
        String metricRegistryStr;
        try {
          metricRegistryStr = ObjectMapperFactory.get().writer().writeValueAsString(metrics);
        } catch (Exception e) {
          throw new RuntimeException(Utils.format("Error converting metric json to string: {}", e), e);
        }
        LOG.info("Queueing last batch of record to be sent to stats aggregator");
        stats.add(AggregatorUtil.createMetricJsonRecord(
            runtimeInfo.getId(),
            runtimeInfo.getMasterSDCId(),
            pipelineConfiguration.getMetadata(),
            false,
            timeSeriesAnalysis,
            true,
            metricRegistryStr
        ));

        statsAggregationHandler.handle(null, null, stats);
      }
    } finally {
        destroyLock.unlock();
    }
  }
Since stages are allowed to produce events during destroy() phase, we handle the destroy event as simplified
runBatch. We go over all the StagePipes and if it's on data path we destroy it immediately, if it's on  event path, we
run it one more time. Since the stages are sorted we know that destroyed stage will never be needed again. Non
stage pipes are always processed to generate required structures in PipeBatch.