whether the topology is active by topology name

@param stormClusterState see Cluster_clj
@param topologyName      topology name
@return boolean if the storm is active, return true, otherwise return false
434508
public boolean isTopologyActive(StormClusterState stormClusterState, String topologyName) throws Exception {
        boolean rtn = false;
        if (Cluster.get_topology_id(stormClusterState, topologyName) != null) {
            rtn = true;
        }
        return rtn;
    }
979740
@SuppressWarnings("unused")
    private Map<Integer, LocalAssignment> readMyTasks(StormClusterState stormClusterState, String topologyId,
                                                      String supervisorId, Assignment assignmentInfo) throws Exception {
        Map<Integer, LocalAssignment> portTasks = new HashMap<>();

        Set<ResourceWorkerSlot> workers = assignmentInfo.getWorkers();
        if (workers == null) {
            LOG.error("No worker found for assignment {}!", assignmentInfo);
            return portTasks;
        }

        for (ResourceWorkerSlot worker : workers) {
            if (!supervisorId.equals(worker.getNodeId())) {
                continue;
            }
            portTasks.put(worker.getPort(), new LocalAssignment(
                    topologyId, worker.getTasks(), Common.topologyIdToName(topologyId), worker.getMemSize(),
                    worker.getCpu(), worker.getJvm(), assignmentInfo.getTimeStamp()));
        }

        return portTasks;
    }
