code: @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);
      }
docs: check all workers is failed or not