/*
1. Read arguments
2. Construct topology/task
3. Upload the JAR to HDFS if we are running on YARN
4. Submit topology to SamzaEngine
495177
public static void main(String[] args) {
		// Read arguments
		List<String> tmpArgs = new ArrayList<String>(Arrays.asList(args));
		parseArguments(tmpArgs);
		
		args = tmpArgs.toArray(new String[0]);
		
		// Init Task
		StringBuilder cliString = new StringBuilder();
        for (int i = 0; i < args.length; i++) {
            cliString.append(" ").append(args[i]);
        }
        logger.debug("Command line string = {}", cliString.toString());
        System.out.println("Command line string = " + cliString.toString());
        
		Task task = null;
        try {
            task = (Task) ClassOption.cliStringToObject(cliString.toString(), Task.class, null);
            logger.info("Sucessfully instantiating {}", task.getClass().getCanonicalName());
        } catch (Exception e) {
            logger.error("Fail to initialize the task", e);
            System.out.println("Fail to initialize the task" + e);
            return;
        }
		task.setFactory(new SamzaComponentFactory());
		task.init();
		
		// Upload JAR file to HDFS
		String hdfsPath = null;
		if (!isLocal) {
			Path path = FileSystems.getDefault().getPath(jarPackagePath);
			hdfsPath = uploadJarToHDFS(path.toFile());
			if(hdfsPath == null) {
				System.out.println("Fail uploading JAR file \""+path.toAbsolutePath().toString()+"\" to HDFS.");
				return;
			}
		}
		
		// Set parameters
		SamzaEngine.getEngine()
		.setLocalMode(isLocal)
		.setZooKeeper(zookeeper)
		.setKafka(kafka)
		.setYarnPackage(hdfsPath)
		.setKafkaReplicationFactor(kafkaReplicationFactor)
		.setConfigHome(yarnConfHome)
		.setAMMemory(amMem)
		.setContainerMemory(containerMem)
		.setPiPerContainerRatio(piPerContainer)
		.setKryoRegisterFile(kryoRegisterFile)
		.setCheckpointFrequency(checkpointFrequency);
		
		// Submit topology
		SamzaEngine.submitTopology((SamzaTopology)task.getTopology());
		
	}
607713
static YarnClusterSubmissionFromCS fromJobSubmissionParametersFile(final File yarnClusterAppSubmissionParametersFile,
                                                                     final File yarnClusterJobSubmissionParametersFile)
      throws IOException {
    try (final FileInputStream appFileInputStream = new FileInputStream(yarnClusterAppSubmissionParametersFile)) {
      try (final FileInputStream jobFileInputStream = new FileInputStream(yarnClusterJobSubmissionParametersFile)) {
        // this is mainly a test hook
        return readYarnClusterSubmissionFromCSFromInputStream(appFileInputStream, jobFileInputStream);
      }
    }
  }
192779
@Override
  public final 
  void setupTask(org.apache.hadoop.mapreduce.TaskAttemptContext taskContext
                 ) throws IOException {
    setupTask((TaskAttemptContext) taskContext);
  }
722771
@Override
    public void run() {
        if (!context.getTopologyMetricContexts().containsKey(topologyId)) {
            LOG.warn("topology {} has been killed or has not started, skip update.", topologyId);
            return;
        }

        // double check and reset stream metrics if disabled
        if (!JStormMetrics.enableStreamMetrics) {
            topologyMetrics.set_streamMetric(new MetricInfo());
        }

        if (!JStormMetrics.CLUSTER_METRIC_KEY.equals(topologyId)) {
            updateClusterMetrics(topologyId, topologyMetrics);
        }

        // overwrite nimbus-local metrics data
        context.getMetricCache().putMetricData(topologyId, topologyMetrics);

        // below process is kind of a transaction, first we lock an empty slot, mark it as PRE_SET
        // by this time the slot is not yet ready for uploading as the upload thread looks for SET slots only
        // after all metrics data has been saved, we mark it as SET, then it's ready for uploading.
        int idx = context.getAndPresetFirstEmptyIndex();
        if (idx < 0) {
            LOG.error("Exceeding maxPendingUploadMetrics(too much metrics in local rocksdb), " +
                    "skip caching metrics data for topology:{}", topologyId);
            return;
        }

        try {
            TopologyMetricDataInfo summary = new TopologyMetricDataInfo();
            int total = 0;
            summary.topologyId = topologyId;
            summary.timestamp = timestamp;
            if (topologyId.equals(JStormMetrics.NIMBUS_METRIC_KEY) || topologyId.equals(JStormMetrics.CLUSTER_METRIC_KEY)) {
                summary.type = MetricUploader.METRIC_TYPE_TOPLOGY;
            } else {
                total += topologyMetrics.get_topologyMetric().get_metrics_size()
                        + topologyMetrics.get_componentMetric().get_metrics_size();

                int compStream = 0;
                if (topologyMetrics.is_set_compStreamMetric()) {
                    compStream = topologyMetrics.get_compStreamMetric().get_metrics_size();
                    total += compStream;
                }
                if (total > 0) {
                    int sub = topologyMetrics.get_taskMetric().get_metrics_size()
                            + topologyMetrics.get_workerMetric().get_metrics_size()
                            + topologyMetrics.get_nettyMetric().get_metrics_size()
                            + topologyMetrics.get_streamMetric().get_metrics_size();
                    if (sub > 0) {
                        total += sub;
                        summary.type = MetricUploader.METRIC_TYPE_ALL;
                    } else {
                        summary.type = MetricUploader.METRIC_TYPE_TOPLOGY;
                    }

                    LOG.debug("tp:{}, comp:{}, comp_stream:{}, task:{}, worker:{}, netty:{}, stream:{}",
                            topologyMetrics.get_topologyMetric().get_metrics_size(),
                            topologyMetrics.get_componentMetric().get_metrics_size(),
                            compStream,
                            topologyMetrics.get_taskMetric().get_metrics_size(),
                            topologyMetrics.get_workerMetric().get_metrics_size(),
                            topologyMetrics.get_nettyMetric().get_metrics_size(),
                            topologyMetrics.get_streamMetric().get_metrics_size());
                } else {
                    summary.type = MetricUploader.METRIC_TYPE_TASK;
                    total += topologyMetrics.get_taskMetric().get_metrics_size();
                }
            }

            context.getMetricCache().put(ClusterMetricsContext.PENDING_UPLOAD_METRIC_DATA_INFO + idx, summary);
            context.getMetricCache().put(ClusterMetricsContext.PENDING_UPLOAD_METRIC_DATA + idx, topologyMetrics);
            context.markSet(idx);
            LOG.debug("Put metric data to local cache, topology:{}, idx:{}, total:{}", topologyId, idx, total);
        } catch (Exception ex) {
            LOG.error("Error", ex);
            context.forceMarkUploaded(idx);
        }
    }
126043
public static void main(String[] args) throws Exception {
    Builder processingGraphBuilder = Builder.newBuilder();

    Streamlet<Integer> randomIntegers = processingGraphBuilder
        .newSource(() -> {
          // Random integers are emitted every 50 milliseconds
          StreamletUtils.sleep(50);
          return ThreadLocalRandom.current().nextInt(100);
        })
        .setNumPartitions(2)
        .setName("random-integer-source");

    randomIntegers
        // The specific repartition logic is applied here
        .repartition(8, RepartitionTopology::repartitionStreamlet)
        .setName("repartition-incoming-values")
        // Here, a generic repartition logic is applied (simply
        // changing the number of partitions without specifying
        // how repartitioning will take place)
        .repartition(2)
        .setName("reduce-partitions-for-logging-operation")
        .log();

    // Fetches the topology name from the first command-line argument
    String topologyName = StreamletUtils.getTopologyName(args);

    Config config = Config.defaultConfig();

    // Finally, the processing graph and configuration are passed to the Runner, which converts
    // the graph into a Heron topology that can be run in a Heron cluster.
    new Runner().run(topologyName, config, processingGraphBuilder);
  }
1239323
public static <K, V> org.apache.flink.api.java.hadoop.mapreduce.HadoopInputFormat<K, V> readHadoopFile(
			org.apache.hadoop.mapreduce.lib.input.FileInputFormat<K, V> mapreduceInputFormat, Class<K> key, Class<V> value, String inputPath, Job job) throws IOException {
		// set input path in Job
		org.apache.hadoop.mapreduce.lib.input.FileInputFormat.addInputPath(job, new org.apache.hadoop.fs.Path(inputPath));
		// return wrapping InputFormat
		return createHadoopInput(mapreduceInputFormat, key, value, job);
	}
1170946
private void createHdfsFilesystem(Configuration conf) throws Exception {
    // Inits HDFS file system object
    mFileSystem = FileSystem.get(URI.create(conf.get("fs.defaultFS")), conf);
    mOutputFilePath = new Path("./MapReduceOutputFile");
    if (mFileSystem.exists(mOutputFilePath)) {
      mFileSystem.delete(mOutputFilePath, true);
    }
  }
836120
@Override
    public void run() {
        StormClusterState clusterState = data.getStormClusterState();

        try {
            // Note: need first check Assignments
            List<String> activeTopologies = clusterState.assignments(null);

            if (activeTopologies == null) {
                LOG.info("Failed to get active topologies");
                return;
            }

            for (String topologyId : activeTopologies) {
                if (clusterState.storm_base(topologyId, null) == null) {
                    continue;
                }

                LOG.debug("Check tasks of topology " + topologyId);

                // Note that we don't check /ZK-dir/taskbeats/topologyId to get task ids
                Set<Integer> taskIds = clusterState.task_ids(topologyId);
                if (taskIds == null) {
                    LOG.info("Failed to get task ids of " + topologyId);
                    continue;
                }
                Assignment assignment = clusterState.assignment_info(topologyId, null);

                Set<Integer> deadTasks = new HashSet<>();
                boolean needReassign = false;
                for (Integer task : taskIds) {
                    boolean isTaskDead = NimbusUtils.isTaskDead(data, topologyId, task);
                    if (isTaskDead) {
                        deadTasks.add(task);
                        needReassign = true;
                    }
                }

                TopologyTaskHbInfo topologyHbInfo = data.getTasksHeartbeat().get(topologyId);
                if (needReassign) {
                    if (topologyHbInfo != null) {
                        int topologyMasterId = topologyHbInfo.get_topologyMasterId();
                        if (deadTasks.contains(topologyMasterId)) {
                            deadTasks.clear();
                            if (assignment != null) {
                                ResourceWorkerSlot resource = assignment.getWorkerByTaskId(topologyMasterId);
                                if (resource != null)
                                    deadTasks.addAll(resource.getTasks());
                                else
                                    deadTasks.add(topologyMasterId);
                            }
                        } else {
                            Map<Integer, TaskHeartbeat> taskHbs = topologyHbInfo.get_taskHbs();
                            int launchTime = JStormUtils.parseInt(data.getConf().get(Config.NIMBUS_TASK_LAUNCH_SECS));
                            if (taskHbs == null || taskHbs.get(topologyMasterId) == null ||
                                    taskHbs.get(topologyMasterId).get_uptime() < launchTime) {
                                /*try {
                                    clusterState.topology_heartbeat(topologyId, topologyHbInfo);
                                } catch (Exception e) {
                                    LOG.error("Failed to update task heartbeat info to ZK for " + topologyId, e);
                                }*/
                                return;
                            }
                        }
                        Map<Integer, ResourceWorkerSlot> deadTaskWorkers = new HashMap<>();
                        for (Integer task : deadTasks) {
                            LOG.info("Found " + topologyId + ", taskId:" + task + " is dead");

                            ResourceWorkerSlot resource = null;
                            if (assignment != null)
                                resource = assignment.getWorkerByTaskId(task);
                            if (resource != null) {
                                deadTaskWorkers.put(task, resource);
                            }
                        }
                        Map<ResourceWorkerSlot, List<Integer>> workersDeadTasks = JStormUtils.reverse_map(deadTaskWorkers);
                        for (Map.Entry<ResourceWorkerSlot, List<Integer>> entry : workersDeadTasks.entrySet()) {
                            ResourceWorkerSlot resource = entry.getKey();
                            //we only report one task
                            for (Integer task : entry.getValue()) {
                                Date now = new Date();
                                String nowStr = TimeFormat.getSecond(now);
                                String errorInfo = "Task-" + entry.getValue().toString() + " is dead on " +
                                        resource.getHostname() + ":" + resource.getPort() + ", " + nowStr;
                                LOG.info(errorInfo);
                                clusterState.report_task_error(topologyId, task, errorInfo, ErrorConstants.ERROR,
                                        ErrorConstants.CODE_TASK_DEAD, ErrorConstants.DURATION_SECS_TASK_DEAD);
                                break;
                            }
                        }
                        if (deadTaskWorkers.size() > 0) {
                            // notify jstorm monitor
                            TaskDeadEvent.pushEvent(topologyId, deadTaskWorkers);
                        }
                    }
                    NimbusUtils.transition(data, topologyId, false, StatusType.monitor);
                }

                if (topologyHbInfo != null) {
                    try {
                        clusterState.topology_heartbeat(topologyId, topologyHbInfo);
                    } catch (Exception e) {
                        LOG.error("Failed to update task heartbeat info to ZK for " + topologyId, e);
                    }
                }
            }

        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
        }
    }
