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 ; }
make assignments for a topology The nimbus core function this function has been totally rewrite