public void checkNewWorkers ( Map conf ) throws IOException , InterruptedException { Set < String > workers = new HashSet <> ( ) ; for ( Entry < String , Pair < Integer , Integer > > entry : workerIdToStartTimeAndPort . entrySet ( ) ) { String workerId = entry . getKey ( ) ; int startTime = entry . getValue ( ) . getFirst ( ) ; LocalState ls = StormConfig . worker_state ( conf , workerId ) ; WorkerHeartbeat whb = ( WorkerHeartbeat ) ls . get ( Common . LS_WORKER_HEARTBEAT ) ; if ( whb == null ) { if ( ( TimeUtils . current_time_secs ( ) - startTime ) < JStormUtils . parseInt ( conf . get ( Config . SUPERVISOR_WORKER_START_TIMEOUT_SECS ) ) ) { LOG . info ( workerId + " still hasn't started" ) ; } else { LOG . error ( "Failed to start Worker " + workerId ) ; workers . add ( workerId ) ; } } else { LOG . info ( "Successfully started worker " + workerId ) ; workers . add ( workerId ) ; } } for ( String workerId : workers ) { Integer port = this . workerIdToStartTimeAndPort . get ( workerId ) . getSecond ( ) ; this . workerIdToStartTimeAndPort . remove ( workerId ) ; this . portToWorkerId . remove ( port ) ; } }
check whether timestamp of new workers is not > SUPERVISOR_WORKER_START_TIMEOUT_SECS otherwise mark as failed