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 .