All Heron topologies require a main function that defines the topology's behavior
at runtime
496514
public static void main(String[] args) throws Exception {
    Builder builder = Builder.newBuilder();

    /**
     * The processing graph consists of a supplier streamlet that emits
     * random integers between 1 and 100. From there, a series of transformers
     * is applied. At the end of the graph, the original value is ultimately
     * unchanged.
     */
    builder.newSource(() -> ThreadLocalRandom.current().nextInt(100))
        .transform(new DoNothingTransformer<>())
        .transform(new IncrementTransformer(10))
        .transform(new IncrementTransformer(-7))
        .transform(new DoNothingTransformer<>())
        .transform(new IncrementTransformer(-3))
        .log();

    Config config = Config.defaultConfig();

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

    // 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, builder);
  }
1045851
@SuppressWarnings("rawtypes")
    public void initialize(Map conf, TopologyContext context, String function)
    {
        this.context = context;
        this.stormConf = conf;
        this.function = function;
        this.idsMap = new RotatingMap<TransactionAttempt, DrpcRequestInfo>(ROTATE_SIZE);
        this.rotateTime = TimeUnit.SECONDS.toMillis(((Number) conf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS)).intValue());
        this.lastRotate = getCurrentTime();
    }
673022
def submit(args=None):
    
    streamsx._streams._version._mismatch_check('streamsx.topology.context')
    cmd_args = _parse_args(args)
    if cmd_args.topology is not None:
        app = _get_topology_app(cmd_args)
    elif cmd_args.main_composite is not None:
        app = _get_spl_app(cmd_args)
    elif cmd_args.bundle is not None:
        app = _get_bundle(cmd_args)
    _job_config_args(cmd_args, app)
    sr = _submit(cmd_args, app)
    if 'return_code' not in sr:
        sr['return_code'] = 1;
    print(sr)
    return sr
110854
@VisibleForTesting
  Configuration getHMDriverConf() {
    String topologyPackageName = new File(topologyPackageLocation).getName();
    String corePackageName = new File(coreReleasePackage).getName();

    // topologyName and other configurations are required by Heron Driver/Scheduler to load
    // configuration files. Using REEF configuration model is better than depending on external
    // persistence.
    // TODO: https://github.com/apache/incubator-heron/issues/952: explore sharing across topologies
    return HeronDriverConfiguration.CONF
        .setMultiple(DriverConfiguration.GLOBAL_LIBRARIES, libJars)
        .set(DriverConfiguration.DRIVER_IDENTIFIER, topologyName)
        .set(DriverConfiguration.ON_DRIVER_STARTED, HeronSchedulerLauncher.class)
        .set(DriverConfiguration.ON_EVALUATOR_ALLOCATED, ContainerAllocationHandler.class)
        .set(DriverConfiguration.ON_EVALUATOR_FAILED, FailedContainerHandler.class)
        .set(DriverConfiguration.ON_CONTEXT_ACTIVE, HeronWorkerLauncher.class)
        .set(DriverConfiguration.ON_TASK_RUNNING, HeronWorkerStartHandler.class)
        .set(DriverConfiguration.ON_TASK_COMPLETED, HeronWorkerTaskCompletedErrorHandler.class)
        .set(DriverConfiguration.ON_TASK_FAILED, HeronWorkerTaskFailureHandler.class)
        .set(DriverConfiguration.GLOBAL_FILES, topologyPackageLocation)
        .set(DriverConfiguration.GLOBAL_FILES, coreReleasePackage)
        .set(HeronDriverConfiguration.TOPOLOGY_NAME, topologyName)
        .set(HeronDriverConfiguration.TOPOLOGY_JAR, topologyJar)
        .set(HeronDriverConfiguration.TOPOLOGY_PACKAGE_NAME, topologyPackageName)
        .set(HeronDriverConfiguration.HERON_CORE_PACKAGE_NAME, corePackageName)
        .set(HeronDriverConfiguration.ROLE, role)
        .set(HeronDriverConfiguration.ENV, env)
        .set(HeronDriverConfiguration.CLUSTER, cluster)
        .set(HeronDriverConfiguration.HTTP_PORT, 0)
        .set(HeronDriverConfiguration.VERBOSE, false)
        .set(YarnDriverConfiguration.QUEUE, queue)
        .set(DriverConfiguration.DRIVER_MEMORY, driverMemory)
        .build();
  }
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();

    }