473471
public void attachJobGraph(List<JobVertex> topologiallySorted) throws JobException {

		assertRunningInJobMasterMainThread();

		LOG.debug("Attaching {} topologically sorted vertices to existing job graph with {} " +
				"vertices and {} intermediate results.",
			topologiallySorted.size(),
			tasks.size(),
			intermediateResults.size());

		final ArrayList<ExecutionJobVertex> newExecJobVertices = new ArrayList<>(topologiallySorted.size());
		final long createTimestamp = System.currentTimeMillis();

		for (JobVertex jobVertex : topologiallySorted) {

			if (jobVertex.isInputVertex() && !jobVertex.isStoppable()) {
				this.isStoppable = false;
			}

			// create the execution job vertex and attach it to the graph
			ExecutionJobVertex ejv = new ExecutionJobVertex(
				this,
				jobVertex,
				1,
				rpcTimeout,
				globalModVersion,
				createTimestamp);

			ejv.connectToPredecessors(this.intermediateResults);

			ExecutionJobVertex previousTask = this.tasks.putIfAbsent(jobVertex.getID(), ejv);
			if (previousTask != null) {
				throw new JobException(String.format("Encountered two job vertices with ID %s : previous=[%s] / new=[%s]",
					jobVertex.getID(), ejv, previousTask));
			}

			for (IntermediateResult res : ejv.getProducedDataSets()) {
				IntermediateResult previousDataSet = this.intermediateResults.putIfAbsent(res.getId(), res);
				if (previousDataSet != null) {
					throw new JobException(String.format("Encountered two intermediate data set with ID %s : previous=[%s] / new=[%s]",
						res.getId(), res, previousDataSet));
				}
			}

			this.verticesInCreationOrder.add(ejv);
			this.numVerticesTotal += ejv.getParallelism();
			newExecJobVertices.add(ejv);
		}

		terminationFuture = new CompletableFuture<>();
		failoverStrategy.notifyNewVertices(newExecJobVertices);
	}
