@ SuppressWarnings ( "unchecked" ) private void checkNeedUpdateTopologies ( Map < String , StateHeartbeat > localWorkerStats , Map < Integer , LocalAssignment > localAssignments ) throws Exception { Set < String > topologies = new HashSet <> ( ) ; for ( Map . Entry < Integer , LocalAssignment > entry : localAssignments . entrySet ( ) ) { topologies . add ( entry . getValue ( ) . getTopologyId ( ) ) ; } for ( StateHeartbeat stateHb : localWorkerStats . values ( ) ) { State state = stateHb . getState ( ) ; if ( ! state . equals ( State . notStarted ) ) { String topologyId = stateHb . getHeartbeat ( ) . getTopologyId ( ) ; topologies . remove ( topologyId ) ; } } long currTime = System . currentTimeMillis ( ) ; Set < String > needRemoveTopologies = new HashSet <> ( ) ; for ( String topologyId : topologies ) { try { long lastModifyTime = StormConfig . get_supervisor_topology_Bianrymodify_time ( conf , topologyId ) ; if ( ( currTime - lastModifyTime ) / 1000 < ( JStormUtils . MIN_1 * 2 ) ) { LOG . debug ( "less than 2 minute, removing " + topologyId ) ; needRemoveTopologies . add ( topologyId ) ; } } catch ( Exception e ) { LOG . error ( "Failed to get last modified time for topology" + topologyId , e ) ; needRemoveTopologies . add ( topologyId ) ; } } topologies . removeAll ( needRemoveTopologies ) ; if ( topologies . size ( ) > 0 ) { LOG . debug ( "Following topologies are going to re-download jars, " + topologies ) ; } needDownloadTopologies . set ( topologies ) ; }
check all workers is failed or not