Create a new Cluster.
Cluster handle pools with patterns and handle failover / distributed load
according to selectors (round robin / random / ordered )

@param args      cluster argurments. see pool-cluster-options.
@constructor
337446
function PoolCluster(args) {
  const opts = new PoolClusterOptions(args);
  const nodes = {};
  let cachedPatterns = {};
  let nodeCounter = 0;

  /**
   * Add a new pool node to cluster.
   *
   * @param id      identifier
   * @param config  pool configuration
   */
  this.add = (id, config) => {
    let identifier;
    if (typeof id === 'string' || id instanceof String) {
      identifier = id;
      if (nodes[identifier])
        throw new Error("Node identifier '" + identifier + "' already exist !");
    } else {
      identifier = 'PoolNode-' + nodeCounter++;
    }
    const options = new PoolOptions(config);
    const pool = _createPool(options);
    pool.initialize();
    nodes[identifier] = pool;
  };

  /**
   * End cluster (and underlying pools).
   *
   * @return {Promise<any[]>}
   */
  this.end = () => {
    cachedPatterns = {};
    const poolEndPromise = [];
    Object.keys(nodes).forEach(pool => {
      poolEndPromise.push(nodes[pool].end());
      delete nodes[pool];
    });

    return Promise.all(poolEndPromise);
  };

  this.of = (pattern, selector) => {
    return new FilteredPoolCluster(this, pattern, selector);
  };

  /**
   * Remove nodes according to pattern.
   *
   * @param pattern  pattern
   */
  this.remove = pattern => {
    if (!pattern)
      throw new Error(
        'pattern parameter in Cluster.remove(pattern)  is mandatory'
      );

    const regex = RegExp(pattern);
    Object.keys(nodes).forEach(key => {
      if (regex.test(key)) {
        nodes[key].end();
        delete nodes[key];
        cachedPatterns = {};
      }
    });
  };

  /**
   * Get connection from available pools matching pattern, according to selector
   *
   * @param pattern       pattern filter (not mandatory)
   * @param selector      node selector ('RR','RANDOM' or 'ORDER')
   * @return {Promise}
   */
  this.getConnection = (pattern, selector) => {
    return _getConnection(pattern, selector);
  };

  /**
   * Force using callback methods.
   */
  this.setCallback = () => {
    this.getConnection = _getConnectionCallback;
    _createPool = _createPoolCallback;
  };

  /**
   * Get connection from available pools matching pattern, according to selector
   * with additional parameter to avoid reusing failing node
   *
   * @param pattern       pattern filter (not mandatory)
   * @param selector      node selector ('RR','RANDOM' or 'ORDER')
   * @param avoidNodeKey  failing node
   * @param lastError     last error
   * @return {Promise}
   * @private
   */
  const _getConnection = (pattern, selector, avoidNodeKey, lastError) => {
    const matchingNodeList = _matchingNodes(pattern || /^/);

    if (matchingNodeList.length === 0) {
      if (Object.keys(nodes).length === 0)
        return Promise.reject(
          new Error(
            'No node have been added to cluster or nodes have been removed due to too much connection error'
          )
        );
      if (avoidNodeKey === undefined)
        return Promise.reject(
          new Error("No node found for pattern '" + pattern + "'")
        );
      const errMsg =
        "No Connection available for '" +
        pattern +
        "'" +
        (lastError ? '. Last connection error was: ' + lastError.message : '');
      return Promise.reject(new Error(errMsg));
    }

    const retry = _getConnection.bind(this, pattern, selector);
    try {
      const nodeKey = _selectPool(matchingNodeList, selector, avoidNodeKey);
      return _handleConnectionError(matchingNodeList, nodeKey, retry);
    } catch (e) {
      return Promise.reject(e);
    }
  };

  let _createPool = options => {
    return new Pool(options, false);
  };

  const _createPoolCallback = options => {
    return new PoolCallback(options, false);
  };

  /**
   * Get connection from available pools matching pattern, according to selector
   * with additional parameter to avoid reusing failing node
   *
   * @param pattern       pattern filter (not mandatory)
   * @param selector      node selector ('RR','RANDOM' or 'ORDER')
   * @param callback      callback function
   * @param avoidNodeKey  failing node
   * @private
   */
  const _getConnectionCallback = (
    pattern,
    selector,
    callback,
    avoidNodeKey
  ) => {
    const matchingNodeList = _matchingNodes(pattern || /^/);

    if (matchingNodeList.length === 0) {
      if (Object.keys(nodes).length === 0) {
        callback(
          new Error(
            'No node have been added to cluster or nodes have been removed due to too much connection error'
          )
        );
        return;
      }
      callback(new Error("No node found for pattern '" + pattern + "'"));
      return;
    }

    const retry = _getConnectionCallback.bind(
      this,
      pattern,
      selector,
      callback
    );
    try {
      const nodeKey = _selectPool(matchingNodeList, selector, avoidNodeKey);
      _handleConnectionCallbackError(
        matchingNodeList,
        nodeKey,
        retry,
        callback
      );
    } catch (e) {
      callback(e);
    }
  };

  /**
   * Selecting nodes according to pattern.
   *
   * @param pattern pattern
   * @return {*}
   * @private
   */
  const _matchingNodes = pattern => {
    if (cachedPatterns[pattern]) return cachedPatterns[pattern];

    const regex = RegExp(pattern);
    const matchingNodeList = [];
    Object.keys(nodes).forEach(key => {
      if (regex.test(key)) {
        matchingNodeList.push(key);
      }
    });

    cachedPatterns[pattern] = matchingNodeList;
    return matchingNodeList;
  };

  /**
   * Select next node to be chosen in nodeList according to selector and failed nodes.
   *
   * @param nodeList        current node list
   * @param selectorParam   selector
   * @param avoidNodeKey    last failing node to avoid selecting this one.
   * @return {Promise}
   * @private
   */
  const _selectPool = (nodeList, selectorParam, avoidNodeKey) => {
    const selector = selectorParam || opts.defaultSelector;
    let retry = 0;
    let selectorFct;
    let nodeKey;
    switch (selector) {
      case 'RR':
        selectorFct = roundRobinSelector;
        break;

      case 'RANDOM':
        selectorFct = randomSelector;
        break;

      case 'ORDER':
        selectorFct = orderedSelector;
        break;

      default:
        throw new Error(
          "Wrong selector value '" +
            selector +
            "'. Possible values are 'RR','RANDOM' or 'ORDER'"
        );
    }

    nodeKey = selectorFct(nodeList, retry);
    while (
      (avoidNodeKey === nodeKey || nodes[nodeKey].timeout > Date.now()) &&
      retry < nodeList.length - 1
    ) {
      retry++;
      nodeKey = selectorFct(nodeList);
    }
    return nodeKey;
  };

  /**
   * Round robin selector: using nodes one after the other.
   *
   * @param nodeList  node list
   * @return {String}
   */
  const roundRobinSelector = nodeList => {
    let lastRoundRobin = nodeList.lastRrIdx;
    if (lastRoundRobin === undefined) lastRoundRobin = -1;
    if (++lastRoundRobin >= nodeList.length) lastRoundRobin = 0;
    nodeList.lastRrIdx = lastRoundRobin;
    return nodeList[lastRoundRobin];
  };

  /**
   * Random selector: use a random node.
   *
   * @param nodeList  node list
   * @return {String}
   */
  const randomSelector = nodeList => {
    let randomIdx = Math.floor(Math.random() * nodeList.length);
    return nodeList[randomIdx];
  };

  /**
   * Ordered selector: always use the nodes in sequence, unless failing.
   *
   * @param nodeList  node list
   * @param retry     sequence number if last node is tagged has failing
   * @return {String}
   */
  const orderedSelector = (nodeList, retry) => {
    return nodeList[retry];
  };

  /**
   * Connect, or if fail handle retry / set timeout error
   *
   * @param nodeList    current node list
   * @param nodeKey     node name to connect
   * @param retryFct    retry function
   * @return {Promise}
   * @private
   */
  const _handleConnectionError = (nodeList, nodeKey, retryFct) => {
    const node = nodes[nodeKey];
    return node
      .getConnection()
      .then(conn => {
        node.errorCount = 0;
        return Promise.resolve(conn);
      })
      .catch(err => {
        node.errorCount = node.errorCount ? node.errorCount + 1 : 1;
        if (node.errorCount >= opts.removeNodeErrorCount && nodes[nodeKey]) {
          if (opts.restoreNodeTimeout === 0) {
            delete nodes[nodeKey];
            cachedPatterns = {};
            delete nodeList.lastRrIdx;
            //remove node from configuration if not already removed
            node.end().catch(err => {
              // dismiss error
            });
          } else {
            node.timeout = Date.now() + opts.restoreNodeTimeout;
          }
          if (nodeList.length === 0) return Promise.reject(err);
        }

        if (opts.canRetry) return retryFct(nodeKey, err);
        return Promise.reject(err);
      });
  };

  /**
   * Connect, or if fail handle retry / set timeout error
   *
   * @param nodeList    current node list
   * @param nodeKey     node name to connect
   * @param retryFct    retry function
   * @param callback    callback function
   * @private
   */
  const _handleConnectionCallbackError = (
    nodeList,
    nodeKey,
    retryFct,
    callback
  ) => {
    const node = nodes[nodeKey];
    node.getConnection((err, conn) => {
      if (err) {
        node.errorCount = node.errorCount ? node.errorCount + 1 : 1;
        if (node.errorCount >= opts.removeNodeErrorCount && nodes[nodeKey]) {
          if (opts.restoreNodeTimeout === 0) {
            delete nodes[nodeKey];
            cachedPatterns = {};
            delete nodeList.lastRrIdx;
            //remove node from configuration if not already removed
            node.end(() => {
              //dismiss error
            });
          } else {
            node.timeout = Date.now() + opts.restoreNodeTimeout;
          }
          if (nodeList.length === 0) return Promise.reject(err);
        }

        if (opts.canRetry) return retryFct(nodeKey);
        callback(err);
      } else {
        node.errorCount = 0;
        callback(null, conn);
      }
    });
  };

  //*****************************************************************
  // internal public testing methods
  //*****************************************************************

  function TestMethods() {}
  TestMethods.prototype.getNodes = () => {
    return nodes;
  };

  this.__tests = new TestMethods();
}
377822
public static void main(String[] args) {

		 List<String> tmpArgs = new ArrayList<String>(Arrays.asList(args));
		
		boolean isLocal = isLocal(tmpArgs);
		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);
    			
    	
		if(isLocal){
			//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();
			
		}else{
			//cluster mode
			conf.setNumWorkers(numWorker);
			try {
				backtype.storm.StormSubmitter.submitTopology(topologyName, conf,
						stormTopo.getStormBuilder().createTopology());
			} catch (backtype.storm.generated.AlreadyAliveException ale) {
				ale.printStackTrace();
			} catch (backtype.storm.generated.InvalidTopologyException ite) {
				ite.printStackTrace();
			}
		}
	}