72568
public void submitTopology(boolean isLocal)
    {
        if (logger.isInfoEnabled())
        {
            logger.info("isLocal=" + isLocal);
        }

        if (isLocal)
        {
            // ローカル環境で実行
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology(getTopologyName(), getConfig(), getBuilder().createTopology());
        }
        else
        {
            // 分散環境で実行
            try
            {
                StormSubmitter.submitTopology(getTopologyName(), getConfig(),
                        getBuilder().createTopology());
            }
            catch (AlreadyAliveException ex)
            {
                String logFormat = "Occur exception at Topology Submit. Skip Topology Submit. : TopologyName={0}";
                String logMessage = MessageFormat.format(logFormat, getTopologyName());
                logger.error(logMessage, ex);
            }
            catch (InvalidTopologyException ex)
            {
                String logFormat = "Occur exception at Topology Submit. Skip Topology Submit. : TopologyName={0}";
                String logMessage = MessageFormat.format(logFormat, getTopologyName());
                logger.error(logMessage, ex);
            }
        }
    }
285442
public JSONObject topologyToJSON() throws JSONException {
        JSONStringer stringer = new JSONStringer();
        stringer.object();

        stringer.keySymbolValuePair(TOPO_VERSION, version);

        stringer.key(TOPO_PARTITIONS).array();
        for (Partition partition : partitionsById.values()) {
            partition.toJSON(stringer);
        }
        stringer.endArray();

        stringer.key(TOPO_HOSTS).array();
        for (Host host : hostsById.values()) {
            host.toJSON(stringer);
        }
        stringer.endArray();
        stringer.keySymbolValuePair(TOPO_SITES_PER_HOST, getSitesPerHost());
        stringer.keySymbolValuePair(TOPO_REPLICATION_FACTOR, getReplicationFactor());
        stringer.keySymbolValuePair(TOPO_UNBALANCED_PART_COUNT, m_unbalancedPartitionCount);

        stringer.endObject();

        return new JSONObject(stringer.toString());
    }