1345009
private Map<StatusType, Map<StatusType, Callback>> stateTransitions(String topologyId, StormStatus currentStatus) {

        /**
         *
         * 1. Status: this status will be stored in ZK killed/inactive/active/rebalancing 2. action:
         *
         * monitor -- every Config.NIMBUS_MONITOR_FREQ_SECS seconds will trigger this only valid when current status is active inactivate -- client will trigger
         * this action, only valid when current status is active activate -- client will trigger this action only valid when current status is inactive startup
         * -- when nimbus startup, it will trigger this action only valid when current status is killed/rebalancing kill -- client kill topology will trigger
         * this action, only valid when current status is active/inactive/killed remove -- 30 seconds after client submit kill command, it will do this action,
         * only valid when current status is killed rebalance -- client submit rebalance command, only valid when current status is active/deactive do_rebalance
         * -- 30 seconds after client submit rebalance command, it will do this action, only valid when current status is rebalance
         */

        Map<StatusType, Map<StatusType, Callback>> rtn = new HashMap<>();

        // current status is active
        Map<StatusType, Callback> activeMap = new HashMap<>();
        activeMap.put(StatusType.monitor, new ReassignTransitionCallback(data, topologyId));
        activeMap.put(StatusType.inactivate, new InactiveTransitionCallback());
        activeMap.put(StatusType.startup, null);
        activeMap.put(StatusType.activate, null);
        activeMap.put(StatusType.kill, new KillTransitionCallback(data, topologyId));
        activeMap.put(StatusType.remove, null);
        activeMap.put(StatusType.rebalance, new RebalanceTransitionCallback(data, topologyId, currentStatus));
        activeMap.put(StatusType.do_rebalance, null);
        activeMap.put(StatusType.done_rebalance, null);
        activeMap.put(StatusType.update_topology, new UpdateTopologyTransitionCallback(data, topologyId, currentStatus));

        rtn.put(StatusType.active, activeMap);

        // current status is inactive
        Map<StatusType, Callback> inactiveMap = new HashMap<>();

        inactiveMap.put(StatusType.monitor, new ReassignTransitionCallback(data, topologyId, new StormStatus(StatusType.inactive)));
        inactiveMap.put(StatusType.inactivate, null);
        inactiveMap.put(StatusType.startup, null);
        inactiveMap.put(StatusType.activate, new ActiveTransitionCallback());
        inactiveMap.put(StatusType.kill, new KillTransitionCallback(data, topologyId));
        inactiveMap.put(StatusType.remove, null);
        inactiveMap.put(StatusType.rebalance, new RebalanceTransitionCallback(data, topologyId, currentStatus));
        inactiveMap.put(StatusType.do_rebalance, null);
        inactiveMap.put(StatusType.done_rebalance, null);
        inactiveMap.put(StatusType.update_topology, null);

        rtn.put(StatusType.inactive, inactiveMap);

        // current status is killed
        Map<StatusType, Callback> killedMap = new HashMap<>();

        killedMap.put(StatusType.monitor, null);
        killedMap.put(StatusType.inactivate, null);
        killedMap.put(StatusType.startup, new KillTransitionCallback(data, topologyId));
        killedMap.put(StatusType.activate, null);
        killedMap.put(StatusType.kill, new KillTransitionCallback(data, topologyId));
        killedMap.put(StatusType.remove, new RemoveTransitionCallback(data, topologyId));
        killedMap.put(StatusType.rebalance, null);
        killedMap.put(StatusType.do_rebalance, null);
        killedMap.put(StatusType.done_rebalance, null);
        killedMap.put(StatusType.update_topology, null);
        rtn.put(StatusType.killed, killedMap);

        // current status is under rebalancing
        Map<StatusType, Callback> rebalancingMap = new HashMap<>();

        StatusType rebalanceOldStatus = StatusType.active;
        if (currentStatus.getOldStatus() != null) {
            rebalanceOldStatus = currentStatus.getOldStatus().getStatusType();
            // fix double rebalance, make the status always as rebalancing
            if (rebalanceOldStatus == StatusType.rebalancing) {
                rebalanceOldStatus = StatusType.active;
            }
        }

        rebalancingMap.put(StatusType.monitor, null);
        rebalancingMap.put(StatusType.inactivate, null);
        rebalancingMap.put(StatusType.startup, new RebalanceTransitionCallback(data, topologyId, new StormStatus(rebalanceOldStatus)));
        rebalancingMap.put(StatusType.activate, null);
        rebalancingMap.put(StatusType.kill, null);
        rebalancingMap.put(StatusType.remove, null);
        rebalancingMap.put(StatusType.rebalance, new RebalanceTransitionCallback(data, topologyId, currentStatus));
        rebalancingMap.put(StatusType.do_rebalance, new DoRebalanceTransitionCallback(data, topologyId, new StormStatus(rebalanceOldStatus)));
        rebalancingMap.put(StatusType.done_rebalance, new DoneRebalanceTransitionCallback(data, topologyId));
        rebalancingMap.put(StatusType.update_topology, null);
        rtn.put(StatusType.rebalancing, rebalancingMap);

        Map<StatusType, Callback> upgradingMap = new HashMap<>();
        upgradingMap.put(StatusType.monitor, new RollbackTransitionCallback(data, topologyId));
        upgradingMap.put(StatusType.inactivate, null);
        upgradingMap.put(StatusType.startup, null);
        upgradingMap.put(StatusType.activate, null);
        upgradingMap.put(StatusType.kill, new KillTransitionCallback(data, topologyId));
        upgradingMap.put(StatusType.remove, new RemoveTransitionCallback(data, topologyId));
        upgradingMap.put(StatusType.rebalance, null);
        upgradingMap.put(StatusType.do_rebalance, null);
        upgradingMap.put(StatusType.done_rebalance, null);
        upgradingMap.put(StatusType.update_topology, null);
        upgradingMap.put(StatusType.upgrading, null);
        rtn.put(StatusType.upgrading, upgradingMap);

        return rtn;
    }
