content: /**
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   * http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package com.alibaba.jstorm.yarn.utils;
  
  import com.alibaba.jstorm.yarn.JstormOnYarn;
  import com.alibaba.jstorm.yarn.Log4jPropertyHelper;
  import com.alibaba.jstorm.yarn.appmaster.JstormMaster;
  import com.alibaba.jstorm.yarn.constants.JOYConstants;
  import com.alibaba.jstorm.yarn.constants.JstormXmlConfKeys;
  import com.alibaba.jstorm.yarn.context.JstormClientContext;
  import com.alibaba.jstorm.yarn.context.JstormMasterContext;
  import com.alibaba.jstorm.yarn.context.MasterContext;
  import org.apache.commons.cli.CommandLine;
  import org.apache.commons.cli.Option;
  import org.apache.commons.cli.Options;
  import org.apache.commons.io.FileUtils;
  import org.apache.commons.io.IOUtils;
  import org.apache.commons.lang.StringUtils;
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.registry.client.api.BindFlags;
  import org.apache.hadoop.registry.client.api.RegistryOperations;
  import org.apache.hadoop.registry.client.binding.RegistryUtils;
  import org.apache.hadoop.registry.client.types.ServiceRecord;
  import org.apache.hadoop.util.Shell;
  import org.apache.hadoop.yarn.api.ApplicationConstants;
  import org.apache.hadoop.yarn.api.records.ContainerId;
  import org.apache.hadoop.yarn.conf.YarnConfiguration;
  import org.apache.hadoop.yarn.util.ConverterUtils;
  
  import java.io.*;
  import java.net.ServerSocket;
  import java.net.Socket;
  import java.util.*;
  
  import static com.alibaba.jstorm.yarn.constants.JOYConstants.log4jPath;
  import static com.alibaba.jstorm.yarn.constants.JOYConstants.shellArgsPath;
  import static com.alibaba.jstorm.yarn.constants.JOYConstants.shellCommandPath;
  
  /**
   * Created by fengjian on 15/12/29.
   */
  public class JstormYarnUtils {
      private static final Log LOG = LogFactory.getLog(JstormYarnUtils.class);
  
      /**
       * Implementation of set-ness, groovy definition of true/false for a string
       *
       * @param s string
       * @return true iff the string is neither null nor empty
       */
      public static boolean isUnset(String s) {
          return s == null || s.isEmpty();
      }
  
      public static boolean isSet(String s) {
          return !isUnset(s);
  
      }
  
      public static int findFreePort(int start, int limit) {
          if (start == 0) {
              //bail out if the default is "dont care"
              return 0;
          }
          int found = 0;
          int port = start;
          int finish = start + limit;
          while (found == 0 && port < finish) {
              if (isPortAvailable(port)) {
                  found = port;
              } else {
                  port++;
              }
          }
          return found;
      }
  
      /**
       * See if a port is available for listening on by trying to listen
       * on it and seeing if that works or fails.
       *
       * @param port port to listen to
       * @return true if the port was available for listening on
       */
      public static boolean isPortAvailable(int port) {
          try {
              ServerSocket socket = new ServerSocket(port);
              socket.close();
              return true;
          } catch (IOException e) {
              return false;
          }
      }
  
      /**
       * See if a port is available for listening on by trying connect to it
       * and seeing if that works or fails
       *
       * @param host
       * @param port
       * @return
       */
      public static boolean isPortAvailable(String host, int port) {
          try {
              Socket socket = new Socket(host, port);
              socket.close();
              return false;
          } catch (IOException e) {
              return true;
          }
      }
  
      public static String stringify(Throwable t) {
          StringWriter sw = new StringWriter();
          sw.append(t.toString()).append('\n');
          t.printStackTrace(new PrintWriter(sw));
          return sw.toString();
      }
  
      /**
       * Create a configuration with Slider-specific tuning.
       * This is done rather than doing custom configs.
       *
       * @return the config
       */
      public static YarnConfiguration createConfiguration() {
          YarnConfiguration conf = new YarnConfiguration();
          patchConfiguration(conf);
          return conf;
      }
  
      /**
       * Take an existing conf and patch it for  needs. Useful
       * in Service.init & RunService methods where a shared config is being
       * passed in
       *
       * @param conf configuration
       * @return the patched configuration
       */
      public static Configuration patchConfiguration(Configuration conf) {
  
          //if the fallback option is NOT set, enable it.
          //if it is explicitly set to anything -leave alone
          if (conf.get(JstormXmlConfKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH) == null) {
              conf.set(JstormXmlConfKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH, "true");
          }
          return conf;
      }
  
      /**
       * Take a collection, return a list containing the string value of every
       * element in the collection.
       *
       * @param c collection
       * @return a stringified list
       */
      public static List<String> collectionToStringList(Collection c) {
          List<String> l = new ArrayList<String>(c.size());
          for (Object o : c) {
              l.add(o.toString());
          }
          return l;
      }
  
      /**
       * Join an collection of objects with a separator that appears after every
       * instance in the list -including at the end
       *
       * @param collection collection to call toString() on each element
       * @param separator  separator string
       * @return the joined entries
       */
      public static String join(Collection collection, String separator) {
          return join(collection, separator, true);
      }
  
      /**
       * Join an collection of objects with a separator that appears after every
       * instance in the list -optionally at the end
       *
       * @param collection collection to call toString() on each element
       * @param separator  separator string
       * @param trailing   add a trailing entry or not
       * @return the joined entries
       */
      public static String join(Collection collection,
                                String separator,
                                boolean trailing) {
          StringBuilder b = new StringBuilder();
          // fast return on empty collection
          if (collection.isEmpty()) {
              return trailing ? separator : "";
          }
          for (Object o : collection) {
              b.append(o);
              b.append(separator);
          }
          int length = separator.length();
          String s = b.toString();
          return (trailing || s.isEmpty()) ?
                  s : (b.substring(0, b.length() - length));
      }
  
      /**
       * Join an array of strings with a separator that appears after every
       * instance in the list -including at the end
       *
       * @param collection strings
       * @param separator  separator string
       * @return the joined entries
       */
      public static String join(String[] collection, String separator) {
          return join(collection, separator, true);
  
  
      }
  
      /**
       * Join an array of strings with a separator that appears after every
       * instance in the list -optionally at the end
       *
       * @param collection strings
       * @param separator  separator string
       * @param trailing   add a trailing entry or not
       * @return the joined entries
       */
      public static String join(String[] collection, String separator,
                                boolean trailing) {
          return join(Arrays.asList(collection), separator, trailing);
      }
  
      /**
       * Join an array of strings with a separator that appears after every
       * instance in the list -except at the end
       *
       * @param collection strings
       * @param separator  separator string
       * @return the list
       */
      public static String joinWithInnerSeparator(String separator,
                                                  Object... collection) {
          StringBuilder b = new StringBuilder();
          boolean first = true;
  
          for (Object o : collection) {
              if (first) {
                  first = false;
              } else {
                  b.append(separator);
              }
              b.append(o.toString());
              b.append(separator);
          }
          return b.toString();
      }
  
      /**
       * this is for jstorm configuration's format
       *
       * @param memory
       * @param vcores
       * @param supervisorHost
       * @return
       */
      public static String getSupervisorSlotPorts(int memory, int vcores, String instanceName, String supervisorHost, RegistryOperations registryOperations) {
          return join(getSupervisorPorts(memory, vcores, instanceName, supervisorHost, registryOperations), JOYConstants.COMMA, false);
      }
  
      /**
       * see if port is using by supervisor
       * if used return true, if not, add this port to registry and return false
       *
       * @param supervisorHost
       * @param port
       * @param registryOperations
       * @return
       */
      public static boolean getSetPortUsedBySupervisor(String instanceName, String supervisorHost, int port, RegistryOperations registryOperations) {
  
          String appPath = RegistryUtils.serviceclassPath(
                  JOYConstants.APP_NAME, JOYConstants.APP_TYPE);
          String path = RegistryUtils.servicePath(
                  JOYConstants.APP_NAME, JOYConstants.APP_TYPE, instanceName);
          String hostPath = RegistryUtils.componentPath(
                  JOYConstants.APP_NAME, JOYConstants.APP_TYPE, instanceName, supervisorHost);
  
          try {
              List<String> instanceNames = registryOperations.list(appPath);
              for (String instance : instanceNames) {
                  String servicePath = RegistryUtils.servicePath(
                          JOYConstants.APP_NAME, JOYConstants.APP_TYPE, instance);
                  Map<String, ServiceRecord> hosts = RegistryUtils.listServiceRecords(registryOperations, servicePath);
                  for (String host : hosts.keySet()) {
                      ServiceRecord sr = hosts.get(JOYConstants.HOST);
                      String[] portList = sr.get(JOYConstants.PORT_LIST).split(JOYConstants.COMMA);
                      for (String usedport : portList) {
                          if (Integer.parseInt(usedport) == port)
                              return true;
                      }
                  }
              }
  
              if (registryOperations.exists(path)) {
                  ServiceRecord sr = registryOperations.resolve(path);
                  String[] portList = sr.get(JOYConstants.PORT_LIST).split(JOYConstants.COMMA);
  
                  String portListUpdate = join(portList, JOYConstants.COMMA, true) + String.valueOf(port);
                  sr.set(JOYConstants.PORT_LIST, portListUpdate);
                  registryOperations.bind(path, sr, BindFlags.OVERWRITE);
                  return false;
              } else {
                  registryOperations.mknode(path, true);
                  ServiceRecord sr = new ServiceRecord();
                  String portListUpdate = String.valueOf(port);
                  sr.set(JOYConstants.PORT_LIST, portListUpdate);
                  registryOperations.bind(path, sr, BindFlags.OVERWRITE);
                  return false;
              }
          } catch (Exception ex) {
              return true;
          }
      }
  
      public static List<String> getSupervisorPorts(int memory, int vcores, String instanceName, String supervisorHost, RegistryOperations registryOperations) {
          List<String> relist = new ArrayList<String>();
          int slotCount = getSlotCount(memory, vcores);
          for (int i = 9000; i < 15000; i++) {
              if (isPortAvailable(supervisorHost, i)) {
                  if (!getSetPortUsedBySupervisor(instanceName, supervisorHost, i, registryOperations))
                      relist.add(String.valueOf(i));
              }
              if (relist.size() >= slotCount) {
                  break;
              }
          }
          return relist;
      }
  
      public static int getSlotCount(int memory, int vcores) {
          int cpuports = (int) Math.ceil(vcores / 1.2);
          int memoryports = (int) Math.ceil(memory / 2000.0);
  //        return cpuports > memoryports ? memoryports : cpuports;
          //  doesn't support cgroup yet
          return memoryports;
      }
  
      public static Options initClientOptions() {
          Options opts = new Options();
          opts.addOption(JOYConstants.APP_NAME_KEY, true, "Application Name. Default value - com.alibaba.jstorm.yarn.JstormOnYarn");
          opts.addOption(JOYConstants.PRIORITY, true, "Application Priority. Default 0");
          opts.addOption(JOYConstants.PRIORITY, true, "RM Queue in which this application is to be submitted");
          opts.addOption(JOYConstants.TIMEOUT, true, "Application timeout in milliseconds");
          opts.addOption(JOYConstants.MASTER_MEMORY, true, "Amount of memory in MB to be requested to run the application master");
          opts.addOption(JOYConstants.MASTER_VCORES, true, "Amount of virtual cores to be requested to run the application master");
          opts.addOption(JOYConstants.JAR, true, "Jar file containing the application master");
          opts.addOption(JOYConstants.LIB_JAR, true, "dependency lib");
          opts.addOption(JOYConstants.HOME_DIR, true, "home ");
          opts.addOption(JOYConstants.CONF_FILE, true, " path of jstorm-yarn.xml ");
          opts.addOption(JOYConstants.RM_ADDRESS, true, "resource manager address eg hostname:port");
          opts.addOption(JOYConstants.NN_ADDRESS, true, "nameNode address eg hostname:port");
          opts.addOption(JOYConstants.HADOOP_CONF_DIR, true, "hadoop config directory which contains hdfs-site.xml/core-site" +
                  ".xml/yarn-site.xml");
          opts.addOption(JOYConstants.INSTANCE_NAME, true, "instance name , which is path of registry");
          opts.addOption(JOYConstants.DEPLOY_PATH, true, "deploy dir on HDFS  ");
          opts.addOption(JOYConstants.SHELL_SCRIPT, true, "Location of the shell script to be " +
                  "executed. Can only specify either --shell_command or --shell_script");
          opts.addOption(JOYConstants.SHELL_ARGS, true, "Command line args for the shell script." +
                  "Multiple args can be separated by empty space.");
          opts.getOption(JOYConstants.SHELL_ARGS).setArgs(Option.UNLIMITED_VALUES);
          opts.addOption(JOYConstants.SHELL_ENV, true, "Environment for shell script. Specified as env_key=env_val pairs");
          opts.addOption(JOYConstants.SHELL_CMD_PRIORITY, true, "Priority for the shell command containers");
          opts.addOption(JOYConstants.CONTAINER_MEMORY, true, "Amount of memory in MB to be requested to run the shell command");
          opts.addOption(JOYConstants.CONTAINER_VCORES, true, "Amount of virtual cores to be requested to run the shell command");
          opts.addOption(JOYConstants.NUM_CONTAINERS, true, "No. of containers on which the shell command needs to be executed");
          opts.addOption(JOYConstants.LOG_PROPERTIES, true, "log4j.properties file");
          opts.addOption(JOYConstants.KEEP_CONTAINERS_ACROSS_APPLICATION_ATTEMPTS, false,
                  "Flag to indicate whether to keep containers across application attempts." +
                          " If the flag is true, running containers will not be killed when" +
                          " application attempt fails and these containers will be retrieved by" +
                          " the new application attempt ");
          opts.addOption(JOYConstants.ATTEMPT_FAILURES_VALIDITY_INTERVAL, true,
                  "when attempt_failures_validity_interval in milliseconds is set to > 0," +
                          "the failure number will not take failures which happen out of " +
                          "the validityInterval into failure count. " +
                          "If failure count reaches to maxAppAttempts, " +
                          "the application will be failed.");
          opts.addOption(JOYConstants.DEBUG, false, "Dump out debug information");
          opts.addOption(JOYConstants.DOMAIN, true, "ID of the timeline domain where the "
                  + "timeline entities will be put");
          opts.addOption(JOYConstants.VIEW_ACLS, true, "Users and groups that allowed to "
                  + "view the timeline entities in the given domain");
          opts.addOption(JOYConstants.MODIFY_ACLS, true, "Users and groups that allowed to "
                  + "modify the timeline entities in the given domain");
          opts.addOption(JOYConstants.CREATE, false, "Flag to indicate whether to create the "
                  + "domain specified with -domain.");
          opts.addOption(JOYConstants.HELP, false, "Print usage");
          opts.addOption(JOYConstants.NODE_LABEL_EXPRESSION, true,
                  "Node label expression to determine the nodes"
                          + " where all the containers of this application"
                          + " will be allocated, \"\" means containers"
                          + " can be allocated anywhere, if you don't specify the option,"
                          + " default node_label_expression of queue will be used.");
          return opts;
      }
  
      public static void getYarnConfFromJar(String jarPath) {
          String confPath = jarPath + JOYConstants.CONF_NAME;
          try {
              InputStream stream = new FileInputStream(confPath);
              FileOutputStream out = new FileOutputStream(JOYConstants.CONF_NAME);
              byte[] data = IOUtils.toByteArray(stream);
              out.write(data);
              out.close();
          } catch (Exception e) {
              throw new IllegalArgumentException(
                      "No configuration file specified to be executed by application master to launch process");
          }
      }
  
      public static void checkAndSetOptions(CommandLine cliParser, JstormClientContext jstormClientContext) {
          if (cliParser.hasOption(JOYConstants.DEBUG)) {
              jstormClientContext.debugFlag = true;
          }
          if (cliParser.hasOption(JOYConstants.KEEP_CONTAINERS_ACROSS_APPLICATION_ATTEMPTS)) {
              jstormClientContext.keepContainers = true;
          }
          if (cliParser.hasOption(JOYConstants.LOG_PROPERTIES)) {
              String log4jPath = cliParser.getOptionValue(JOYConstants.LOG_PROPERTIES);
              try {
                  Log4jPropertyHelper.updateLog4jConfiguration(JstormOnYarn.class, log4jPath);
              } catch (Exception e) {
                  LOG.warn("Can not set up custom log4j properties. " + e);
              }
          }
          if (jstormClientContext.amMemory < 0) {
              throw new IllegalArgumentException("Invalid memory specified for application master, exiting."
                      + " Specified memory=" + jstormClientContext.amMemory);
          }
          if (jstormClientContext.amVCores < 0) {
              throw new IllegalArgumentException("Invalid virtual cores specified for application master, exiting."
                      + " Specified virtual cores=" + jstormClientContext.amVCores);
          }
          if (!cliParser.hasOption(JOYConstants.JAR)) {
              throw new IllegalArgumentException("No jar file specified for application master");
          }
          if (!jstormClientContext.rmHost.equals(JOYConstants.EMPTY)) {
              jstormClientContext.conf.set(JOYConstants.RM_ADDRESS_KEY, jstormClientContext.rmHost, JOYConstants.YARN_CONF_MODE);
          }
          if (!jstormClientContext.nameNodeHost.equals(JOYConstants.EMPTY)) {
              jstormClientContext.conf.set(JOYConstants.FS_DEFAULTFS_KEY, jstormClientContext.nameNodeHost);
          }
          if (!StringUtils.isBlank(jstormClientContext.hadoopConfDir)) {
              try {
                  Collection<File> files = FileUtils.listFiles(new File(jstormClientContext.hadoopConfDir), new String[]{JOYConstants.XML}, true);
                  for (File file : files) {
                      LOG.info("adding hadoop conf file to conf: " + file.getAbsolutePath());
                      jstormClientContext.conf.addResource(file.getAbsolutePath());
                  }
              } catch (Exception ex) {
                  LOG.error("failed to list hadoop conf dir: " + jstormClientContext.hadoopConfDir);
              }
          }
          String jarPath = JstormOnYarn.class.getProtectionDomain()
                  .getCodeSource().getLocation().getPath();
          if (jstormClientContext.confFile == null) {
              JstormYarnUtils.getYarnConfFromJar(jarPath);
              jstormClientContext.conf.addResource(JOYConstants.CONF_NAME);
          } else {
              Path jstormyarnConfPath = new Path(jstormClientContext.confFile);
              jstormClientContext.conf.addResource(jstormyarnConfPath);
          }
          if (!cliParser.hasOption(JOYConstants.SHELL_SCRIPT)) {
              String jarShellScriptPath = jarPath + JOYConstants.START_JSTORM_SHELL;
              try {
                  InputStream stream = new FileInputStream(jarShellScriptPath);
                  FileOutputStream out = new FileOutputStream(JOYConstants.START_JSTORM_SHELL);
                  out.write(IOUtils.toByteArray(stream));
                  out.close();
                  jstormClientContext.shellScriptPath = JOYConstants.START_JSTORM_SHELL;
              } catch (Exception e) {
                  throw new IllegalArgumentException(
                          "No shell script specified to be executed by application master to start nimbus and supervisor");
              }
          } else if (cliParser.hasOption(JOYConstants.SHELL_COMMAND) && cliParser.hasOption(JOYConstants.SHELL_SCRIPT)) {
              throw new IllegalArgumentException("Can not specify shell_command option " +
                      "and shell_script option at the same time");
          } else if (cliParser.hasOption(JOYConstants.SHELL_COMMAND)) {
              jstormClientContext.shellCommand = cliParser.getOptionValue(JOYConstants.SHELL_COMMAND);
          } else {
              jstormClientContext.shellScriptPath = cliParser.getOptionValue(JOYConstants.SHELL_SCRIPT);
          }
          if (cliParser.hasOption(JOYConstants.SHELL_ARGS)) {
              jstormClientContext.shellArgs = cliParser.getOptionValues(JOYConstants.SHELL_ARGS);
          }
          setShellEnv(cliParser, jstormClientContext);
          jstormClientContext.shellCmdPriority = Integer.parseInt(cliParser.getOptionValue(JOYConstants.SHELL_CMD_PRIORITY, JOYConstants.SHELL_CMD_PRIORITY_DEFAULT_VALUE));
          //set AM memory default to 1000mb
          jstormClientContext.containerMemory = Integer.parseInt(cliParser.getOptionValue(JOYConstants.CONTAINER_MEMORY, JOYConstants.DEFAULT_CONTAINER_MEMORY));
          jstormClientContext.containerVirtualCores = Integer.parseInt(cliParser.getOptionValue(JOYConstants.CONTAINER_VCORES, JOYConstants.DEFAULT_CONTAINER_VCORES));
          jstormClientContext.numContainers = Integer.parseInt(cliParser.getOptionValue(JOYConstants.NUM_CONTAINERS, JOYConstants.DEFAULT_NUM_CONTAINER));
  
          if (jstormClientContext.containerMemory < 0 || jstormClientContext.containerVirtualCores < 0 || jstormClientContext.numContainers < 1) {
              throw new IllegalArgumentException("Invalid no. of containers or container memory/vcores specified,"
                      + " exiting."
                      + " Specified containerMemory=" + jstormClientContext.containerMemory
                      + ", containerVirtualCores=" + jstormClientContext.containerVirtualCores
                      + ", numContainer=" + jstormClientContext.numContainers);
          }
  
          jstormClientContext.nodeLabelExpression = cliParser.getOptionValue(JOYConstants.NODE_LABEL_EXPRESSION, null);
          jstormClientContext.clientTimeout = Integer.parseInt(cliParser.getOptionValue(JOYConstants.TIMEOUT, JOYConstants.DEFAULT_CLIENT_TIME_OUT));
  
          jstormClientContext.attemptFailuresValidityInterval =
                  Long.parseLong(cliParser.getOptionValue(
                          JOYConstants.ATTEMPT_FAILURES_VALIDITY_INTERVAL, JOYConstants.DEFAULT_ATTEMPT_FAILURES_VALIDITY_INTERVAL));
  
          jstormClientContext.log4jPropFile = cliParser.getOptionValue(JOYConstants.LOG_PROPERTIES, JOYConstants.EMPTY);
  
          // Get timeline domain options
          if (cliParser.hasOption(JOYConstants.DOMAIN)) {
              jstormClientContext.domainId = cliParser.getOptionValue(JOYConstants.DOMAIN);
              jstormClientContext.toCreateDomain = cliParser.hasOption(JOYConstants.CREATE);
              if (cliParser.hasOption(JOYConstants.VIEW_ACLS)) {
                  jstormClientContext.viewACLs = cliParser.getOptionValue(JOYConstants.VIEW_ACLS);
              }
              if (cliParser.hasOption(JOYConstants.MODIFY_ACLS)) {
                  jstormClientContext.modifyACLs = cliParser.getOptionValue(JOYConstants.MODIFY_ACLS);
              }
          }
      }
  
      public static void setShellEnv(CommandLine cliParser, MasterContext jstormContext) {
          if (cliParser.hasOption(JOYConstants.SHELL_ENV)) {
              String envs[] = cliParser.getOptionValues(JOYConstants.SHELL_ENV);
              for (String env : envs) {
                  env = env.trim();
                  int index = env.indexOf(JOYConstants.EQUAL);
                  if (index == -1) {
                      jstormContext.getShellEnv().put(env, JOYConstants.EMPTY);
                      continue;
                  }
                  String key = env.substring(0, index);
                  String val = JOYConstants.EMPTY;
                  if (index < (env.length() - 1)) {
                      val = env.substring(index + 1);
                  }
                  jstormContext.getShellEnv().put(key, val);
              }
          }
      }
  
      private static boolean fileExist(String filePath) {
          return new File(filePath).exists();
      }
  
      /**
       * Dump out contents of $CWD and the environment to stdout for debugging
       */
      private static void dumpOutDebugInfo() {
  
          LOG.info("Dump debug output");
          Map<String, String> envs = System.getenv();
          for (Map.Entry<String, String> env : envs.entrySet()) {
              LOG.info("System env: key=" + env.getKey() + ", val=" + env.getValue());
              System.out.println("System env: key=" + env.getKey() + ", val="
                      + env.getValue());
          }
  
          BufferedReader buf = null;
          try {
              String lines = Shell.WINDOWS ? Shell.execCommand("cmd", "/c", "dir") :
                      Shell.execCommand("ls", "-al");
              buf = new BufferedReader(new StringReader(lines));
              String line = "";
              while ((line = buf.readLine()) != null) {
                  LOG.info("System CWD content: " + line);
                  System.out.println("System CWD content: " + line);
              }
          } catch (IOException e) {
              e.printStackTrace();
          } finally {
              org.apache.hadoop.io.IOUtils.cleanup(LOG, buf);
          }
      }
  
      private static String readContent(String filePath) throws IOException {
          DataInputStream ds = null;
          try {
              ds = new DataInputStream(new FileInputStream(filePath));
              return ds.readUTF();
          } finally {
              org.apache.commons.io.IOUtils.closeQuietly(ds);
          }
      }
  
      public static void checkAndSetMasterOptions(CommandLine cliParser, JstormMasterContext jstormMasterContext, Configuration conf) throws Exception {
  
          //Check whether customer log4j.properties file exists
          if (fileExist(log4jPath)) {
              try {
                  Log4jPropertyHelper.updateLog4jConfiguration(JstormMaster.class,
                          log4jPath);
              } catch (Exception e) {
                  LOG.warn("Can not set up custom log4j properties. " + e);
              }
          }
  
          if (cliParser.hasOption(JOYConstants.DEBUG)) {
              dumpOutDebugInfo();
          }
  
          Map<String, String> envs = System.getenv();
  
          if (!envs.containsKey(ApplicationConstants.Environment.CONTAINER_ID.name())) {
              if (cliParser.hasOption(JOYConstants.APP_ATTEMPT_ID)) {
                  String appIdStr = cliParser.getOptionValue(JOYConstants.APP_ATTEMPT_ID, JOYConstants.EMPTY);
                  jstormMasterContext.appAttemptID = ConverterUtils.toApplicationAttemptId(appIdStr);
              } else {
                  throw new IllegalArgumentException(
                          "Application Attempt Id not set in the environment");
              }
          } else {
              ContainerId containerId = ConverterUtils.toContainerId(envs
                      .get(ApplicationConstants.Environment.CONTAINER_ID.name()));
              jstormMasterContext.appAttemptID = containerId.getApplicationAttemptId();
          }
  
          if (!envs.containsKey(ApplicationConstants.APP_SUBMIT_TIME_ENV)) {
              throw new RuntimeException(ApplicationConstants.APP_SUBMIT_TIME_ENV
                      + " not set in the environment");
          }
          if (!envs.containsKey(ApplicationConstants.Environment.NM_HOST.name())) {
              throw new RuntimeException(ApplicationConstants.Environment.NM_HOST.name()
                      + " not set in the environment");
          }
          if (!envs.containsKey(ApplicationConstants.Environment.NM_HTTP_PORT.name())) {
              throw new RuntimeException(ApplicationConstants.Environment.NM_HTTP_PORT
                      + " not set in the environment");
          }
          if (!envs.containsKey(ApplicationConstants.Environment.NM_PORT.name())) {
              throw new RuntimeException(ApplicationConstants.Environment.NM_PORT.name()
                      + " not set in the environment");
          }
  
          LOG.info("Application master for app" + ", appId="
                  + jstormMasterContext.appAttemptID.getApplicationId().getId() + ", clustertimestamp="
                  + jstormMasterContext.appAttemptID.getApplicationId().getClusterTimestamp()
                  + ", attemptId=" + jstormMasterContext.appAttemptID.getAttemptId());
  
          if (!fileExist(shellCommandPath)
                  && envs.get(JOYConstants.DISTRIBUTEDSHELLSCRIPTLOCATION).isEmpty()) {
              throw new IllegalArgumentException(
                      "No shell command or shell script specified to be executed by application master");
          }
  
          if (fileExist(shellCommandPath)) {
              jstormMasterContext.shellCommand = readContent(shellCommandPath);
          }
  
          if (fileExist(shellArgsPath)) {
              jstormMasterContext.shellArgs = readContent(shellArgsPath);
          }
  
          JstormYarnUtils.setShellEnv(cliParser, jstormMasterContext);
  
          if (envs.containsKey(JOYConstants.DISTRIBUTEDSHELLSCRIPTLOCATION)) {
              jstormMasterContext.scriptPath = envs.get(JOYConstants.DISTRIBUTEDSHELLSCRIPTLOCATION);
  
              jstormMasterContext.appMasterJarPath = envs.get(JOYConstants.APPMASTERJARSCRIPTLOCATION);
              if (envs.containsKey(JOYConstants.DISTRIBUTEDSHELLSCRIPTTIMESTAMP)) {
                  jstormMasterContext.shellScriptPathTimestamp = Long.parseLong(envs
                          .get(JOYConstants.DISTRIBUTEDSHELLSCRIPTTIMESTAMP));
                  jstormMasterContext.jarTimestamp = Long.parseLong(envs
                          .get(JOYConstants.APPMASTERTIMESTAMP));
              }
              if (envs.containsKey(JOYConstants.DISTRIBUTEDSHELLSCRIPTLEN)) {
                  jstormMasterContext.shellScriptPathLen = Long.parseLong(envs
                          .get(JOYConstants.DISTRIBUTEDSHELLSCRIPTLEN));
                  jstormMasterContext.jarPathLen = Long.parseLong(envs
                          .get(JOYConstants.APPMASTERLEN));
              }
  
              if (!jstormMasterContext.scriptPath.isEmpty()
                      && (jstormMasterContext.shellScriptPathTimestamp <= 0 || jstormMasterContext.shellScriptPathLen <= 0)) {
                  LOG.error("Illegal values in env for shell script path" + ", path="
                          + jstormMasterContext.scriptPath + ", len=" + jstormMasterContext.shellScriptPathLen + ", timestamp="
                          + jstormMasterContext.shellScriptPathTimestamp);
                  throw new IllegalArgumentException(
                          "Illegal values in env for shell script path");
              }
          }
  
          if (envs.containsKey(JOYConstants.DISTRIBUTEDSHELLTIMELINEDOMAIN)) {
              jstormMasterContext.domainId = envs.get(JOYConstants.DISTRIBUTEDSHELLTIMELINEDOMAIN);
          }
  
          if (envs.containsKey(JOYConstants.BINARYFILEDEPLOYPATH)
                  && !envs.get(JOYConstants.BINARYFILEDEPLOYPATH).equals(JOYConstants.EMPTY)) {
              conf.set(JOYConstants.INSTANCE_DEPLOY_DIR_KEY, envs.get(JOYConstants.BINARYFILEDEPLOYPATH));
              jstormMasterContext.deployPath = envs.get(JOYConstants.BINARYFILEDEPLOYPATH);
          }
  
          if (envs.containsKey(JOYConstants.INSTANCENAME)
                  && !envs.get(JOYConstants.INSTANCENAME).equals(JOYConstants.EMPTY)) {
              conf.set(JOYConstants.INSTANCE_NAME_KEY, envs.get(JOYConstants.INSTANCENAME));
              jstormMasterContext.instanceName = envs.get(JOYConstants.INSTANCENAME);
          }
          jstormMasterContext.containerVirtualCores = Integer.parseInt(cliParser.getOptionValue(
                  JOYConstants.CONTAINER_VCORES, JOYConstants.DEFAULT_CONTAINER_VCORES));
          jstormMasterContext.numTotalContainers = Integer.parseInt(cliParser.getOptionValue(
                  JOYConstants.NUM_CONTAINERS, JOYConstants.DEFAULT_NUM_CONTAINER));
          if (jstormMasterContext.numTotalContainers == 0) {
              throw new IllegalArgumentException(
                      "Cannot run distributed shell with no containers");
          }
      }
  }
  