1178538
private synchronized void loadTopologyConfiguration()
            throws ConfigurationException
    {
        logger.debug("Loading topology configuration from {}",
                topologyConfigFilename);

        final TypeDescription topologyConfigTypeDescription = new TypeDescription(
                TopologyConfig.class);
        topologyConfigTypeDescription.putListPropertyType("topology",
                Datacenter.class);

        final TypeDescription topologyTypeDescription = new TypeDescription(
                Datacenter.class);
        topologyTypeDescription.putListPropertyType("racks", Rack.class);

        final TypeDescription rackTypeDescription = new TypeDescription(
                Rack.class);
        rackTypeDescription.putListPropertyType("nodes", Node.class);

        final Constructor configConstructor = new Constructor(
                TopologyConfig.class);
        configConstructor.addTypeDescription(topologyConfigTypeDescription);
        configConstructor.addTypeDescription(topologyTypeDescription);
        configConstructor.addTypeDescription(rackTypeDescription);

        final InputStream configFileInputStream = getClass().getClassLoader()
                .getResourceAsStream(topologyConfigFilename);
        if (configFileInputStream == null)
        {
            throw new ConfigurationException(
                    "Could not read topology config file "
                            + topologyConfigFilename);
        }
        Yaml yaml;
        TopologyConfig topologyConfig;
        try
        {
            yaml = new Yaml(configConstructor);
            topologyConfig = (TopologyConfig) yaml.load(configFileInputStream);
        }
        finally
        {
            FileUtils.closeQuietly(configFileInputStream);
        }
        final Map<InetAddress, NodeData> nodeDataMap = new HashMap<InetAddress, NodeData>();

        if (topologyConfig.topology == null)
        {
            throw new ConfigurationException(
                    "Topology configuration file is missing the topology section");
        }

        for (final Datacenter datacenter : topologyConfig.topology)
        {
            if (datacenter.dc_name == null)
            {
                throw new ConfigurationException(
                        "Topology configuration file is missing a data center name for some data center");
            }

            for (final Rack rack : datacenter.racks)
            {
                if (rack.rack_name == null)
                {
                    throw new ConfigurationException(
                            String.format(
                                    "Topology configuration file is missing a rack name for some rack under data center '%s'",
                                    datacenter.dc_name));
                }

                for (final Node node : rack.nodes)
                {
                    if (node.broadcast_address == null)
                    {
                        throw new ConfigurationException(
                                String.format(
                                        "Topology configuration file is missing a broadcast address for some node under data center '%s' rack '%s'",
                                        datacenter.dc_name, rack.rack_name));
                    }

                    final InetAddress endpoint = InetAddresses
                            .forString(node.broadcast_address);
                    final InetAddress dcLocalAddress = node.dc_local_address == null ? null
                            : InetAddresses.forString(node.dc_local_address);

                    final NodeData nodeData = new NodeData();
                    nodeData.datacenter = datacenter.dc_name;
                    nodeData.rack = rack.rack_name;
                    nodeData.dcLocalAddress = dcLocalAddress;

                    if (nodeDataMap.put(endpoint, nodeData) != null)
                    {
                        throw new ConfigurationException(
                                String.format(
                                        "IP address '%s' appears more than once in the topology configuration file",
                                        endpoint));
                    }

                    if (dcLocalAddress != null
                            && nodeDataMap.put(dcLocalAddress, nodeData) != null)
                    {
                        throw new ConfigurationException(
                                String.format(
                                        "IP address '%s' appears more than once in the topology configuration file",
                                        dcLocalAddress));
                    }
                }
            }
        }

        final NodeData localNodeData = nodeDataMap.get(FBUtilities
                .getBroadcastAddress());
        if (localNodeData == null)
        {
            throw new ConfigurationException(
                    "Topology configuration missing information for the local node");
        }

        final NodeData defaultNodeData = new NodeData();

        if (topologyConfig.default_dc_name == null)
        {
            throw new ConfigurationException(
                    "default_dc_name must be specified");
        }
        if (topologyConfig.default_rack_name == null)
        {
            throw new ConfigurationException(
                    "default_rack_name must be specified");
        }

        defaultNodeData.datacenter = topologyConfig.default_dc_name;
        defaultNodeData.rack = topologyConfig.default_rack_name;

        // YAML configuration looks good; now make the changes

        this.nodeDataMap = nodeDataMap;
        this.localNodeData = localNodeData;
        this.defaultNodeData = defaultNodeData;
        maybeSetApplicationState();

        if (logger.isDebugEnabled())
        {
            logger.debug(
                    "Built topology map from config file: localNodeData={}, nodeDataMap={}",
                    localNodeData, nodeDataMap);
        }

        if (gossiperInitialized)
        {
            StorageService.instance.gossipSnitchInfo();
        }
    }
