<commit_before>package com.demandcube.yarn;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.GnuParser;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerExitStatus;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
import org.apache.hadoop.yarn.api.records.ContainerState;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.NodeReport;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.client.api.NMClient;
import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.util.Records;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;

public abstract class AbstractApplicationMaster {
  private AMRMClientAsync<ContainerRequest> resourceManager;
  private NMClient nodeManager;

  private Configuration conf;
  protected static final Logger LOG = Logger.getLogger(AbstractApplicationMaster.class.getName());
  protected Options opts;
  private int containerMem;

  private int totalContainerCount;
  private String command;
  private AtomicInteger completedContainerCount;
  private AtomicInteger allocatedContainerCount;
  private AtomicInteger failedContainerCount;
  private AtomicInteger requestedContainerCount;

  private String appMasterHostname = "";     // TODO: What should this really be?
  private int appMasterRpcPort = 0;          // TODO: What should this really be?
  private String appMasterTrackingUrl = "";  // TODO: What should this really be?

  private boolean done;
  protected Map<ContainerId, String> containerIdCommandMap;
  protected List<String> failedCommandList;

  public AbstractApplicationMaster() {
    conf = new YarnConfiguration();
    opts = new Options();
    completedContainerCount = new AtomicInteger();
    allocatedContainerCount = new AtomicInteger();
    failedContainerCount = new AtomicInteger();
    requestedContainerCount = new AtomicInteger();

    opts.addOption(Constants.OPT_CONTAINER_MEM, true, "container memory");
    opts.addOption(Constants.OPT_CONTAINER_COUNT, true, "number of containers");
    //opts.addOption(Constants.OPT_COMMAND, true, "Command to run on the cluster.");

    containerIdCommandMap = new HashMap<ContainerId, String>();
    failedCommandList = new ArrayList<String>();
  }

  public void init(String[] args) throws ParseException {
    LOG.setLevel(Level.INFO);
    CommandLine cliParser = new GnuParser().parse(this.opts, args);
    done = false;

    this.containerMem = Integer.parseInt( cliParser.getOptionValue(Constants.OPT_CONTAINER_MEM) );
    this.totalContainerCount = Integer.parseInt( cliParser.getOptionValue(Constants.OPT_CONTAINER_COUNT) );
    this.command = cliParser.getOptionValue(Constants.OPT_COMMAND);
    LOG.info("command template: this.command");
  }

  public boolean run() throws IOException, YarnException {
    // Initialize clients to RM and NMs.
    LOG.info("ApplicationMaster::run"); //xxx
    AMRMClientAsync.CallbackHandler rmListener = new RMCallbackHandler();
    resourceManager = AMRMClientAsync.createAMRMClientAsync(1000, rmListener);
    resourceManager.init(conf);
    resourceManager.start();

    nodeManager = NMClient.createNMClient();
    nodeManager.init(conf);
    nodeManager.start();<commit_msg>Fixed: OPT_COMMAND option was incorrectly commented out.<commit_after>package com.demandcube.yarn;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.GnuParser;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerExitStatus;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
import org.apache.hadoop.yarn.api.records.ContainerState;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.NodeReport;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.client.api.NMClient;
import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.util.Records;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;

public abstract class AbstractApplicationMaster {
  private AMRMClientAsync<ContainerRequest> resourceManager;
  private NMClient nodeManager;

  private Configuration conf;
  protected static final Logger LOG = Logger.getLogger(AbstractApplicationMaster.class.getName());
  protected Options opts;
  private int containerMem;

  protected int totalContainerCount;
  private String command;
  private AtomicInteger completedContainerCount;
  private AtomicInteger allocatedContainerCount;
  private AtomicInteger failedContainerCount;
  private AtomicInteger requestedContainerCount;

  private String appMasterHostname = "";     // TODO: What should this really be?
  private int appMasterRpcPort = 0;          // TODO: What should this really be?
  private String appMasterTrackingUrl = "";  // TODO: What should this really be?

  private boolean done;
  protected Map<ContainerId, String> containerIdCommandMap;
  protected List<String> failedCommandList;

  public AbstractApplicationMaster() {
    conf = new YarnConfiguration();
    opts = new Options();
    completedContainerCount = new AtomicInteger();
    allocatedContainerCount = new AtomicInteger();
    failedContainerCount = new AtomicInteger();
    requestedContainerCount = new AtomicInteger();

    opts.addOption(Constants.OPT_CONTAINER_MEM, true, "container memory");
    opts.addOption(Constants.OPT_CONTAINER_COUNT, true, "number of containers");
    opts.addOption(Constants.OPT_COMMAND, true, "Command to run on the cluster.");

    containerIdCommandMap = new HashMap<ContainerId, String>();
    failedCommandList = new ArrayList<String>();
  }

  public void init(String[] args) throws ParseException {
    LOG.setLevel(Level.INFO);
    CommandLine cliParser = new GnuParser().parse(this.opts, args);
    done = false;

    this.containerMem = Integer.parseInt( cliParser.getOptionValue(Constants.OPT_CONTAINER_MEM) );
    this.totalContainerCount = Integer.parseInt( cliParser.getOptionValue(Constants.OPT_CONTAINER_COUNT) );
    this.command = cliParser.getOptionValue(Constants.OPT_COMMAND);
    LOG.info("command template: this.command");
  }

  public boolean run() throws IOException, YarnException {
    // Initialize clients to RM and NMs.
    LOG.info("ApplicationMaster::run"); //xxx
    AMRMClientAsync.CallbackHandler rmListener = new RMCallbackHandler();
    resourceManager = AMRMClientAsync.createAMRMClientAsync(1000, rmListener);
    resourceManager.init(conf);
    resourceManager.start();

    nodeManager = NMClient.createNMClient();
    nodeManager.init(conf);
    nodeManager.start();
{'fraction_non_alphanumeric': 0.09925979856813494, 'fraction_numerical': 0.0016988229583788375, 'mean_word_length': 5.33026113671275, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '137981', 'n_tokens_mistral': 2326, 'n_tokens_neox': 2279, 'n_words': 487}