300249
public static void addExecutorTopologyArgs(List<String> args, Config config, Config runtime) {
    TopologyAPI.Topology topology = Runtime.topology(runtime);
    args.add(createCommandArg(ExecutorFlag.TopologyName, topology.getName()));
    args.add(createCommandArg(ExecutorFlag.TopologyId, topology.getId()));
    args.add(createCommandArg(ExecutorFlag.TopologyDefinitionFile,
        FileUtils.getBaseName(Context.topologyDefinitionFile(config))));
    args.add(createCommandArg(ExecutorFlag.StateManagerConnection,
        Context.stateManagerConnectionString(config)));
    args.add(createCommandArg(ExecutorFlag.StateManagerRoot,
        Context.stateManagerRootPath(config)));
    args.add(createCommandArg(ExecutorFlag.StateManagerConfigFile,
        Context.stateManagerFile(config)));
    args.add(createCommandArg(ExecutorFlag.TMasterBinary, Context.tmasterBinary(config)));
    args.add(createCommandArg(ExecutorFlag.StmgrBinary, Context.stmgrBinary(config)));
    args.add(createCommandArg(ExecutorFlag.MetricsManagerClasspath,
        Context.metricsManagerClassPath(config)));
    args.add(createCommandArg(ExecutorFlag.InstanceJvmOpts,
        SchedulerUtils.encodeJavaOpts(TopologyUtils.getInstanceJvmOptions(topology))));
    args.add(createCommandArg(ExecutorFlag.Classpath,
        TopologyUtils.makeClassPath(topology, Context.topologyBinaryFile(config))));
    args.add(createCommandArg(ExecutorFlag.HeronInternalsConfigFile,
        Context.systemConfigFile(config)));
    args.add(createCommandArg(ExecutorFlag.OverrideConfigFile, Context.overrideFile(config)));
    args.add(createCommandArg(ExecutorFlag.ComponentRamMap, Runtime.componentRamMap(runtime)));
    args.add(createCommandArg(ExecutorFlag.ComponentJvmOpts,
        SchedulerUtils.encodeJavaOpts(TopologyUtils.getComponentJvmOptions(topology))));
    args.add(createCommandArg(ExecutorFlag.PkgType,
        Context.topologyPackageType(config).name().toLowerCase()));
    args.add(createCommandArg(ExecutorFlag.TopologyBinaryFile,
        Context.topologyBinaryFile(config)));
    args.add(createCommandArg(ExecutorFlag.HeronJavaHome, Context.clusterJavaHome(config)));
    args.add(createCommandArg(ExecutorFlag.HeronShellBinary, Context.shellBinary(config)));
    args.add(createCommandArg(ExecutorFlag.Cluster, Context.cluster(config)));
    args.add(createCommandArg(ExecutorFlag.Role, Context.role(config)));
    args.add(createCommandArg(ExecutorFlag.Environment, Context.environ(config)));
    args.add(createCommandArg(ExecutorFlag.InstanceClasspath, Context.instanceClassPath(config)));
    args.add(createCommandArg(ExecutorFlag.MetricsSinksConfigFile,
        Context.metricsSinksFile(config)));

    String completeSchedulerProcessClassPath = String.format("%s:%s:%s",
        Context.schedulerClassPath(config),
        Context.packingClassPath(config), Context.stateManagerClassPath(config));

    args.add(createCommandArg(ExecutorFlag.SchedulerClasspath, completeSchedulerProcessClassPath));
    args.add(createCommandArg(ExecutorFlag.PythonInstanceBinary,
        Context.pythonInstanceBinary(config)));
    args.add(createCommandArg(ExecutorFlag.CppInstanceBinary, Context.cppInstanceBinary(config)));

    args.add(createCommandArg(ExecutorFlag.MetricsCacheManagerClasspath,
        Context.metricsCacheManagerClassPath(config)));
    String metricscacheMgrMode = Context.metricscacheMgrMode(config)
        == null ? "disabled" : Context.metricscacheMgrMode(config);
    args.add(createCommandArg(ExecutorFlag.MetricsCacheManagerMode, metricscacheMgrMode));

    Boolean ckptMgrEnabled = TopologyUtils.shouldStartCkptMgr(topology);
    args.add(createCommandArg(ExecutorFlag.IsStateful, Boolean.toString(ckptMgrEnabled)));
    String completeCkptmgrProcessClassPath = String.format("%s:%s:%s",
        Context.ckptmgrClassPath(config),
        Context.statefulStoragesClassPath(config),
        Context.statefulStorageCustomClassPath(config));
    args.add(createCommandArg(ExecutorFlag.CheckpointManagerClasspath,
        completeCkptmgrProcessClassPath));
    args.add(createCommandArg(ExecutorFlag.StatefulConfigFile, Context.statefulConfigFile(config)));
    args.add(createCommandArg(
        ExecutorFlag.CheckpointManagerRam,
        Long.toString(TopologyUtils.getCheckpointManagerRam(topology).asBytes())));

    String healthMgrMode = Context.healthMgrMode(config)
        == null ? "disabled" : Context.healthMgrMode(config);
    args.add(createCommandArg(ExecutorFlag.HealthManagerMode, healthMgrMode));
    args.add(createCommandArg(ExecutorFlag.HealthManagerClasspath,
        Context.healthMgrClassPath(config)));
  }