282358
public static void main(final String[] args) throws InjectionException, IOException, YarnException {
    final File jobSubmissionParametersFile = new File(args[0]);
    final File appSubmissionParametersFile = new File(args[1]);

    if (!(appSubmissionParametersFile.exists() && appSubmissionParametersFile.canRead())) {
      throw new IOException("Unable to open and read " + appSubmissionParametersFile.getAbsolutePath());
    }

    if (!(jobSubmissionParametersFile.exists() && jobSubmissionParametersFile.canRead())) {
      throw new IOException("Unable to open and read " + jobSubmissionParametersFile.getAbsolutePath());
    }

    final YarnClusterSubmissionFromCS yarnSubmission =
        YarnClusterSubmissionFromCS.fromJobSubmissionParametersFile(
            appSubmissionParametersFile, jobSubmissionParametersFile);

    LOG.log(Level.INFO, "YARN job submission received from C#: {0}", yarnSubmission);

    if (!yarnSubmission.getFileSystemUrl().equalsIgnoreCase(FileSystemUrl.DEFAULT_VALUE)) {
      LOG.log(Level.INFO, "getFileSystemUrl: {0}", yarnSubmission.getFileSystemUrl());
    } else {
      LOG.log(Level.INFO, "FileSystemUrl is not set, use default from the environment.");
    }

    final List<String> launchCommandPrefix = new ArrayList<String>() {{
        final REEFFileNames reefFileNames = TANG.newInjector().getInstance(REEFFileNames.class);
        add(reefFileNames.getDriverLauncherExeFile().getPath());
      }};

    final Configuration yarnJobSubmissionClientConfig = TANG.newConfigurationBuilder()
        .bindImplementation(RuntimeClasspathProvider.class, YarnClasspathProvider.class)
        .bindConstructor(org.apache.hadoop.yarn.conf.YarnConfiguration.class, YarnConfigurationConstructor.class)
        .bindNamedParameter(JobSubmissionDirectoryPrefix.class, yarnSubmission.getJobSubmissionDirectoryPrefix())
        .bindNamedParameter(FileSystemUrl.class, yarnSubmission.getFileSystemUrl())
        .bindList(DriverLaunchCommandPrefix.class, launchCommandPrefix)
        .build();

    final YarnJobSubmissionClient client =
        TANG.newInjector(yarnJobSubmissionClientConfig).getInstance(YarnJobSubmissionClient.class);

    client.launch(yarnSubmission);

    LOG.log(Level.INFO, "Returned from launch in Java YarnJobSubmissionClient");
    System.exit(0);
    LOG.log(Level.INFO, "End of main in Java YarnJobSubmissionClient");
  }
