make assignments for a topology The nimbus core function, this function has been totally rewrite

@throws Exception
437361
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;
    }
1160899
public <T> void transitionLock(String topologyId, boolean errorOnNoTransition, StatusType changeStatus, T... args)
            throws Exception {
        // get ZK's topology node's data, which is StormBase
        StormBase stormbase = data.getStormClusterState().storm_base(topologyId, null);
        if (stormbase == null) {
            LOG.error("Cannot apply event: changing status " + topologyId + " -> " + changeStatus.getStatus() +
                    ", cause: failed to get StormBase from ZK");
            return;
        }

        StormStatus currentStatus = stormbase.getStatus();
        if (currentStatus == null) {
            LOG.error("Cannot apply event: changing status " + topologyId + " -> " + changeStatus.getStatus() +
                    ", cause: topologyStatus is null in ZK");
            return;
        }

        // <currentStatus, Map<changingStatus, callback>>
        Map<StatusType, Map<StatusType, Callback>> callbackMap = stateTransitions(topologyId, currentStatus);

        // get current changingCallbacks
        Map<StatusType, Callback> changingCallbacks = callbackMap.get(currentStatus.getStatusType());

        if (changingCallbacks == null || !changingCallbacks.containsKey(changeStatus) ||
                changingCallbacks.get(changeStatus) == null) {
            String msg = "No transition for event: changing status:" + changeStatus.getStatus() +
                    ", current status: " + currentStatus.getStatusType() + ", topology-id: " + topologyId;
            LOG.info(msg);
            if (errorOnNoTransition) {
                throw new RuntimeException(msg);
            }
            return;
        }

        Callback callback = changingCallbacks.get(changeStatus);

        Object obj = callback.execute(args);
        if (obj != null && obj instanceof StormStatus) {
            StormStatus newStatus = (StormStatus) obj;
            // update status to ZK
            data.getStormClusterState().update_storm(topologyId, newStatus);
            LOG.info("Successfully updated " + topologyId + " to status " + newStatus);
        }

        LOG.info("Successfully apply event: changing status " + topologyId + " -> " + changeStatus.getStatus());
    }
949819
public TopologyExecutionStatus getExecution(String topologyName, String nimbusHost,
            int nimbusPort, int interval, int number) throws ConnectFailException
    {

        NimbusClientFactory factory = new NimbusClientFactory();
        // RuntimeExceptionが発生
        NimbusClient nimbusClient = factory.createClient(nimbusHost, nimbusPort);
        Nimbus.Client client = nimbusClient.getClient();
        ClusterSummary clusterSummary = null;
        try
        {
            clusterSummary = client.getClusterInfo();
        }
        catch (org.apache.thrift7.TException ex)
        {
            throw new ConnectFailException(ex);
        }

        String topologyId = StormSummaryExtractor.extractStormTopologyId(clusterSummary,
                topologyName);

        if (topologyId == null)
        {
            return TopologyExecutionStatus.NOT_ALIVED;
        }

        TopologyExecutionCount compareCount = null;

        for (int getCount = 0; getCount < number; getCount++)
        {
            TopologyInfo topologyInfo = null;

            try
            {
                topologyInfo = client.getTopologyInfo(topologyId);
            }
            catch (NotAliveException ex)
            {
                return TopologyExecutionStatus.NOT_ALIVED;
            }
            catch (org.apache.thrift7.TException ex)
            {
                throw new ConnectFailException(ex);
            }

            TopologyExecutionCount nowCount = StormSummaryExtractor.convertToTotalCount(topologyInfo);
            // 初回の情報取得の場合、初回の取得結果を比較対象として設定
            if (compareCount == null)
            {
                compareCount = nowCount;
            }
            else
            {
                // 文字列表現が異なる場合、処理が継続しているため、ログ出力して「処理中」と返す。
                if (StringUtils.equals(compareCount.toString(), nowCount.toString()) == false)
                {
                    return TopologyExecutionStatus.EXECUTING;
                }
            }

            try
            {
                TimeUnit.MILLISECONDS.sleep(interval);
            }
            catch (InterruptedException ex)
            {
                continue;
            }
        }

        // 指定回数変化がなかった場合、停止として扱う。
        return TopologyExecutionStatus.STOP;
    }
692583
private void refreshEffectiveTopology() throws ParentDeadException {
    LOG.entering("OperatorTopologyImpl", "refreshEffectiveTopology", getQualifiedName());
    LOG.finest(getQualifiedName() + "Waiting to acquire topoLock");
    synchronized (topologyLock) {
      LOG.finest(getQualifiedName() + "Acquired topoLock");

      assert effectiveTopology != null;

      final Set<GroupCommunicationMessage> deletionDeltasSet = new HashSet<>();
      copyDeletionDeltas(deletionDeltasSet);

      LOG.finest(getQualifiedName() + "Updating effective topology struct with deletion msgs");
      effectiveTopology.update(deletionDeltasSet);
      LOG.finest(getQualifiedName() + "Released topoLock");
    }
    LOG.exiting("OperatorTopologyImpl", "refreshEffectiveTopology", getQualifiedName());
  }
816318
@PostConstruct
    public void start() throws Exception {
        queryMbeanObjectName = JmxUtils2.registerObject(this, "org.jmxtrans.embedded:Type=Query,id=" + id, getEmbeddedJmxTrans().getMbeanServer());


        for (OutputWriter outputWriter : outputWriters) {
            outputWriter.start();
        }
    }