1246950
def main():
    

    logging.captureWarnings(True)
    logging.basicConfig(format=('%(asctime)s - %(name)s - %(levelname)s - ' +
                                '%(message)s'), level=logging.INFO)

    print("=====================================")
    print("========   Submit and Wait   ========")
    print("=====================================")
    print("")

    functionJobs = make_jobs()

    print("sending function jobs to cluster")
    print("")

    job_outputs = process_jobs(functionJobs, max_processes=4)

    print("results from each job")
    for (i, result) in enumerate(job_outputs):
        print("Job {0}- result: {1}".format(i, result))
563001
Task getTaskSpecRawDriver(Task task, String taskName, int containerIndex) {
    String executorBinary = Context.executorBinary(this.clusterConfig);
    // get arguments for heron executor command
    String[] executorArgs = SchedulerUtils.executorCommandArgs(
        this.clusterConfig, this.runtimeConfig, NomadConstants.EXECUTOR_PORTS,
        String.valueOf(containerIndex));
    // get complete heron executor command
    String executorCmd = executorBinary + " " + String.join(" ", executorArgs);
    // get heron_downloader command for downloading topology package
    String topologyDownloadCmd = getFetchCommand(this.localConfig,
        this.clusterConfig, this.runtimeConfig);
    // read nomad heron executor start up script from file
    String heronNomadScript = getHeronNomadScript(this.localConfig);

    task.setName(taskName);
    // use raw_exec driver
    task.setDriver(NomadConstants.NomadDriver.RAW_EXEC.getName());
    // call nomad heron start up script
    task.addConfig(NomadConstants.NOMAD_TASK_COMMAND, NomadConstants.SHELL_CMD);
    String[] args = {NomadConstants.NOMAD_HERON_SCRIPT_NAME};
    task.addConfig(NomadConstants.NOMAD_TASK_COMMAND_ARGS, args);
    Template template = new Template();
    template.setEmbeddedTmpl(heronNomadScript);
    template.setDestPath(NomadConstants.NOMAD_HERON_SCRIPT_NAME);
    task.addTemplates(template);

    // configure nomad to allocate dynamic ports
    Port[] ports = new Port[NomadConstants.EXECUTOR_PORTS.size()];
    int i = 0;
    for (SchedulerUtils.ExecutorPort port : NomadConstants.EXECUTOR_PORTS.keySet()) {
      ports[i] = new Port().setLabel(port.getName().replace("-", "_"));
      i++;
    }

    // set enviroment variables used int the heron nomad start up script
    Map<String, String> envVars = new HashMap<>();
    envVars.put(NomadConstants.HERON_NOMAD_WORKING_DIR,
        NomadContext.workingDirectory(this.localConfig) + "/container-"
            + String.valueOf(containerIndex));

    if (NomadContext.useCorePackageUri(this.localConfig)) {
      envVars.put(NomadConstants.HERON_USE_CORE_PACKAGE_URI, "true");
      envVars.put(NomadConstants.HERON_CORE_PACKAGE_URI,
          NomadContext.corePackageUri(this.localConfig));
    } else {
      envVars.put(NomadConstants.HERON_USE_CORE_PACKAGE_URI, "false");
      envVars.put(NomadConstants.HERON_CORE_PACKAGE_DIR,
          NomadContext.corePackageDirectory(this.localConfig));
    }

    envVars.put(NomadConstants.HERON_TOPOLOGY_DOWNLOAD_CMD, topologyDownloadCmd);
    envVars.put(NomadConstants.HERON_EXECUTOR_CMD, executorCmd);
    task.setEnv(envVars);
    return task;
  }
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);
        }
    }