1035925
@Override
    public TopologyInfo getTopologyInfo(String topologyId) throws TException {
        long start = System.nanoTime();
        StormClusterState stormClusterState = data.getStormClusterState();

        try {
            // get topology's StormBase
            StormBase base = stormClusterState.storm_base(topologyId, null);
            if (base == null) {
                throw new NotAliveException("No topology of " + topologyId);
            }

            Assignment assignment = stormClusterState.assignment_info(topologyId, null);
            if (assignment == null) {
                throw new NotAliveException("No topology of " + topologyId);
            }

            TopologyTaskHbInfo topologyTaskHbInfo = data.getTasksHeartbeat().get(topologyId);
            Map<Integer, TaskHeartbeat> taskHbMap = null;
            if (topologyTaskHbInfo != null)
                taskHbMap = topologyTaskHbInfo.get_taskHbs();

            Map<Integer, TaskInfo> taskInfoMap = Cluster.get_all_taskInfo(stormClusterState, topologyId);
            Map<Integer, String> taskToComponent = Cluster.get_all_task_component(stormClusterState, topologyId, taskInfoMap);
            Map<Integer, String> taskToType = Cluster.get_all_task_type(stormClusterState, topologyId, taskInfoMap);


            String errorString;
            if (Cluster.is_topology_exist_error(stormClusterState, topologyId)) {
                errorString = "Y";
            } else {
                errorString = "";
            }

            TopologySummary topologySummary = new TopologySummary();
            topologySummary.set_id(topologyId);
            topologySummary.set_name(base.getStormName());
            topologySummary.set_uptimeSecs(TimeUtils.time_delta(base.getLanchTimeSecs()));
            topologySummary.set_status(base.getStatusString());
            topologySummary.set_numTasks(NimbusUtils.getTopologyTaskNum(assignment));
            topologySummary.set_numWorkers(assignment.getWorkers().size());
            topologySummary.set_errorInfo(errorString);

            Map<String, ComponentSummary> componentSummaryMap = new HashMap<>();
            HashMap<String, List<Integer>> componentToTasks = JStormUtils.reverse_map(taskToComponent);
            for (Entry<String, List<Integer>> entry : componentToTasks.entrySet()) {
                String name = entry.getKey();
                List<Integer> taskIds = entry.getValue();
                if (taskIds == null || taskIds.size() == 0) {
                    LOG.warn("No task of component " + name);
                    continue;
                }

                ComponentSummary componentSummary = new ComponentSummary();
                componentSummaryMap.put(name, componentSummary);

                componentSummary.set_name(name);
                componentSummary.set_type(taskToType.get(taskIds.get(0)));
                componentSummary.set_parallel(taskIds.size());
                componentSummary.set_taskIds(taskIds);
            }

            Map<Integer, TaskSummary> taskSummaryMap = new TreeMap<>();
            Map<Integer, List<TaskError>> taskErrors = Cluster.get_all_task_errors(stormClusterState, topologyId);

            for (Integer taskId : taskInfoMap.keySet()) {
                TaskSummary taskSummary = new TaskSummary();
                taskSummaryMap.put(taskId, taskSummary);

                taskSummary.set_taskId(taskId);
                if (taskHbMap == null) {
                    taskSummary.set_status("Starting");
                    taskSummary.set_uptime(0);
                } else {
                    TaskHeartbeat hb = taskHbMap.get(taskId);
                    if (hb == null) {
                        taskSummary.set_status("Starting");
                        taskSummary.set_uptime(0);
                    } else {
                        boolean isInactive = NimbusUtils.isTaskDead(data, topologyId, taskId);
                        if (isInactive)
                            taskSummary.set_status("INACTIVE");
                        else
                            taskSummary.set_status("ACTIVE");
                        taskSummary.set_uptime(hb.get_uptime());
                    }
                }

                if (StringUtils.isBlank(errorString)) {
                    continue;
                }

                List<TaskError> taskErrorList = taskErrors.get(taskId);
                if (taskErrorList != null && taskErrorList.size() != 0) {
                    for (TaskError taskError : taskErrorList) {
                        ErrorInfo errorInfo = new ErrorInfo(taskError.getError(), taskError.getTimSecs(),
                                taskError.getLevel(), taskError.getCode());
                        taskSummary.add_to_errors(errorInfo);
                        String component = taskToComponent.get(taskId);
                        componentSummaryMap.get(component).add_to_errors(errorInfo);
                    }
                }
            }

            for (ResourceWorkerSlot workerSlot : assignment.getWorkers()) {
                String hostname = workerSlot.getHostname();
                int port = workerSlot.getPort();

                for (Integer taskId : workerSlot.getTasks()) {
                    TaskSummary taskSummary = taskSummaryMap.get(taskId);
                    taskSummary.set_host(hostname);
                    taskSummary.set_port(port);
                }
            }

            TopologyInfo topologyInfo = new TopologyInfo();
            topologyInfo.set_topology(topologySummary);
            topologyInfo.set_components(JStormUtils.mk_list(componentSummaryMap.values()));
            topologyInfo.set_tasks(JStormUtils.mk_list(taskSummaryMap.values()));

            // return topology metric & component metric only
            List<MetricInfo> tpMetricList = data.getMetricCache().getMetricData(topologyId, MetaType.TOPOLOGY);
            List<MetricInfo> compMetricList = data.getMetricCache().getMetricData(topologyId, MetaType.COMPONENT);
            List<MetricInfo> workerMetricList = data.getMetricCache().getMetricData(topologyId, MetaType.WORKER);
            List<MetricInfo> compStreamMetricList = data.getMetricCache().getMetricData(topologyId, MetaType.COMPONENT_STREAM);

            MetricInfo taskMetric = MetricUtils.mkMetricInfo();
            MetricInfo streamMetric = MetricUtils.mkMetricInfo();
            MetricInfo nettyMetric = MetricUtils.mkMetricInfo();
            MetricInfo tpMetric, compMetric, compStreamMetric, workerMetric;

            if (tpMetricList == null || tpMetricList.size() == 0) {
                tpMetric = MetricUtils.mkMetricInfo();
            } else {
                // get the last min topology metric
                tpMetric = tpMetricList.get(tpMetricList.size() - 1);
            }
            if (compMetricList == null || compMetricList.size() == 0) {
                compMetric = MetricUtils.mkMetricInfo();
            } else {
                compMetric = compMetricList.get(0);
            }
            if (compStreamMetricList == null || compStreamMetricList.size() == 0) {
                compStreamMetric = MetricUtils.mkMetricInfo();
            } else {
                compStreamMetric = compStreamMetricList.get(0);
            }
            if (workerMetricList == null || workerMetricList.size() == 0) {
                workerMetric = MetricUtils.mkMetricInfo();
            } else {
                workerMetric = workerMetricList.get(0);
            }
            TopologyMetric topologyMetrics = new TopologyMetric(tpMetric, compMetric, workerMetric,
                    taskMetric, streamMetric, nettyMetric);
            topologyMetrics.set_compStreamMetric(compStreamMetric);
            topologyInfo.set_metrics(topologyMetrics);

            return topologyInfo;
        } catch (TException e) {
            LOG.info("Failed to get topologyInfo " + topologyId, e);
            throw e;
        } catch (Exception e) {
            LOG.info("Failed to get topologyInfo " + topologyId, e);
            throw new TException("Failed to get topologyInfo" + topologyId);
        } finally {
            long end = System.nanoTime();
            SimpleJStormMetric.updateNimbusHistogram("getTopologyInfo", (end - start) / TimeUtils.NS_PER_US);
        }

    }
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;
  }