678326
public static boolean verifyTopology(TopologyAPI.Topology topology) {
    if (!topology.hasName() || topology.getName().isEmpty()) {
      LOG.severe("Missing topology name");
      return false;
    }
    if (topology.getName().contains(".") || topology.getName().contains("/")) {
      LOG.severe("Invalid topology name. Topology name shouldn't have . or /");
      return false;
    }

    // Only verify RAM map string well-formed.
    getComponentRamMapConfig(topology);
    // Verify all bolts input streams exist. First get all output streams.
    Set<String> outputStreams = new HashSet<>();
    for (TopologyAPI.Spout spout : topology.getSpoutsList()) {
      for (TopologyAPI.OutputStream stream : spout.getOutputsList()) {
        outputStreams.add(
            stream.getStream().getComponentName() + "/" + stream.getStream().getId());
      }
    }
    for (TopologyAPI.Bolt bolt : topology.getBoltsList()) {
      for (TopologyAPI.OutputStream stream : bolt.getOutputsList()) {
        outputStreams.add(
            stream.getStream().getComponentName() + "/" + stream.getStream().getId());
      }
    }
    // Match output streams with input streams.
    for (TopologyAPI.Bolt bolt : topology.getBoltsList()) {
      for (TopologyAPI.InputStream stream : bolt.getInputsList()) {
        String key = stream.getStream().getComponentName() + "/" + stream.getStream().getId();
        if (!outputStreams.contains(key)) {
          LOG.severe("Invalid input stream " + key + " existing streams are " + outputStreams);
          return false;
        }
      }
    }
    // TODO(nbhagat): Should we enforce all output stream must be consumed?

    return true;
  }
696083
protected boolean isHadoopCluster() {
        boolean isHadoopCluster = false;
        try {
            isHadoopCluster = Shell.getHadoopHome() != null;
        } catch (IOException e) {
            // ignore - false is default setting
        }
        return isHadoopCluster;
    }
964020
private Set<String> get_cleanup_ids(StormClusterState clusterState, List<String> activeTopologies) throws Exception {
        List<String> task_ids = clusterState.task_storms();
        List<String> heartbeat_ids = clusterState.heartbeat_storms();
        List<String> error_ids = clusterState.task_error_storms();
        List<String> assignment_ids = clusterState.assignments(null);
        List<String> metric_ids = clusterState.get_metrics();

        HashSet<String> latest_code_ids = new HashSet<>();

        Set<String> code_ids = BlobStoreUtils.code_ids(nimbusData.getBlobStore());
        Set<String> to_cleanup_ids = new HashSet<>();
        Set<String> pendingTopologies = nimbusData.getPendingSubmitTopologies().buildMap().keySet();

        if (task_ids != null) {
            to_cleanup_ids.addAll(task_ids);
        }

        if (heartbeat_ids != null) {
            to_cleanup_ids.addAll(heartbeat_ids);
        }

        if (error_ids != null) {
            to_cleanup_ids.addAll(error_ids);
        }

        if (assignment_ids != null) {
            to_cleanup_ids.addAll(assignment_ids);
        }

        if (code_ids != null) {
            to_cleanup_ids.addAll(code_ids);
        }

        if (metric_ids != null) {
            to_cleanup_ids.addAll(metric_ids);
        }

        if (activeTopologies != null) {
            to_cleanup_ids.removeAll(activeTopologies);
            latest_code_ids.removeAll(activeTopologies);
        }

        to_cleanup_ids.removeAll(pendingTopologies);

        /**
         * Why need to remove latest code. Due to competition between Thrift.threads and TopologyAssign thread
         */
        to_cleanup_ids.removeAll(latest_code_ids);
        LOG.info("Skip removing topology of " + latest_code_ids);

        return to_cleanup_ids;
    }
881712
public static void main(String[] args) {

        List<String> tmpArgs = new ArrayList<String>(Arrays.asList(args));

        int numWorker = StormSamoaUtils.numWorkers(tmpArgs);

        args = tmpArgs.toArray(new String[0]);

        //convert the arguments into Storm topology
        StormTopology stormTopo = StormSamoaUtils.argsToTopology(args);
        String topologyName = stormTopo.getTopologyName();

        Config conf = new Config();
        //conf.putAll(Utils.readStormConfig());
        conf.setDebug(false);

        //local mode
        conf.setMaxTaskParallelism(numWorker);

        backtype.storm.LocalCluster cluster = new backtype.storm.LocalCluster();
        cluster.submitTopology(topologyName, conf, stormTopo.getStormBuilder().createTopology());

        backtype.storm.utils.Utils.sleep(600 * 1000);

        cluster.killTopology(topologyName);
        cluster.shutdown();

    }