1317765
def create_cluster(self):
        
        self.create_stack(
            self.cluster_name,
            'amazon-eks-cluster.yaml',
            parameters=define_parameters(
                ClusterName=self.cluster_name,
                ControlPlaneSecurityGroup=self.security_groups,
                Subnets=self.subnet_ids
            )
        )
869382
def create(self, params):
        
        sh_id = params.get('id', str(uuid4()))
        if sh_id in self:
            raise ShardedClusterError(
                "Sharded cluster with id %s already exists." % sh_id)
        params['id'] = sh_id
        cluster = ShardedCluster(params)
        self[cluster.id] = cluster
        return cluster.id
453320
def cluster(args):
    
    p = OptionParser(cluster.__doc__)
    add_consensus_options(p)
    p.set_align(pctid=95)
    p.set_outdir()
    p.set_cpus()
    opts, args = p.parse_args(args)

    if len(args) < 2:
        sys.exit(not p.print_help())

    prefix = args[0]
    fastqfiles = args[1:]
    cpus = opts.cpus
    pctid = opts.pctid
    mindepth = opts.mindepth
    minlength = opts.minlength
    fastafile, qualfile = fasta(fastqfiles + ["--seqtk",
                                "--outdir={0}".format(opts.outdir),
                                "--outfile={0}".format(prefix + ".fasta")])

    prefix = op.join(opts.outdir, prefix)
    pf = prefix + ".P{0}".format(pctid)
    derepfile = prefix + ".derep"
    if need_update(fastafile, derepfile):
        derep(fastafile, derepfile, minlength, cpus)

    userfile = pf + ".u"
    notmatchedfile = pf + ".notmatched"
    if need_update(derepfile, userfile):
        cluster_smallmem(derepfile, userfile, notmatchedfile,
                         minlength, pctid, cpus)

    clustfile = pf + ".clust"
    if need_update((derepfile, userfile, notmatchedfile), clustfile):
        makeclust(derepfile, userfile, notmatchedfile, clustfile,
                  mindepth=mindepth)

    clustSfile = pf + ".clustS"
    if need_update(clustfile, clustSfile):
        parallel_musclewrap(clustfile, cpus)

    statsfile = pf + ".stats"
    if need_update(clustSfile, statsfile):
        makestats(clustSfile, statsfile, mindepth=mindepth)