346580
public static void main(String... args) throws Exception {
		
		if(!parseParameters(args)) {
			return;
		}
		
		// set up execution environment
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		
		// read vertex and edge data
		DataSet<Long> vertices = getVertexDataSet(env);
		DataSet<Tuple2<Long, Long>> edges = getEdgeDataSet(env).flatMap(new UndirectEdge());
		
		// assign the initial components (equal to the vertex id)
		DataSet<Tuple2<Long, Long>> verticesWithInitialId = vertices.map(new DuplicateValue<Long>());
				
		// open a delta iteration
		DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration =
				verticesWithInitialId.iterateDelta(verticesWithInitialId, maxIterations, 0);
		
		// apply the step logic: join with the edges, select the minimum neighbor, update if the component of the candidate is smaller
		DataSet<Tuple2<Long, Long>> changes = iteration.getWorkset().join(edges).where(0).equalTo(0).with(new NeighborWithComponentIDJoin())
				.groupBy(0).aggregate(Aggregations.MIN, 1)
				.join(iteration.getSolutionSet()).where(0).equalTo(0)
				.with(new ComponentIdFilter());

		// close the delta iteration (delta and new workset are identical)
		DataSet<Tuple2<Long, Long>> result = iteration.closeWith(changes, changes);
		
		// emit result
		if(fileOutput) {
			result.writeAsCsv(outputPath, "\n", " ");
		} else {
			result.print();
		}
				
		// execute program
		env.execute("Connected Components Example");
				
	}
782598
def _submit_topology(cmd_args, app):
    
    cfg = app.cfg
    if cmd_args.create_bundle:
        ctxtype = ctx.ContextTypes.BUNDLE
    elif cmd_args.service_name:
        cfg[ctx.ConfigParams.FORCE_REMOTE_BUILD] = True
        cfg[ctx.ConfigParams.SERVICE_NAME] = cmd_args.service_name
        ctxtype = ctx.ContextTypes.STREAMING_ANALYTICS_SERVICE
    sr = ctx.submit(ctxtype, app.app, cfg)
    return sr
473716
protected void copyDataOutBound(int sourceTaskId,
                                  boolean isLocalSpout,
                                  TopologyAPI.StreamId streamId,
                                  HeronTuples.HeronDataTuple tuple,
                                  List<Integer> outTasks) {
    boolean firstIteration = true;
    boolean isAnchored = tuple.getRootsCount() > 0;

    for (Integer outTask : outTasks) {
      long tupleKey = tupleCache.addDataTuple(sourceTaskId, outTask, streamId, tuple, isAnchored);
      if (isAnchored) {
        // Anchored tuple

        if (isLocalSpout) {
          // This is from a local spout. We need to maintain xors
          if (firstIteration) {
            xorManager.create(sourceTaskId, tuple.getRoots(0).getKey(), tupleKey);
          } else {
            xorManager.anchor(sourceTaskId, tuple.getRoots(0).getKey(), tupleKey);
          }
        } else {
          // Anchored emits from local bolt
          for (HeronTuples.RootId rootId : tuple.getRootsList()) {
            HeronTuples.AckTuple t =
                HeronTuples.AckTuple.newBuilder().
                    addRoots(rootId).
                    setAckedtuple(tupleKey).
                    build();

            tupleCache.addEmitTuple(sourceTaskId, rootId.getTaskid(), t);
          }
        }
      }

      firstIteration = false;
    }
  }
