container/src/main/java/com/streamsets/datacollector/execution/runner/common/ProductionPipelineRunner.java
public int produceEmptyBatchesForIdleRunners(long idleTime) throws PipelineException, StageException {
    LOG.debug("Checking if any active runner is idle");

    // The empty batch is suppose to be fast - almost as a zero time. It could however happened that from some reason it
    // will take a long time (possibly more then idleTime). To avoid infinite loops, this method will only processes up
    // to total number of runners before returning.
    int counter = 0;

    try {
      destroyLock.lock();

      while(running && counter < pipes.size()) {
        counter++;

        PipeRunner runner = null;
        try {
          runner = runnerPool.getIdleRunner(idleTime);

          // No more idle runners, simply stop the idle execution now
          if(runner == null) {
            return counter;
          }

          LOG.debug("Generating empty batch for runner: {}", runner.getRunnerId());
          pipeContext.getRuntimeStats().incIdleBatchCount();

          // Pipe batch to keep the batch info
          FullPipeBatch pipeBatch = new FullPipeBatch(null, null, 0, false);
          pipeBatch.setIdleBatch(true);

          // We're explicitly skipping origin because this is framework generated, empty batch
          pipeBatch.skipStage(originPipe);

          executeRunner(
            runner,
            System.currentTimeMillis(),
            pipeBatch,
            null,
            null,
            new HashMap<>(),
            new HashMap<>()
          );
        } finally {
          if(runner != null) {
            runnerPool.returnRunner(runner);
          }
        }
      }
    } finally {
      destroyLock.unlock();
    }

    return counter;
  }
public int produceEmptyBatchesForIdleRunners(long idleTime) throws PipelineException, StageException {
    LOG.debug("Checking if any active runner is idle");

    // The empty batch is suppose to be fast - almost as a zero time. It could however happened that from some reason it
    // will take a long time (possibly more then idleTime). To avoid infinite loops, this method will only processes up
    // to total number of runners before returning.
    int counter = 0;

    try {
      destroyLock.lock();

      while(running && counter < pipes.size()) {
        counter++;

        PipeRunner runner = null;
        try {
          runner = runnerPool.getIdleRunner(idleTime);

          // No more idle runners, simply stop the idle execution now
          if(runner == null) {
            return counter;
          }

          LOG.debug("Generating empty batch for runner: {}", runner.getRunnerId());
          pipeContext.getRuntimeStats().incIdleBatchCount();

          // Pipe batch to keep the batch info
          FullPipeBatch pipeBatch = new FullPipeBatch(null, null, 0, false);
          pipeBatch.setIdleBatch(true);

          // We're explicitly skipping origin because this is framework generated, empty batch
          pipeBatch.skipStage(originPipe);

          executeRunner(
            runner,
            System.currentTimeMillis(),
            pipeBatch,
            null,
            null,
            new HashMap<>(),
            new HashMap<>()
          );
        } finally {
          if(runner != null) {
            runnerPool.returnRunner(runner);
          }
        }
      }
    } finally {
      destroyLock.unlock();
    }

    return counter;
  }
This method should be called periodically from a scheduler if the pipeline should not allow runners to be "idle"
for more then idleTime.

@param idleTime Number of milliseconds after which a runner is considered "idle"
@return Number of idle batches generated in this iteration