656849
public void startNewBatchForCluster(String cluster) {
        // Only start a new batch if the current one is non-empty
        PlanBatch batch = _clusterTails.get(cluster);
        if (batch != null && !batch.getItems().isEmpty()) {
            PlanBatch newBatch = new PlanBatch();
            batch.setNextBatch(newBatch);
            _clusterTails.put(cluster, newBatch);
        }
    }
1001375
function (i) {
                    workers[i] = cluster.fork();

                    console.log('Creating Worker: ', workers[i].process.pid);
                    workers[i].on('message', m => {
                        switch (m.type) {
                            case 'STARTED':
                                !startedWorkers && startWatches();
                                break;
                        }
                    });
                    // Optional: Restart worker on exit
                    workers[i].on('exit', function (code, signal) {
                        console.log('Respawning worker', i);
                        spawn(i);
                    });
                }
315908
public static void main(String[] args) {
    if (args.length != 0) {
      LOG.info("java -cp {} {}", RuntimeConstants.ALLUXIO_JAR,
          AlluxioWorker.class.getCanonicalName());
      System.exit(-1);
    }

    if (!ConfigurationUtils.masterHostConfigured(ServerConfiguration.global())) {
      ProcessUtils.fatalError(LOG,
          ConfigurationUtils.getMasterHostNotConfiguredMessage("Alluxio worker"));
    }

    CommonUtils.PROCESS_TYPE.set(CommonUtils.ProcessType.WORKER);
    MasterInquireClient masterInquireClient =
        MasterInquireClient.Factory.create(ServerConfiguration.global());
    try {
      RetryUtils.retry("load cluster default configuration with master", () -> {
        InetSocketAddress masterAddress = masterInquireClient.getPrimaryRpcAddress();
        ServerConfiguration.loadClusterDefaults(masterAddress);
      }, RetryUtils.defaultWorkerMasterClientRetry(
          ServerConfiguration.getDuration(PropertyKey.WORKER_MASTER_CONNECT_RETRY_TIMEOUT)));
    } catch (IOException e) {
      ProcessUtils.fatalError(LOG,
          "Failed to load cluster default configuration for worker: %s", e.getMessage());
    }
    WorkerProcess process = WorkerProcess.Factory.create();
    ProcessUtils.run(process);
  }