298643
public void loadTopology() {
        if (fake) {
            getView().updateHosts(deriveGroups(generateFakeDomain()), hostIndex);
        } else {
            Outcome<FunctionContext> outcome = new Outcome<FunctionContext>() {
                @Override
                public void onFailure(final FunctionContext context) {
                    Console.error("Unable to load topology", context.getErrorMessage()); // TODO i18n
                }

                @Override
                public void onSuccess(final FunctionContext context) {
                    List<HostInfo> hosts = context.pop();
                    getView().updateHosts(deriveGroups(hosts), hostIndex);
                }
            };
            new Async<FunctionContext>(Footer.PROGRESS_ELEMENT).waterfall(new FunctionContext(), outcome,
                    new TopologyFunctions.ReadHostsAndGroups(dispatcher),
                    new TopologyFunctions.ReadServerConfigs(dispatcher, beanFactory),
                    new TopologyFunctions.FindRunningServerInstances(dispatcher));
        }
    }
339303
public static void downloadDistributeStormCode(Map conf, String topologyId, String masterCodeDir)
            throws IOException, TException {
        String tmpToot = null;
        try {
            // STORM_LOCAL_DIR/supervisor/tmp/(UUID)
            tmpToot = StormConfig.supervisorTmpDir(conf) + File.separator + UUID.randomUUID().toString();

            // STORM_LOCAL_DIR/supervisor/stormdist/topologyId
            String stormRoot = StormConfig.supervisor_stormdist_root(conf, topologyId);

//        JStormServerUtils.downloadCodeFromMaster(conf, tmproot, masterCodeDir, topologyId, true);
            JStormServerUtils.downloadCodeFromBlobStore(conf, tmpToot, topologyId);

            // tmproot/stormjar.jar
            String localFileJarTmp = StormConfig.stormjar_path(tmpToot);

            // extract dir from jar
            JStormUtils.extractDirFromJar(localFileJarTmp, StormConfig.RESOURCES_SUBDIR, tmpToot);

            File srcDir = new File(tmpToot);
            File destDir = new File(stormRoot);
            try {
                FileUtils.moveDirectory(srcDir, destDir);
            } catch (FileExistsException e) {
                FileUtils.copyDirectory(srcDir, destDir);
                FileUtils.deleteQuietly(srcDir);
            }
        } finally {
            if (tmpToot != null) {
                File srcDir = new File(tmpToot);
                FileUtils.deleteQuietly(srcDir);
            }
        }
    }