634624
def main(http_port, peer_name, node_name, app_id):
    
    # Create the framework
    framework = pelix.framework.create_framework(
        ('pelix.ipopo.core',
         'pelix.ipopo.waiting',
         'pelix.shell.core',
         'pelix.shell.ipopo',
         'pelix.shell.console',
         'pelix.http.basic',

         # Herald core
         'herald.core',
         'herald.directory',
         'herald.shell',

         # Herald HTTP
         'herald.transports.http.directory',
         'herald.transports.http.discovery_multicast',
         'herald.transports.http.servlet',
         'herald.transports.http.transport',

         # RPC
         'pelix.remote.dispatcher',
         'pelix.remote.registry',
         'herald.remote.discovery',
         'herald.remote.herald_xmlrpc',),
        {herald.FWPROP_NODE_UID: node_name,
         herald.FWPROP_NODE_NAME: node_name,
         herald.FWPROP_PEER_NAME: peer_name,
         herald.FWPROP_APPLICATION_ID: app_id})

    # Start everything
    framework.start()
    context = framework.get_bundle_context()

    # Instantiate components
    with use_waiting_list(context) as ipopo:
        # ... HTTP server
        ipopo.add(pelix.http.FACTORY_HTTP_BASIC, "http-server",
                  {pelix.http.HTTP_SERVICE_PORT: http_port})

        # ... HTTP reception servlet
        ipopo.add(herald.transports.http.FACTORY_SERVLET,
                  "herald-http-servlet")

        # ... HTTP multicast discovery
        ipopo.add(herald.transports.http.FACTORY_DISCOVERY_MULTICAST,
                  "herald-http-discovery-multicast")

    # Start the framework and wait for it to stop
    framework.wait_for_stop()
1158197
private static Options constructOptions() {
    Options options = new Options();

    Option cluster = Option.builder("c")
        .desc("Cluster name in which the topology needs to run on")
        .longOpt("cluster")
        .hasArgs()
        .argName("cluster")
        .required()
        .build();

    Option role = Option.builder("r")
        .desc("Role under which the topology needs to run")
        .longOpt("role")
        .hasArgs()
        .argName("role")
        .required()
        .build();

    Option environment = Option.builder("e")
        .desc("Environment under which the topology needs to run")
        .longOpt("environment")
        .hasArgs()
        .argName("environment")
        .required()
        .build();

    Option submitUser = Option.builder("s")
        .desc("User submitting the topology")
        .longOpt("submit_user")
        .hasArgs()
        .argName("submit userid")
        .required()
        .build();

    Option heronHome = Option.builder("d")
        .desc("Directory where heron is installed")
        .longOpt("heron_home")
        .hasArgs()
        .argName("heron home dir")
        .required()
        .build();

    Option configFile = Option.builder("p")
        .desc("Path of the config files")
        .longOpt("config_path")
        .hasArgs()
        .argName("config path")
        .required()
        .build();

    Option configOverrides = Option.builder("o")
        .desc("Command line override config path")
        .longOpt("override_config_file")
        .hasArgs()
        .argName("override config file")
        .build();

    Option releaseFile = Option.builder("b")
        .desc("Release file name")
        .longOpt("release_file")
        .hasArgs()
        .argName("release information")
        .build();

    Option topologyPackage = Option.builder("y")
        .desc("tar ball containing user submitted jar/tar, defn and config")
        .longOpt("topology_package")
        .hasArgs()
        .argName("topology package")
        .required()
        .build();

    Option topologyDefn = Option.builder("f")
        .desc("serialized file containing Topology protobuf")
        .longOpt("topology_defn")
        .hasArgs()
        .argName("topology definition")
        .required()
        .build();

    Option topologyJar = Option.builder("j")
        .desc("The filename of the heron topology jar/tar/pex file to be run by the executor")
        .longOpt("topology_bin")
        .hasArgs()
        .argName("topology binary filename on the cluster")
        .required()
        .build();

    Option dryRun = Option.builder("u")
        .desc("run in dry-run mode")
        .longOpt("dry_run")
        .required(false)
        .build();

    Option dryRunFormat = Option.builder("t")
        .desc("dry-run format")
        .longOpt("dry_run_format")
        .hasArg()
        .required(false)
        .build();

    Option verbose = Option.builder("v")
        .desc("Enable debug logs")
        .longOpt("verbose")
        .build();

    options.addOption(cluster);
    options.addOption(role);
    options.addOption(environment);
    options.addOption(submitUser);
    options.addOption(heronHome);
    options.addOption(configFile);
    options.addOption(configOverrides);
    options.addOption(releaseFile);
    options.addOption(topologyPackage);
    options.addOption(topologyDefn);
    options.addOption(topologyJar);
    options.addOption(dryRun);
    options.addOption(dryRunFormat);
    options.addOption(verbose);

    return options;
  }