193274
public static void main(String[] args) {
		EnvironmentInformation.logEnvironmentInfo(LOG, "YARN TaskExecutor runner", args);
		SignalHandler.register(LOG);
		JvmShutdownSafeguard.installAsShutdownHook(LOG);

		run(args);
	}
1071813
private static void run(String[] args) {
		try {
			LOG.debug("All environment variables: {}", ENV);

			final String currDir = ENV.get(Environment.PWD.key());
			LOG.info("Current working Directory: {}", currDir);

			final Configuration configuration = GlobalConfiguration.loadConfiguration(currDir);

			//TODO provide path.
			FileSystem.initialize(configuration, PluginUtils.createPluginManagerFromRootFolder(Optional.empty()));

			setupConfigurationAndInstallSecurityContext(configuration, currDir, ENV);

			final String containerId = ENV.get(YarnResourceManager.ENV_FLINK_CONTAINER_ID);
			Preconditions.checkArgument(containerId != null,
				"ContainerId variable %s not set", YarnResourceManager.ENV_FLINK_CONTAINER_ID);

			SecurityUtils.getInstalledContext().runSecured((Callable<Void>) () -> {
				TaskManagerRunner.runTaskManager(configuration, new ResourceID(containerId));
				return null;
			});
		}
		catch (Throwable t) {
			final Throwable strippedThrowable = ExceptionUtils.stripException(t, UndeclaredThrowableException.class);
			// make sure that everything whatever ends up in the log
			LOG.error("YARN TaskManager initialization failed.", strippedThrowable);
			System.exit(INIT_ERROR_EXIT_CODE);
		}
	}
