private Set < String > get_cleanup_ids ( StormClusterState clusterState , List < String > activeTopologies ) throws Exception { List < String > task_ids = clusterState . task_storms ( ) ; List < String > heartbeat_ids = clusterState . heartbeat_storms ( ) ; List < String > error_ids = clusterState . task_error_storms ( ) ; List < String > assignment_ids = clusterState . assignments ( null ) ; List < String > metric_ids = clusterState . get_metrics ( ) ; HashSet < String > latest_code_ids = new HashSet <> ( ) ; Set < String > code_ids = BlobStoreUtils . code_ids ( nimbusData . getBlobStore ( ) ) ; Set < String > to_cleanup_ids = new HashSet <> ( ) ; Set < String > pendingTopologies = nimbusData . getPendingSubmitTopologies ( ) . buildMap ( ) . keySet ( ) ; if ( task_ids != null ) { to_cleanup_ids . addAll ( task_ids ) ; } if ( heartbeat_ids != null ) { to_cleanup_ids . addAll ( heartbeat_ids ) ; } if ( error_ids != null ) { to_cleanup_ids . addAll ( error_ids ) ; } if ( assignment_ids != null ) { to_cleanup_ids . addAll ( assignment_ids ) ; } if ( code_ids != null ) { to_cleanup_ids . addAll ( code_ids ) ; } if ( metric_ids != null ) { to_cleanup_ids . addAll ( metric_ids ) ; } if ( activeTopologies != null ) { to_cleanup_ids . removeAll ( activeTopologies ) ; latest_code_ids . removeAll ( activeTopologies ) ; } to_cleanup_ids . removeAll ( pendingTopologies ) ; /** * Why need to remove latest code. Due to competition between Thrift.threads and TopologyAssign thread */ to_cleanup_ids . removeAll ( latest_code_ids ) ; LOG . info ( "Skip removing topology of " + latest_code_ids ) ; return to_cleanup_ids ; }
get topology ids that need to be cleaned up