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;
  }
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
This method should be called periodically from a scheduler if the pipeline should not allow runners to be idle for more then idleTime .