478616
public static <K, V> Map<K, V> newHashMap(Object... args) {
        Map<K, V> m = new HashMap<K, V>();
        for (int i = 0; i < args.length; i += 2) {
            m.put((K) args[i], i + 1 >= args.length ? null : (V) args[i + 1]);
        }
        return m;
    }
156973
public static void main(String[] args) {
    if (args.length != 0) {
      LOG.info("java -cp {} {}", RuntimeConstants.ALLUXIO_JAR,
          AlluxioJobWorker.class.getCanonicalName());
      System.exit(-1);
    }

    if (!ConfigurationUtils.masterHostConfigured(ServerConfiguration.global())) {
      System.out.println(ConfigurationUtils
          .getMasterHostNotConfiguredMessage("Alluxio job worker"));
      System.exit(1);
    }

    if (!ConfigurationUtils.jobMasterHostConfigured(ServerConfiguration.global())) {
      System.out.println(ConfigurationUtils
          .getJobMasterHostNotConfiguredMessage("Alluxio job worker"));
      System.exit(1);
    }

    CommonUtils.PROCESS_TYPE.set(CommonUtils.ProcessType.JOB_WORKER);
    MasterInquireClient masterInquireClient =
        MasterInquireClient.Factory.create(ServerConfiguration.global());
    try {
      RetryUtils.retry("load cluster default configuration with master", () -> {
        InetSocketAddress masterAddress = masterInquireClient.getPrimaryRpcAddress();
        ServerConfiguration.loadClusterDefaults(masterAddress);
      },
          RetryUtils.defaultWorkerMasterClientRetry(
              ServerConfiguration.getDuration(PropertyKey.WORKER_MASTER_CONNECT_RETRY_TIMEOUT)));
    } catch (IOException e) {
      ProcessUtils.fatalError(LOG,
          "Failed to load cluster default configuration for job worker: %s", e.getMessage());
    }
    JobWorkerProcess process = JobWorkerProcess.Factory.create();
    ProcessUtils.run(process);
  }
904997
public long allocAppend(long cluster)
            throws IOException {
        
        testCluster(cluster);
        
        while (!isEofCluster(entries[(int) cluster])) {
            cluster = entries[(int) cluster];
        }
        
        long newCluster = allocNew();
        entries[(int) cluster] = newCluster;

        return newCluster;
    }
930621
public static void main(String[] args) throws Exception {
    MapReduceIntegrationChecker checker = new MapReduceIntegrationChecker();
    System.exit(checker.run(args));
  }
