code: public Assignment mkAssignment(TopologyAssignEvent event) throws Exception {
          String topologyId = event.getTopologyId();
          LOG.info("Determining assignment for " + topologyId);
          TopologyAssignContext context = prepareTopologyAssign(event);
          Set<ResourceWorkerSlot> assignments;
          if (!StormConfig.local_mode(nimbusData.getConf())) {
              IToplogyScheduler scheduler = schedulers.get(DEFAULT_SCHEDULER_NAME);
              assignments = scheduler.assignTasks(context);
          } else {
              assignments = mkLocalAssignment(context);
          }
  
          Assignment assignment = null;
          if (assignments != null && assignments.size() > 0) {
              Map<String, String> nodeHost = getTopologyNodeHost(
                      context.getCluster(), context.getOldAssignment(), assignments);
              Map<Integer, Integer> startTimes = getTaskStartTimes(
                      context, nimbusData, topologyId, context.getOldAssignment(), assignments);
  
              String codeDir = (String) nimbusData.getConf().get(Config.STORM_LOCAL_DIR);
  
              assignment = new Assignment(codeDir, assignments, nodeHost, startTimes);
  
              //  the topology binary changed.
              if (event.isScaleTopology()) {
                  assignment.setAssignmentType(Assignment.AssignmentType.ScaleTopology);
              }
              StormClusterState stormClusterState = nimbusData.getStormClusterState();
  
              stormClusterState.set_assignment(topologyId, assignment);
  
              // update task heartbeat's start time
              NimbusUtils.updateTaskHbStartTime(nimbusData, assignment, topologyId);
  
              // Update metrics information in ZK when rebalance or reassignment
              // Only update metrics monitor status when creating topology
              // if (context.getAssignType() ==
              // TopologyAssignContext.ASSIGN_TYPE_REBALANCE
              // || context.getAssignType() ==
              // TopologyAssignContext.ASSIGN_TYPE_MONITOR)
              // NimbusUtils.updateMetricsInfo(nimbusData, topologyId, assignment);
  
              NimbusUtils.updateTopologyTaskTimeout(nimbusData, topologyId);
  
              LOG.info("Successfully make assignment for topology id " + topologyId + ": " + assignment);
          }
          return assignment;
      }
docs: make assignments for a topology The nimbus core function, this function has been totally rewrite
  
  @throws Exception