330810
public SchedulerAssignment getAssignmentById(String topologyId) {
        if (this.assignments.containsKey(topologyId)) {
            return this.assignments.get(topologyId);
        }
        return null;
    }
1066997
void scheduleHeronWorkers(PackingPlan topologyPacking) throws ContainerAllocationException {
    this.componentRamMap = topologyPacking.getComponentRamDistribution();
    scheduleHeronWorkers(topologyPacking.getContainers());
  }
327278
public function run()
    {
        $this->tryAssignController();
        if (!$this->isControllerAssigned()) {
            throw new Exception("It's unable to assign controller to this route path.");
        }
        $oCtrl = $this->createCtrl();
        $oCtrl->{$this->actionName}($this->actionArgs);
    }
1359926
@Override
    public void init(TopologyBuilder builder, Instances dataset, int parallelism) {
        this.builder = builder;
        this.dataset = dataset;
//        this.parallelism = parallelism;
        this.setLayout();
    }
211301
@VisibleForTesting
  void restartTopologyHandler(String topologyName) throws TopologyRuntimeManagementException {
    assert !potentialStaleExecutionData;
    Integer containerId = Context.topologyContainerId(config);
    Scheduler.RestartTopologyRequest restartTopologyRequest =
        Scheduler.RestartTopologyRequest.newBuilder()
            .setTopologyName(topologyName)
            .setContainerIndex(containerId)
            .build();
    // If we restart the container including TMaster, wee need to clean TMasterLocation,
    // since when starting up, TMaster expects no other existing TMaster,
    // i.e. TMasterLocation does not exist
    if (containerId == -1 || containerId == 0) {
      // get the instance of state manager to clean state
      SchedulerStateManagerAdaptor stateManager = Runtime.schedulerStateManagerAdaptor(runtime);
      Boolean result = stateManager.deleteTMasterLocation(topologyName);
      if (result == null || !result) {
        throw new TopologyRuntimeManagementException(
            "Failed to clear TMaster location. Check whether TMaster set it correctly.");
      }
    }

    if (!schedulerClient.restartTopology(restartTopologyRequest)) {
      throw new TopologyRuntimeManagementException(String.format(
          "Failed to restart topology '%s'", topologyName));
    }
    // Clean the connection when we are done.
    LOG.fine("Scheduler restarted topology successfully.");
  }
1257753
private void activateTopologyHandler(String topologyName) throws TMasterException {
    assert !potentialStaleExecutionData;
    NetworkUtils.TunnelConfig tunnelConfig =
        NetworkUtils.TunnelConfig.build(config, NetworkUtils.HeronSystem.SCHEDULER);
    TMasterUtils.transitionTopologyState(topologyName,
        TMasterUtils.TMasterCommand.ACTIVATE, Runtime.schedulerStateManagerAdaptor(runtime),
        TopologyAPI.TopologyState.PAUSED, TopologyAPI.TopologyState.RUNNING, tunnelConfig);
  }
345529
public void restartTopology(String stormId){
        String currentTopology = StringUtils.substringBefore(stormId, "-");
        String topologyPrefix = StringUtils.substringBefore(currentTopology, "_");
        String topologyIndex =  StringUtils.substringAfter(currentTopology, "_");
        int newIndex = Integer.parseInt(topologyIndex) + 1;
        String newTopologyName = topologyPrefix + "_" + newIndex;
        LOGGER.info("new topology name = " + newTopologyName);
        execStartCommand(newTopologyName);
        LOGGER.info("execStartCommand finish ..");
        execShutdownCommand(currentTopology);
        LOGGER.info("execShutdownCommand finish ..");
    }
47665
public void setTopologyPaths(String[] topologyPaths) throws IOException {
    if (topologyPaths == null) {
      this.topologyPaths = new String[0];
    } else {
      this.topologyPaths = topologyPaths;
    }
  }
965073
public Boolean setTopology(TopologyAPI.Topology topology, String topologyName) {
    return awaitResult(delegate.setTopology(topology, topologyName));
  }
488933
@Override
    public void prepare(Map stormConf, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.outputCollector = outputCollector;
        this.client = clientFactory.makeClient(stormConf);
        this.queue = new LinkedBlockingQueue<>(QUEUE_MAX_SIZE);
    }
695664
private void initMappings() throws Exception {
        checkClient();
        // We extract indexes and mappings to manage from mappings definition
        if (mappings != null && mappings.length > 0) {
            Map<String, Collection<String>> indices = getIndexMappings(mappings);

            // Let's initialize indexes and mappings if needed
            for (String index : indices.keySet()) {
                createIndex(client.getLowLevelClient(), classpathRoot, index, forceMapping);
                if (mergeSettings) {
                    updateSettings(client.getLowLevelClient(), classpathRoot, index);
                }

                createMapping(client, classpathRoot, index, mergeMapping);
            }
        }
    }
737220
public Collection<ExecutorDetails> getUnassignedExecutors(TopologyDetails topology) {
        if (topology == null) {
            return new ArrayList<>(0);
        }

        Collection<ExecutorDetails> ret = new HashSet<>(topology.getExecutors());

        SchedulerAssignment assignment = this.getAssignmentById(topology.getId());
        if (assignment != null) {
            Set<ExecutorDetails> assignedExecutors = assignment.getExecutors();
            ret.removeAll(assignedExecutors);
        }

        return ret;
    }