code: 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;
      }
docs: get topology ids that need to be cleaned up