1306849
public int run(String[] args) throws Exception {

    JobConf jobConf = new JobConf(getConf(), Sort.class);
    jobConf.setJobName("sorter");

    jobConf.setMapperClass(IdentityMapper.class);        
    jobConf.setReducerClass(IdentityReducer.class);

    JobClient client = new JobClient(jobConf);
    ClusterStatus cluster = client.getClusterStatus();
    int num_reduces = (int) (cluster.getMaxReduceTasks() * 0.9);
    String sort_reduces = jobConf.get("test.sort.reduces_per_host");
    if (sort_reduces != null) {
       num_reduces = cluster.getTaskTrackers() * 
                       Integer.parseInt(sort_reduces);
    }
    Class<? extends InputFormat> inputFormatClass = 
      SequenceFileInputFormat.class;
    Class<? extends OutputFormat> outputFormatClass = 
      SequenceFileOutputFormat.class;
    Class<? extends WritableComparable> outputKeyClass = BytesWritable.class;
    Class<? extends Writable> outputValueClass = BytesWritable.class;
    List<String> otherArgs = new ArrayList<String>();
    InputSampler.Sampler<K,V> sampler = null;
    for(int i=0; i < args.length; ++i) {
      try {
        if ("-m".equals(args[i])) {
          jobConf.setNumMapTasks(Integer.parseInt(args[++i]));
        } else if ("-r".equals(args[i])) {
          num_reduces = Integer.parseInt(args[++i]);
        } else if ("-inFormat".equals(args[i])) {
          inputFormatClass = 
            Class.forName(args[++i]).asSubclass(InputFormat.class);
        } else if ("-outFormat".equals(args[i])) {
          outputFormatClass = 
            Class.forName(args[++i]).asSubclass(OutputFormat.class);
        } else if ("-outKey".equals(args[i])) {
          outputKeyClass = 
            Class.forName(args[++i]).asSubclass(WritableComparable.class);
        } else if ("-outValue".equals(args[i])) {
          outputValueClass = 
            Class.forName(args[++i]).asSubclass(Writable.class);
        } else if ("-totalOrder".equals(args[i])) {
          double pcnt = Double.parseDouble(args[++i]);
          int numSamples = Integer.parseInt(args[++i]);
          int maxSplits = Integer.parseInt(args[++i]);
          if (0 >= maxSplits) maxSplits = Integer.MAX_VALUE;
          sampler =
            new InputSampler.RandomSampler<K,V>(pcnt, numSamples, maxSplits);
        } else {
          otherArgs.add(args[i]);
        }
      } catch (NumberFormatException except) {
        System.out.println("ERROR: Integer expected instead of " + args[i]);
        return printUsage();
      } catch (ArrayIndexOutOfBoundsException except) {
        System.out.println("ERROR: Required parameter missing from " +
            args[i-1]);
        return printUsage(); // exits
      }
    }

    // Set user-supplied (possibly default) job configs
    jobConf.setNumReduceTasks(num_reduces);

    jobConf.setInputFormat(inputFormatClass);
    jobConf.setOutputFormat(outputFormatClass);

    jobConf.setOutputKeyClass(outputKeyClass);
    jobConf.setOutputValueClass(outputValueClass);

    // Make sure there are exactly 2 parameters left.
    if (otherArgs.size() != 2) {
      System.out.println("ERROR: Wrong number of parameters: " +
          otherArgs.size() + " instead of 2.");
      return printUsage();
    }
    FileInputFormat.setInputPaths(jobConf, otherArgs.get(0));
    FileOutputFormat.setOutputPath(jobConf, new Path(otherArgs.get(1)));

    if (sampler != null) {
      System.out.println("Sampling input to effect total-order sort...");
      jobConf.setPartitionerClass(TotalOrderPartitioner.class);
      Path inputDir = FileInputFormat.getInputPaths(jobConf)[0];
      inputDir = inputDir.makeQualified(inputDir.getFileSystem(jobConf));
      Path partitionFile = new Path(inputDir, "_sortPartitioning");
      TotalOrderPartitioner.setPartitionFile(jobConf, partitionFile);
      InputSampler.<K,V>writePartitionFile(jobConf, sampler);
      URI partitionUri = new URI(partitionFile.toString() +
                                 "#" + "_sortPartitioning");
      DistributedCache.addCacheFile(partitionUri, jobConf);
      DistributedCache.createSymlink(jobConf);
    }

    System.out.println("Running on " +
        cluster.getTaskTrackers() +
        " nodes to sort from " + 
        FileInputFormat.getInputPaths(jobConf)[0] + " into " +
        FileOutputFormat.getOutputPath(jobConf) +
        " with " + num_reduces + " reduces.");
    Date startTime = new Date();
    System.out.println("Job started: " + startTime);
    jobResult = JobClient.runJob(jobConf);
    Date end_time = new Date();
    System.out.println("Job ended: " + end_time);
    System.out.println("The job took " + 
        (end_time.getTime() - startTime.getTime()) /1000 + " seconds.");
    return 0;
  }
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);
  }
1022977
protected final void args(Object... args) {
        if (args == null) args = NO_ARGS;
        this.args = args;
        retry();
    }