code: public void backupAssignment(Assignment assignment, TopologyAssignEvent event) {
          String topologyId = event.getTopologyId();
          String topologyName = event.getTopologyName();
          try {
  
              StormClusterState zkClusterState = nimbusData.getStormClusterState();
              // one little problem, get tasks twice when assign one topology
              Map<Integer, String> tasks = Cluster.get_all_task_component(zkClusterState, topologyId, null);
  
              Map<String, List<Integer>> componentTasks = JStormUtils.reverse_map(tasks);
  
              for (Entry<String, List<Integer>> entry : componentTasks.entrySet()) {
                  List<Integer> keys = entry.getValue();
                  Collections.sort(keys);
              }
  
              AssignmentBak assignmentBak = new AssignmentBak(componentTasks, assignment);
              zkClusterState.backup_assignment(topologyName, assignmentBak);
          } catch (Exception e) {
              LOG.warn("Failed to backup " + topologyId + " assignment " + assignment, e);
          }
      }
docs: Backup topology assignment to ZK
  
  todo: Do we need to do backup operation every time?