342024
void build(int nodeOffset, GraphMeshData graphData, List<int[]> connections) {
		for (int n = 0; n < connections.size(); n++) {
			int[] nodeConnections = connections.get(n);
			MeshData tile = graphData.getTile(n);
			Poly node = graphData.getNode(n);
			for (int connection : nodeConnections) {
				MeshData neighbourTile = graphData.getTile(connection - nodeOffset);
				if (neighbourTile != tile) {
					buildExternalLink(tile, node, neighbourTile);
				} else {
					Poly neighbour = graphData.getNode(connection - nodeOffset);
					buildInternalLink(tile, node, neighbourTile, neighbour);
				}
			}
		}
	}
1165657
protected void processAcksAndFails(int srcTaskId, int taskId,
                                     HeronTuples.HeronControlTupleSet controlTupleSet) {
    HeronTuples.HeronTupleSet.Builder out = HeronTuples.HeronTupleSet.newBuilder();
    out.setSrcTaskId(srcTaskId);

    // First go over emits. This makes sure that new emits makes
    // a tuples stay alive before we process its acks
    for (HeronTuples.AckTuple emitTuple : controlTupleSet.getEmitsList()) {
      for (HeronTuples.RootId rootId : emitTuple.getRootsList()) {
        xorManager.anchor(taskId, rootId.getKey(), emitTuple.getAckedtuple());
      }
    }

    // Then go over acks
    for (HeronTuples.AckTuple ackTuple : controlTupleSet.getAcksList()) {
      for (HeronTuples.RootId rootId : ackTuple.getRootsList()) {
        if (xorManager.anchor(taskId, rootId.getKey(), ackTuple.getAckedtuple())) {
          // This tuple tree is all over

          HeronTuples.AckTuple.Builder a = out.getControlBuilder().addAcksBuilder();
          HeronTuples.RootId.Builder r = a.addRootsBuilder();

          r.setKey(rootId.getKey());
          r.setTaskid(taskId);

          a.setAckedtuple(0); //  This is ignored

          xorManager.remove(taskId, rootId.getKey());
        }
      }
    }

    // Now go over the fails
    for (HeronTuples.AckTuple failTuple : controlTupleSet.getFailsList()) {
      for (HeronTuples.RootId rootId : failTuple.getRootsList()) {
        if (xorManager.remove(taskId, rootId.getKey())) {
          // This tuple tree is failed

          HeronTuples.AckTuple.Builder f =
              out.getControlBuilder().addFailsBuilder();
          HeronTuples.RootId.Builder r = f.addRootsBuilder();

          r.setKey(rootId.getKey());
          r.setTaskid(taskId);
          f.setAckedtuple(0); //  This is ignored
        }
      }
    }

    // Check if we need to send ack tuples to spout task
    if (out.hasControl()) {
      sendMessageToInstance(taskId, out.build());
    }
  }
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));
        }
    }