1166978
protected void validateExecutionState(
      String topologyName,
      ExecutionEnvironment.ExecutionState executionState)
      throws TopologyRuntimeManagementException {
    String stateCluster = executionState.getCluster();
    String stateRole = executionState.getRole();
    String stateEnv = executionState.getEnviron();
    String configCluster = Context.cluster(config);
    String configRole = Context.role(config);
    String configEnv = Context.environ(config);

    if (!stateCluster.equals(configCluster)
        || !stateRole.equals(configRole)
        || !stateEnv.equals(configEnv)) {
      String currentState = String.format("%s/%s/%s", stateCluster, stateRole, stateEnv);
      String configState = String.format("%s/%s/%s", configCluster, configRole, configEnv);
      throw new TopologyRuntimeManagementException(String.format(
          "cluster/role/environ does not match. Topology '%s' is running at %s, not %s",
          topologyName, currentState, configState));
    }
  }
911388
public boolean markInactive() {
		if (TaskSlotState.ACTIVE == state || TaskSlotState.ALLOCATED == state) {
			state = TaskSlotState.ALLOCATED;

			return true;
		} else {
			return false;
		}
	}
1201181
public ExecutionEnvironment.ExecutionState getExecutionState(String topologyName) {
    return awaitResult(delegate.getExecutionState(null, topologyName));
  }
1182289
public boolean isHeadMember() {
		final Address localAddress = getLocalAddress();
		if (localAddress != null) {
			final List<Address> clusterMembers = getClusterMembers();
			return !clusterMembers.isEmpty() && clusterMembers.get(0).equals(localAddress);
		}
		else {
			return true;
		}
	}
337482
static boolean checkGridSize(List<List<NodeInfo>> grid ,
								 int clusterSize ) {
		int total = 0;
		int expected = grid.get(0).size();
		for (int i = 0; i < grid.size(); i++) {
			if( expected != grid.get(i).size() )
				return false;
			total += grid.get(i).size();
		}

		return total == clusterSize;
	}
1122426
@Override
    public void rebalance(String topologyName, RebalanceOptions options) throws TException {
        try {
            checkTopologyActive(data, topologyName, true);
            Integer wait_amt = null;
            String jsonConf = null;
            Boolean reassign = false;
            if (options != null) {
                if (options.is_set_wait_secs())
                    wait_amt = options.get_wait_secs();
                if (options.is_set_reassign())
                    reassign = options.is_reassign();
                if (options.is_set_conf())
                    jsonConf = options.get_conf();
            }

            LOG.info("Begin to rebalance " + topologyName + "wait_time:" + wait_amt + ", reassign: " + reassign +
                    ", new worker/bolt configuration:" + jsonConf);

            Map<Object, Object> conf = (Map<Object, Object>) JStormUtils.from_json(jsonConf);

            NimbusUtils.transitionName(data, topologyName, true, StatusType.rebalance, wait_amt, reassign, conf);

            notifyTopologyActionListener(topologyName, "rebalance");
        } catch (NotAliveException e) {
            String errMsg = "Rebalance error, topology " + topologyName + " is not alive!";
            LOG.error(errMsg, e);
            throw new NotAliveException(errMsg);
        } catch (Exception e) {
            String errMsg = "Failed to rebalance topology " + topologyName;
            LOG.error(errMsg, e);
            throw new TException(errMsg);
        }

    }
976572
public boolean wasExecuted(@NonNull String nodeName) {
        ensureNodeStateExists(nodeName);

        return states.get(nodeName).isExecuted();
    }
1159892
public boolean isLeader() {
    MemberId leader = this.leader;
    return leader != null && leader.equals(cluster.getMember().memberId());
  }
408906
private boolean preCheckClusterState(ClusterService clusterService) {
        ClusterState initialState = clusterService.getClusterState();
        if (!initialState.isJoinAllowed()) {
            logger.warning("Could not prepare cluster state since it has been changed to " + initialState);
            return false;
        }

        return true;
    }
936889
function(topology) {
  const maxWireVersion = topology.lastIsMaster().maxWireVersion;
  if (maxWireVersion < RETRYABLE_WIRE_VERSION) {
    return false;
  }

  if (!topology.logicalSessionTimeoutMinutes) {
    return false;
  }

  if (topologyType(topology) === TopologyType.Single) {
    return false;
  }

  return true;
}