@SuppressWarnings("deprecation")
    private boolean run() throws IOException, NoSuchAlgorithmException {
        LOG.info("ApplicationMaster Starting ...");
        regResp = registerApplicationMaster();

        LOG.info("Application submit hbox client is: "
                + conf.get(HboxConfiguration.HBOX_CLIENT_HOSTNAME, HboxConfiguration.DEFAULT_HBOX_CLIENT_HOSTNAME));
        LOG.info("HBox release version: " + HboxVersion.VERSION);

        final Span span = Span.current();
        if (conf.getBoolean(HboxConfiguration.HBOX_AM_CMD_ENABLE, HboxConfiguration.DEFAULT_HBOX_AM_ENABLE)) {
            String cmd = conf.get(HboxConfiguration.HBOX_AM_CMD, HboxConfiguration.DEDAULT_HBOX_AM_CMD);
            if (cmd != null && !cmd.equals("")) {
                LOG.info("ApplicationMaster enable extra command: " + cmd);
                final Process extraAmCmdProcess = Runtime.getRuntime().exec(cmd);
                Thread stdout = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            BufferedReader reader;
                            reader = new BufferedReader(new InputStreamReader(extraAmCmdProcess.getInputStream()));
                            String extraAmCmdStdoutLog;
                            while ((extraAmCmdStdoutLog = reader.readLine()) != null) {
                                LOG.info(extraAmCmdStdoutLog);
                                if (conf.getBoolean(
                                        HboxConfiguration.HBOX_CONTAINER_RUNNING_LOG_ENABLE,
                                        HboxConfiguration.DEFAULT_HBOX_CONTAINER_RUNNING_LOG_ENABLE)) {
                                    amContainerStdOut.append(extraAmCmdStdoutLog);
                                }
                            }
                        } catch (Exception e) {
                            LOG.warn("Exception in thread extraAmCmdProcess stdout Thread");
                            e.printStackTrace();
                        }
                    }
                });
                stdout.start();

                Thread stderr = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            BufferedReader reader;
                            reader = new BufferedReader(new InputStreamReader(extraAmCmdProcess.getInputStream()));
                            String extraAmCmdStdoutLog;
                            while ((extraAmCmdStdoutLog = reader.readLine()) != null) {
                                LOG.info(extraAmCmdStdoutLog);
                                if (conf.getBoolean(
                                        HboxConfiguration.HBOX_CONTAINER_RUNNING_LOG_ENABLE,
                                        HboxConfiguration.DEFAULT_HBOX_CONTAINER_RUNNING_LOG_ENABLE)) {
                                    amContainerStdErr.append(extraAmCmdStdoutLog);
                                }
                            }
                        } catch (Exception e) {
                            LOG.warn("Exception in thread extraAmCmdProcess stderr Thread");
                            e.printStackTrace();
                        }
                    }
                });
                stderr.start();
            }
        }

        if (conf.getBoolean(HboxConfiguration.HBOX_INPUT_STREAM, HboxConfiguration.DEFAULT_HBOX_INPUT_STREAM)
                || conf.get(HboxConfiguration.HBOX_INPUT_STRATEGY, HboxConfiguration.DEFAULT_HBOX_INPUT_STRATEGY)
                        .equals("STREAM")) {
            buildInputStreamFileStatus();
        } else {
            buildInputFileStatus();
            buildS3InputFileInfo();
        }

        if ("TENSORFLOW".equals(hboxAppType)
                || "TENSOR2TENSOR".equals(hboxAppType)
                || "MXNET".equals(hboxAppType)
                || "DISTLIGHTLDA".equals(hboxAppType)
                || "XFLOW".equals(hboxAppType)) {
            this.appendMessage(
                    "Hbox application needs " + workerNum + " worker and " + psNum + " ps containers in fact", true);
        } else {
            this.appendMessage("Hbox application needs " + workerNum + " worker container in fact", true);
        }

        buildContainerRequest(hostLocals);

        int requestWorkerNum = workerNum;
        if ("TENSORFLOW".equals(hboxAppType) || "TENSOR2TENSOR".equals(hboxAppType)) {
            if (chiefWorker) requestWorkerNum--;
            if (tfEvaluator) requestWorkerNum--;
        }

        rmCallbackHandler.setNeededPsContainersCount(psNum);
        rmCallbackHandler.setNeededWorkerContainersCount(requestWorkerNum);
        rmCallbackHandler.setHboxAppType(hboxAppType);

        int allocateInterval =
                conf.getInt(HboxConfiguration.HBOX_ALLOCATE_INTERVAL, HboxConfiguration.DEFAULT_HBOX_ALLOCATE_INTERVAL);
        amrmAsync.setHeartbeatInterval(allocateInterval);

        if (conf.get(HboxConfiguration.HBOX_HOST_BLACKLIST) != null) {
            String[] blacklist = conf.getStrings(HboxConfiguration.HBOX_HOST_BLACKLIST);
            List<String> hostBlackList = new ArrayList<>(blacklist.length);
            for (String host : blacklist) {
                if (host != null || host != "") {
                    hostBlackList.add(host);
                }
            }
            amrmSync.updateBlacklist(hostBlackList, null);
        }

        for (int i = 0; i < psNum; i++) {
            amrmAsync.addContainerRequest(psContainerRequest);
        }

        if (psNum > 0) {
            LOG.info("Try to allocate " + psNum + " ps containers");
        }

        Boolean startAllocatedContainer = false;
        Long startAllocatedTimeStamp = Long.MIN_VALUE;
        int psCancelCount = 0;
        while (rmCallbackHandler.getAllocatedPsContainerNumber() < psNum) {
            List<Container> cancelContainers = rmCallbackHandler.getCancelContainer();
            List<String> blackHosts = rmCallbackHandler.getBlackHosts();
            amrmSync.updateBlacklist(blackHosts, null);
            synchronized (cancelContainers) {
                if (cancelContainers.size() != 0) {
                    for (Container container : cancelContainers) {
                        LOG.info("Canceling container: " + container.getId().toString());
                        amrmAsync.releaseAssignedContainer(container.getId());
                        amrmAsync.addContainerRequest(psContainerRequest);
                        psCancelCount++;
                    }
                    cancelContainers.clear();
                }
            }
            if (rmCallbackHandler.getAllocatedPsContainerNumber() > 0 && !startAllocatedContainer) {
                startAllocatedContainer = true;
                startAllocatedTimeStamp = System.currentTimeMillis();
            }
            if (startAllocatedContainer
                    && (System.currentTimeMillis() - startAllocatedTimeStamp)
                            > conf.getInt(
                                    YarnConfiguration.RM_CONTAINER_ALLOC_EXPIRY_INTERVAL_MS,
                                    YarnConfiguration.DEFAULT_RM_CONTAINER_ALLOC_EXPIRY_INTERVAL_MS)) {
                String failMessage =
                        "Container waiting except the allocated expiry time. Maybe the Cluster resource not satisfied the user necessary. Please resubmit !";
                LOG.info(failMessage);
                this.appendMessage("Unregister  Application", true);
                unregisterApp(FinalApplicationStatus.FAILED, failMessage);
                return false;
            }
            Utilities.sleep(allocateInterval);
        }

        acquiredPsContainers = rmCallbackHandler.getAcquiredPsContainer();
        if (psNum > 0) {
            int totalNumAllocatedPs = rmCallbackHandler.getAllocatedPsContainerNumber();
            if (totalNumAllocatedPs > psNum) {
                while (acquiredPsContainers.size() > psNum) {
                    Container releaseContainer = acquiredPsContainers.remove(acquiredPsContainers.size() - 1);
                    amrmAsync.releaseAssignedContainer(releaseContainer.getId());
                    LOG.info("Release container " + releaseContainer.getId().toString());
                }
            }
            LOG.info("Total " + acquiredPsContainers.size() + " ps containers has allocated.");
        }

        rmCallbackHandler.setWorkerContainersAllocating();

        for (int i = 0; i < requestWorkerNum; i++) {
            amrmAsync.addContainerRequest(workerContainerRequest);
        }

        LOG.info("Try to allocate " + requestWorkerNum + " worker containers");
        int workerCancelCount = 0;
        while (rmCallbackHandler.getAllocatedWorkerContainerNumber() < requestWorkerNum) {
            if (hboxAppType.equals("MPI") || hboxAppType.equals("TENSORNET") || hboxAppType.equals("HOROVOD")) {
                rmCallbackHandler.addBlackHost(applicationMasterHostname);
                List<String> blackAMs = rmCallbackHandler.getBlackHosts();
                amrmSync.updateBlacklist(blackAMs, null);
            }
            List<Container> cancelContainers = rmCallbackHandler.getCancelContainer();
            List<String> blackHosts = rmCallbackHandler.getBlackHosts();
            amrmSync.updateBlacklist(blackHosts, null);
            synchronized (cancelContainers) {
                if (cancelContainers.size() != 0) {
                    for (Container container : cancelContainers) {
                        LOG.info("Canceling container: " + container.getId().toString());
                        amrmAsync.releaseAssignedContainer(container.getId());
                        amrmAsync.addContainerRequest(workerContainerRequest);
                        workerCancelCount++;
                    }
                    cancelContainers.clear();
                }
            }
            if (rmCallbackHandler.getAllocatedWorkerContainerNumber() > 0 && !startAllocatedContainer) {
                startAllocatedContainer = true;
                startAllocatedTimeStamp = System.currentTimeMillis();
            }
            if (startAllocatedContainer
                    && (System.currentTimeMillis() - startAllocatedTimeStamp)
                            > conf.getInt(
                                    YarnConfiguration.RM_CONTAINER_ALLOC_EXPIRY_INTERVAL_MS,
                                    YarnConfiguration.DEFAULT_RM_CONTAINER_ALLOC_EXPIRY_INTERVAL_MS)) {
                String failMessage =
                        "Container waiting except the allocated expiry time. Maybe the Cluster resource not satisfied the user necessary. Please resubmit !";
                LOG.info(failMessage);
                this.appendMessage("Unregister  Application", true);
                unregisterApp(FinalApplicationStatus.FAILED, failMessage);
                return false;
            }
            Utilities.sleep(allocateInterval);
        }

        acquiredWorkerContainers = rmCallbackHandler.getAcquiredWorkerContainer();
        int totalNumAllocatedWorkers = rmCallbackHandler.getAllocatedWorkerContainerNumber();
        if (totalNumAllocatedWorkers > requestWorkerNum) {
            while (acquiredWorkerContainers.size() > requestWorkerNum) {
                Container releaseContainer = acquiredWorkerContainers.remove(0);
                amrmAsync.releaseAssignedContainer(releaseContainer.getId());
                LOG.info("Release container " + releaseContainer.getId().toString());
            }
        }
        LOG.info("Total " + acquiredWorkerContainers.size() + " worker containers has allocated.");

        for (int i = 0; i < psNum + psCancelCount; i++) {
            amrmAsync.removeContainerRequest(psContainerRequest);
        }
        for (int i = 0; i < requestWorkerNum + workerCancelCount; i++) {
            amrmAsync.removeContainerRequest(workerContainerRequest);
        }

        List<Container> cancelContainersTotal = rmCallbackHandler.getCancelContainer();
        synchronized (cancelContainersTotal) {
            if (cancelContainersTotal.size() != 0) {
                for (Container container : cancelContainersTotal) {
                    LOG.info("Canceling container: " + container.getId().toString());
                    amrmAsync.releaseAssignedContainer(container.getId());
                }
                cancelContainersTotal.clear();
            }
        }

        if (requestWorkerNum != workerNum) {
            int chiefWorkerCancelCount = 0;
            int evaluatorCancelCount = 0;
            if (chiefWorker) {
                LOG.info("Try to allocate chief worker containers");
                rmCallbackHandler.setChiefWorkerContainersAllocating();
                amrmAsync.addContainerRequest(chiefWorkerContainerRequest);
                chiefWorkerCancelCount++;
                while (rmCallbackHandler.getAcquiredChiefWorkerContainers().size() < 1) {
                    List<Container> cancelContainers = rmCallbackHandler.getCancelContainer();
                    List<String> blackHosts = rmCallbackHandler.getBlackHosts();
                    amrmSync.updateBlacklist(blackHosts, null);
                    synchronized (cancelContainers) {
                        if (cancelContainers.size() != 0) {
                            for (Container container : cancelContainers) {
                                LOG.info("Canceling container: "
                                        + container.getId().toString());
                                amrmAsync.releaseAssignedContainer(container.getId());
                                amrmAsync.addContainerRequest(chiefWorkerContainerRequest);
                                chiefWorkerCancelCount++;
                            }
                            cancelContainers.clear();
                        }
                    }
                    if (startAllocatedContainer
                            && (System.currentTimeMillis() - startAllocatedTimeStamp)
                                    > conf.getInt(
                                            YarnConfiguration.RM_CONTAINER_ALLOC_EXPIRY_INTERVAL_MS,
                                            YarnConfiguration.DEFAULT_RM_CONTAINER_ALLOC_EXPIRY_INTERVAL_MS)) {
                        String failMessage =
                                "Container waiting except the allocated expiry time. Maybe the Cluster resource not satisfied the user necessary. Please resubmit !";
                        LOG.info(failMessage);
                        this.appendMessage("Unregister  Application", true);
                        unregisterApp(FinalApplicationStatus.FAILED, failMessage);
                        return false;
                    }
                    Utilities.sleep(allocateInterval);
                }

                acquiredChiefWorkerContainers = rmCallbackHandler.getAcquiredChiefWorkerContainers();
                synchronized (acquiredChiefWorkerContainers) {
                    if (acquiredChiefWorkerContainers.size() > 1) {
                        while (acquiredChiefWorkerContainers.size() > 1) {
                            Container releaseContainer = acquiredChiefWorkerContainers.remove(0);
                            amrmAsync.releaseAssignedContainer(releaseContainer.getId());
                            LOG.info("Release chief container "
                                    + releaseContainer.getId().toString());
                        }
                    }
                }
                LOG.info("Total " + acquiredChiefWorkerContainers.size() + " chief worker containers has allocated.");
                acquiredWorkerContainers.add(0, acquiredChiefWorkerContainers.get(0));
            }

            if (tfEvaluator) {
                LOG.info("Try to allocate evaluator worker containers");
                rmCallbackHandler.setEvaluatorWorkerContainersAllocating();
                amrmAsync.addContainerRequest(evaluatorWorkerContainerRequest);
                LOG.info("evaluatorWorkerContainerRequest:" + evaluatorWorkerContainerRequest);
                evaluatorCancelCount++;
                while (rmCallbackHandler.getAcquiredEvaluatorWorkerContainers().size() < 1) {
                    LOG.info("start request evaluator");
                    List<Container> cancelContainers = rmCallbackHandler.getCancelContainer();
                    List<String> blackHosts = rmCallbackHandler.getBlackHosts();
                    amrmSync.updateBlacklist(blackHosts, null);
                    synchronized (cancelContainers) {
                        if (cancelContainers.size() != 0) {
                            LOG.info("cancelContainers");
                            for (Container container : cancelContainers) {
                                LOG.info("Canceling container: "
                                        + container.getId().toString());
                                amrmAsync.releaseAssignedContainer(container.getId());
                                amrmAsync.addContainerRequest(evaluatorWorkerContainerRequest);
                                evaluatorCancelCount++;
                            }
                            cancelContainers.clear();
                        }
                    }
                    if (startAllocatedContainer
                            && (System.currentTimeMillis() - startAllocatedTimeStamp)
                                    > conf.getInt(
                                            YarnConfiguration.RM_CONTAINER_ALLOC_EXPIRY_INTERVAL_MS,
                                            YarnConfiguration.DEFAULT_RM_CONTAINER_ALLOC_EXPIRY_INTERVAL_MS)) {
                        String failMessage =
                                "Container waiting except the allocated expiry time. Maybe the Cluster resource not satisfied the user necessary. Please resubmit !";
                        LOG.info(failMessage);
                        this.appendMessage("Unregister  Application", true);
                        unregisterApp(FinalApplicationStatus.FAILED, failMessage);
                        return false;
                    }
                    Utilities.sleep(allocateInterval);
                }

                LOG.info("release extra evaluator");
                acquiredEvaluatorWorkerContainers = rmCallbackHandler.getAcquiredEvaluatorWorkerContainers();
                synchronized (acquiredEvaluatorWorkerContainers) {
                    if (acquiredEvaluatorWorkerContainers.size() > 1) {
                        LOG.info("start release");
                        while (acquiredEvaluatorWorkerContainers.size() > 1) {
                            Container releaseContainer = acquiredEvaluatorWorkerContainers.remove(0);
                            amrmAsync.releaseAssignedContainer(releaseContainer.getId());
                            LOG.info("Release evaluator container "
                                    + releaseContainer.getId().toString());
                        }
                    }
                }
                LOG.info("Total " + acquiredEvaluatorWorkerContainers.size()
                        + " evaluator worker containers has allocated.");
                acquiredWorkerContainers.add(acquiredEvaluatorWorkerContainers.get(0));
            }

            for (int i = 0; i < chiefWorkerCancelCount; i++) {
                LOG.info("remove request chief container.");
                amrmAsync.removeContainerRequest(chiefWorkerContainerRequest);
            }
            for (int i = 0; i < evaluatorCancelCount; i++) {
                LOG.info("remove request evaluator container.");
                amrmAsync.removeContainerRequest(evaluatorWorkerContainerRequest);
            }
        }

        List<Container> cancelContainers = rmCallbackHandler.getCancelContainer();
        synchronized (cancelContainers) {
            if (cancelContainers.size() != 0) {
                for (Container container : cancelContainers) {
                    LOG.info("Canceling unnecessary container: "
                            + container.getId().toString());
                    amrmAsync.releaseAssignedContainer(container.getId());
                }
                cancelContainers.clear();
            }
        }

        if (conf.getBoolean(
                HboxConfiguration.HBOX_HOST_LOCAL_ENABLE, HboxConfiguration.DEFAULT_HBOX_HOST_LOCAL_ENABLE)) {
            containerHostnames = new HashSet<>();
            if (acquiredPsContainers.size() > 0) {
                for (Container container : acquiredPsContainers) {
                    containerHostnames.add(container.getNodeId().getHost());
                }
            }
            if (acquiredWorkerContainers.size() > 0) {
                for (Container container : acquiredWorkerContainers) {
                    containerHostnames.add(container.getNodeId().getHost());
                }
            }
            LOG.info("host local enable is true, host list is: " + containerHostnames.toString());
        }

        // launch dist mxnet scheduler
        if (hboxAppType.equals("MXNET") && !singleMx) {
            LOG.info("Seting environments for the mxnet scheduler");
            dmlcPsRootUri = applicationMasterHostname;
            Socket schedulerReservedSocket = new Socket();
            try {
                Utilities.getReservePort(
                        schedulerReservedSocket,
                        InetAddress.getByName(applicationMasterHostname).getHostAddress(),
                        reservePortBegin,
                        reservePortEnd);
            } catch (IOException e) {
                LOG.error("Can not get available port");
            }
            dmlcPsRootPort = schedulerReservedSocket.getLocalPort();

            List<String> schedulerEnv = new ArrayList<>(20);
            Map<String, String> userEnv = new HashMap<>();
            if (conf.get(HboxConfiguration.HBOX_CONTAINER_ENV) != null) {
                String[] env = StringUtils.split(conf.get(HboxConfiguration.HBOX_CONTAINER_ENV), "|");
                for (String envPair : env) {
                    String[] userEnvPair = StringUtils.split(envPair, "=");
                    if (userEnvPair.length != 2) {
                        LOG.error(envPair + " is not correct");
                    } else {
                        schedulerEnv.add(envPair);
                        userEnv.put(userEnvPair[0], userEnvPair[1]);
                    }
                }
            }
            if (userEnv.containsKey("PATH")) {
                schedulerEnv.add(
                        "PATH=" + userEnv.get("PATH") + System.getProperty("path.separator") + System.getenv("PATH"));
            } else {
                schedulerEnv.add("PATH=" + System.getenv("PATH"));
            }
            schedulerEnv.add("JAVA_HOME=" + System.getenv("JAVA_HOME"));
            schedulerEnv.add("HADOOP_HOME=" + System.getenv("HADOOP_HOME"));
            schedulerEnv.add("HADOOP_HDFS_HOME=" + System.getenv("HADOOP_HDFS_HOME"));
            if (userEnv.containsKey("LD_LIBRARY_PATH")) {
                schedulerEnv.add("LD_LIBRARY_PATH=" + "./:" + userEnv.get("LD_LIBRARY_PATH")
                        + System.getProperty("path.separator") + System.getenv("LD_LIBRARY_PATH") + ":"
                        + System.getenv("JAVA_HOME") + "/jre/lib/amd64/server:" + System.getenv("HADOOP_HOME")
                        + "/lib/native");
            } else {
                schedulerEnv.add(
                        "LD_LIBRARY_PATH=" + "./:" + System.getenv("LD_LIBRARY_PATH") + ":" + System.getenv("JAVA_HOME")
                                + "/jre/lib/amd64/server:" + System.getenv("HADOOP_HOME") + "/lib/native");
            }
            if (userEnv.containsKey("CLASSPATH")) {
                schedulerEnv.add("CLASSPATH=" + "./:" + userEnv.get("CLASSPATH") + System.getProperty("path.separator")
                        + System.getenv("CLASSPATH") + ":" + System.getProperty("java.class.path"));
            } else {
                schedulerEnv.add("CLASSPATH=" + "./:" + System.getenv("CLASSPATH") + ":"
                        + System.getProperty("java.class.path"));
            }
            schedulerEnv.add("DMLC_ROLE=scheduler");
            schedulerEnv.add("DMLC_PS_ROOT_URI=" + dmlcPsRootUri);
            schedulerEnv.add("DMLC_PS_ROOT_PORT=" + dmlcPsRootPort);
            schedulerEnv.add(HboxConstants.Environment.HBOX_DMLC_WORKER_NUM.toString() + "=" + workerNum);
            schedulerEnv.add(HboxConstants.Environment.HBOX_DMLC_SERVER_NUM.toString() + "=" + psNum);
            schedulerEnv.add("PYTHONUNBUFFERED=1");

            LOG.info("Executing command:" + String.join(" ", hboxCommandArgs));
            LOG.info("DMLC_PS_ROOT_URI is " + dmlcPsRootUri);
            LOG.info("DMLC_PS_ROOT_PORT is " + dmlcPsRootPort);
            LOG.info(HboxConstants.Environment.HBOX_DMLC_WORKER_NUM.toString() + "=" + workerNum);
            LOG.info(HboxConstants.Environment.HBOX_DMLC_SERVER_NUM.toString() + "=" + psNum);

            try {
                schedulerReservedSocket.close();
                final Process mxnetSchedulerProcess = processLaunch.exec(
                        hboxCommandArgs, schedulerEnv.toArray(new String[schedulerEnv.size()]), this.envs, null);
                LOG.info("Starting thread to redirect stdout of mxnet scheduler process");
                Thread mxnetSchedulerRedirectThread = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            BufferedReader reader;
                            reader = new BufferedReader(new InputStreamReader(mxnetSchedulerProcess.getInputStream()));
                            String mxnetSchedulerStdoutLog;
                            while ((mxnetSchedulerStdoutLog = reader.readLine()) != null) {
                                LOG.debug(mxnetSchedulerStdoutLog);
                                if (conf.getBoolean(
                                        HboxConfiguration.HBOX_CONTAINER_RUNNING_LOG_ENABLE,
                                        HboxConfiguration.DEFAULT_HBOX_CONTAINER_RUNNING_LOG_ENABLE)) {
                                    amContainerStdOut.append(mxnetSchedulerStdoutLog);
                                }
                            }
                        } catch (Exception e) {
                            LOG.warn("Exception in thread mxnetSchedulerRedirectThread");
                            e.printStackTrace();
                        }
                    }
                });
                mxnetSchedulerRedirectThread.start();

                LOG.info("Starting thread to redirect stderr of mxnet scheduler process");
                Thread boardStderrRedirectThread = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            BufferedReader reader;
                            reader = new BufferedReader(new InputStreamReader(mxnetSchedulerProcess.getErrorStream()));
                            String mxnetSchedulerStderrLog;
                            while ((mxnetSchedulerStderrLog = reader.readLine()) != null) {
                                LOG.info(mxnetSchedulerStderrLog);
                                if (conf.getBoolean(
                                        HboxConfiguration.HBOX_CONTAINER_RUNNING_LOG_ENABLE,
                                        HboxConfiguration.DEFAULT_HBOX_CONTAINER_RUNNING_LOG_ENABLE)) {
                                    amContainerStdErr.append(mxnetSchedulerStderrLog);
                                }
                            }
                        } catch (Exception e) {
                            LOG.warn("Error in thread mxnetSchedulerStderrRedirectThread");
                            e.printStackTrace();
                        }
                    }
                });
                boardStderrRedirectThread.start();
            } catch (Exception e) {
                LOG.info("start mxnet scheduler error " + e);
            }
        }

        // launch dist xgboost scheduler
        if (hboxAppType.equals("DISTXGBOOST")) {
            LOG.info("Seting environments for the dist xgboost scheduler");
            dmlcTrackerUri = applicationMasterHostname;
            Socket schedulerReservedSocket = new Socket();
            try {
                Utilities.getReservePort(
                        schedulerReservedSocket,
                        InetAddress.getByName(applicationMasterHostname).getHostAddress(),
                        reservePortBegin,
                        reservePortEnd);
            } catch (IOException e) {
                LOG.error("Can not get available port");
            }
            dmlcTrackerPort = schedulerReservedSocket.getLocalPort();

            List<String> schedulerEnv = new ArrayList<>(20);
            Map<String, String> userEnv = new HashMap<>();
            if (conf.get(HboxConfiguration.HBOX_CONTAINER_ENV) != null) {
                String[] env = StringUtils.split(conf.get(HboxConfiguration.HBOX_CONTAINER_ENV), "|");
                for (String envPair : env) {
                    String[] userEnvPair = StringUtils.split(envPair, "=");
                    if (userEnvPair.length != 2) {
                        LOG.error(envPair + " is not correct");
                    } else {
                        schedulerEnv.add(envPair);
                        userEnv.put(userEnvPair[0], userEnvPair[1]);
                    }
                }
            }
            if (userEnv.containsKey("PATH")) {
                schedulerEnv.add(
                        "PATH=" + userEnv.get("PATH") + System.getProperty("path.separator") + System.getenv("PATH"));
            } else {
                schedulerEnv.add("PATH=" + System.getenv("PATH"));
            }
            schedulerEnv.add("JAVA_HOME=" + System.getenv("JAVA_HOME"));
            schedulerEnv.add("HADOOP_HOME=" + System.getenv("HADOOP_HOME"));
            schedulerEnv.add("HADOOP_HDFS_HOME=" + System.getenv("HADOOP_HDFS_HOME"));
            if (userEnv.containsKey("LD_LIBRARY_PATH")) {
                schedulerEnv.add("LD_LIBRARY_PATH=" + "./:" + userEnv.get("LD_LIBRARY_PATH")
                        + System.getProperty("path.separator") + System.getenv("LD_LIBRARY_PATH") + ":"
                        + System.getenv("JAVA_HOME") + "/jre/lib/amd64/server:" + System.getenv("HADOOP_HOME")
                        + "/lib/native");
            } else {
                schedulerEnv.add(
                        "LD_LIBRARY_PATH=" + "./:" + System.getenv("LD_LIBRARY_PATH") + ":" + System.getenv("JAVA_HOME")
                                + "/jre/lib/amd64/server:" + System.getenv("HADOOP_HOME") + "/lib/native");
            }
            if (userEnv.containsKey("CLASSPATH")) {
                schedulerEnv.add("CLASSPATH=" + "./:" + userEnv.get("CLASSPATH") + System.getProperty("path.separator")
                        + System.getenv("CLASSPATH") + ":" + System.getProperty("java.class.path"));
            } else {
                schedulerEnv.add("CLASSPATH=" + "./:" + System.getenv("CLASSPATH") + ":"
                        + System.getProperty("java.class.path"));
            }
            schedulerEnv.add("PYTHONUNBUFFERED=1");

            String distXgboostSchedulerCmd = "python xgboost/self-define/rabitTracker.py --num-workers=" + workerNum
                    + " --host-ip=" + dmlcTrackerUri + " --port=" + dmlcTrackerPort;
            LOG.info("Dist xgboost scheduler executing command:" + distXgboostSchedulerCmd);
            LOG.info("DMLC_TRACKER_URI is " + dmlcTrackerUri);
            LOG.info("DMLC_TRACKER_PORT is " + dmlcTrackerPort);
            LOG.info("DMLC_NUM_WORKER=" + workerNum);

            try {
                schedulerReservedSocket.close();
                final Process xgboostSchedulerProcess = processLaunch.exec(
                        distXgboostSchedulerCmd,
                        schedulerEnv.toArray(new String[schedulerEnv.size()]),
                        this.envs,
                        null);
                LOG.info("Starting thread to redirect stdout of xgboost scheduler process");
                Thread xgboostSchedulerRedirectThread = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            BufferedReader reader;
                            reader =
                                    new BufferedReader(new InputStreamReader(xgboostSchedulerProcess.getInputStream()));
                            String xgboostSchedulerStdoutLog;
                            while ((xgboostSchedulerStdoutLog = reader.readLine()) != null) {
                                LOG.info(xgboostSchedulerStdoutLog);
                                appendMessage(xgboostSchedulerStdoutLog, false);
                                if (conf.getBoolean(
                                        HboxConfiguration.HBOX_CONTAINER_RUNNING_LOG_ENABLE,
                                        HboxConfiguration.DEFAULT_HBOX_CONTAINER_RUNNING_LOG_ENABLE)) {
                                    amContainerStdOut.append(xgboostSchedulerStdoutLog);
                                }
                            }
                        } catch (Exception e) {
                            LOG.warn("Exception in thread xgboostSchedulerRedirectThread");
                            e.printStackTrace();
                        }
                    }
                });
                xgboostSchedulerRedirectThread.start();

                LOG.info("Starting thread to redirect stderr of xgboost scheduler process");
                Thread xgboostSchedulerStderrRedirectThread = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            BufferedReader reader;
                            reader =
                                    new BufferedReader(new InputStreamReader(xgboostSchedulerProcess.getErrorStream()));
                            String xgboostSchedulerStderrLog;
                            while ((xgboostSchedulerStderrLog = reader.readLine()) != null) {
                                LOG.info(xgboostSchedulerStderrLog);
                                appendMessage(xgboostSchedulerStderrLog, false);
                                if (conf.getBoolean(
                                        HboxConfiguration.HBOX_CONTAINER_RUNNING_LOG_ENABLE,
                                        HboxConfiguration.DEFAULT_HBOX_CONTAINER_RUNNING_LOG_ENABLE)) {
                                    amContainerStdErr.append(xgboostSchedulerStderrLog);
                                }
                            }
                        } catch (Exception e) {
                            LOG.warn("Error in thread xgboostSchedulerStderrRedirectThread");
                            e.printStackTrace();
                        }
                    }
                });
                xgboostSchedulerStderrRedirectThread.start();
            } catch (Exception e) {
                LOG.info("start xgboost scheduler error " + e);
            }
        }

        // launch xflow scheduler
        if (("XFLOW").equals(hboxAppType)) {
            LOG.info("Setting environments for the xflow scheduler");
            InetAddress address = null;
            try {
                address = InetAddress.getByName(applicationMasterHostname);
                dmlcPsRootUri = address.getHostAddress();
            } catch (UnknownHostException e) {
                LOG.info("acquire host ip failed " + e);
            }
            Socket schedulerReservedSocket = new Socket();
            try {
                Utilities.getReservePort(
                        schedulerReservedSocket,
                        InetAddress.getByName(applicationMasterHostname).getHostAddress(),
                        reservePortBegin,
                        reservePortEnd);
            } catch (IOException e) {
                LOG.error("Can not get available port");
            }
            dmlcPsRootPort = schedulerReservedSocket.getLocalPort();

            List<String> schedulerEnv = new ArrayList<>(20);
            Map<String, String> userEnv = new HashMap<>();
            if (conf.get(HboxConfiguration.HBOX_CONTAINER_ENV) != null) {
                String[] env = StringUtils.split(conf.get(HboxConfiguration.HBOX_CONTAINER_ENV), "|");
                for (String envPair : env) {
                    String[] userEnvPair = StringUtils.split(envPair, "=");
                    if (userEnvPair.length != 2) {
                        LOG.error(envPair + " is not correct");
                    } else {
                        schedulerEnv.add(envPair);
                        userEnv.put(userEnvPair[0], userEnvPair[1]);
                    }
                }
            }
            if (userEnv.containsKey("PATH")) {
                schedulerEnv.add(
                        "PATH=" + userEnv.get("PATH") + System.getProperty("path.separator") + System.getenv("PATH"));
            } else {
                schedulerEnv.add("PATH=" + System.getenv("PATH"));
            }
            schedulerEnv.add("JAVA_HOME=" + System.getenv("JAVA_HOME"));
            schedulerEnv.add("HADOOP_HOME=" + System.getenv("HADOOP_HOME"));
            schedulerEnv.add("HADOOP_HDFS_HOME=" + System.getenv("HADOOP_HDFS_HOME"));
            if (userEnv.containsKey("LD_LIBRARY_PATH")) {
                schedulerEnv.add("LD_LIBRARY_PATH=" + "./:" + userEnv.get("LD_LIBRARY_PATH")
                        + System.getProperty("path.separator") + System.getenv("LD_LIBRARY_PATH") + ":"
                        + System.getenv("JAVA_HOME") + "/jre/lib/amd64/server:" + System.getenv("HADOOP_HOME")
                        + "/lib/native");
            } else {
                schedulerEnv.add(
                        "LD_LIBRARY_PATH=" + "./:" + System.getenv("LD_LIBRARY_PATH") + ":" + System.getenv("JAVA_HOME")
                                + "/jre/lib/amd64/server:" + System.getenv("HADOOP_HOME") + "/lib/native");
            }
            if (userEnv.containsKey("CLASSPATH")) {
                schedulerEnv.add("CLASSPATH=" + "./:" + userEnv.get("CLASSPATH") + System.getProperty("path.separator")
                        + System.getenv("CLASSPATH") + ":" + System.getProperty("java.class.path"));
            } else {
                schedulerEnv.add("CLASSPATH=" + "./:" + System.getenv("CLASSPATH") + ":"
                        + System.getProperty("java.class.path"));
            }
            schedulerEnv.add("DMLC_ROLE=scheduler");
            schedulerEnv.add("DMLC_PS_ROOT_URI=" + dmlcPsRootUri);
            schedulerEnv.add("DMLC_PS_ROOT_PORT=" + dmlcPsRootPort);
            schedulerEnv.add(HboxConstants.Environment.HBOX_DMLC_WORKER_NUM.toString() + "=" + workerNum);
            schedulerEnv.add(HboxConstants.Environment.HBOX_DMLC_SERVER_NUM.toString() + "=" + psNum);
            schedulerEnv.add("PYTHONUNBUFFERED=1");

            LOG.info("Executing command: " + String.join(" ", hboxCommandArgs));
            LOG.info("DMLC_PS_ROOT_URI is " + dmlcPsRootUri);
            LOG.info("DMLC_PS_ROOT_PORT is " + dmlcPsRootPort);
            LOG.info(HboxConstants.Environment.HBOX_DMLC_WORKER_NUM.toString() + "=" + workerNum);
            LOG.info(HboxConstants.Environment.HBOX_DMLC_SERVER_NUM.toString() + "=" + psNum);

            try {
                schedulerReservedSocket.close();
                final Process xflowSchedulerProcess = processLaunch.exec(
                        hboxCommandArgs, schedulerEnv.toArray(new String[schedulerEnv.size()]), this.envs, null);
                LOG.info("Starting thread to redirect stdout of xflow scheduler process");
                Thread xflowSchedulerRedirectThread = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            BufferedReader reader;
                            reader = new BufferedReader(new InputStreamReader(xflowSchedulerProcess.getInputStream()));
                            String xflowSchedulerStdoutLog;
                            while ((xflowSchedulerStdoutLog = reader.readLine()) != null) {
                                LOG.info(xflowSchedulerStdoutLog);
                                if (conf.getBoolean(
                                        HboxConfiguration.HBOX_CONTAINER_RUNNING_LOG_ENABLE,
                                        HboxConfiguration.DEFAULT_HBOX_CONTAINER_RUNNING_LOG_ENABLE)) {
                                    amContainerStdOut.append(xflowSchedulerStdoutLog);
                                }
                            }
                        } catch (Exception e) {
                            LOG.warn("Exception in thread xflowSchedulerRedirectThread");
                            e.printStackTrace();
                        }
                    }
                });
                xflowSchedulerRedirectThread.start();

                LOG.info("Starting thread to redirect stderr of xflow scheduler process");
                Thread xflowSchedulerStderrRedirectThread = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            BufferedReader reader;
                            reader = new BufferedReader(new InputStreamReader(xflowSchedulerProcess.getErrorStream()));
                            String xflowSchedulerStderrLog;
                            while ((xflowSchedulerStderrLog = reader.readLine()) != null) {
                                LOG.info(xflowSchedulerStderrLog);
                                if (conf.getBoolean(
                                        HboxConfiguration.HBOX_CONTAINER_RUNNING_LOG_ENABLE,
                                        HboxConfiguration.DEFAULT_HBOX_CONTAINER_RUNNING_LOG_ENABLE)) {
                                    amContainerStdErr.append(xflowSchedulerStderrLog);
                                }
                            }
                        } catch (Exception e) {
                            LOG.warn("Error in thread xflowSchedulerStderrRedirectThread");
                            e.printStackTrace();
                        }
                    }
                });
                xflowSchedulerStderrRedirectThread.start();
            } catch (Exception e) {
                LOG.info("start xflow scheduler error " + e);
            }
        }

        // launch mpi exec process
        if (hboxAppType.equals("MPI") || hboxAppType.equals("TENSORNET") || hboxAppType.equals("HOROVOD")) {
            if (hboxAppType.equals("MPI") || hboxAppType.equals("TENSORNET")) launchMpiExec();
            else launchHorovodExec();
            mpiExitCode = -1;
            final long start = System.currentTimeMillis();
            while (mpiContainerCommand == null) {
                Utilities.sleep(statusUpdateInterval);
                try {
                    mpiExitCode = mpiExecProcess.exitValue();
                } catch (IllegalThreadStateException e) {
                    LOG.debug(hboxAppType.toLowerCase() + " mpiexec process is running");
                }
                if (mpiExitCode != -1) {
                    appendMessage(new Message(
                            LogType.STDERR, hboxAppType.toLowerCase() + " exec exit with code " + mpiExitCode));
                    throw new HboxExecException(hboxAppType.toLowerCase() + " mpiexec exit with code " + mpiExitCode);
                }
                if (System.currentTimeMillis() - start > 30_000) { // wait at most 30s
                    final String errMsg =
                            hboxAppType.toLowerCase() + " timeout for extracting mpi rte command from mpiexec output!";
                    LOG.error(errMsg);
                    appendMessage(new Message(LogType.STDERR, errMsg));
                    throw new HboxExecException(errMsg);
                }
            }
        }

        // launch xdl zk and scheduler process
        if (hboxAppType.equals("XDL")) {
            LOG.info("Setting environments for the xdl scheduler");
            InetAddress address = null;
            try {
                address = InetAddress.getByName(applicationMasterHostname);
                xdlZkUri = address.getHostAddress();
            } catch (UnknownHostException e) {
                LOG.info("acquire host ip failed " + e);
            }
            Socket schedulerReservedSocket = new Socket();
            try {
                Utilities.getReservePort(
                        schedulerReservedSocket,
                        InetAddress.getByName(applicationMasterHostname).getHostAddress(),
                        reservePortBegin,
                        reservePortEnd);
            } catch (IOException e) {
                LOG.error("Can not get available port");
            }
            xdlZkPort = schedulerReservedSocket.getLocalPort();
            String xdlZKAddr = "zfs://" + xdlZkUri + ":" + xdlZkPort + "/" + applicationAttemptID.toString();
            List<String> schedulerEnv = new ArrayList<>(20);
            Map<String, String> userEnv = new HashMap<>();
            if (conf.get(HboxConfiguration.HBOX_CONTAINER_ENV) != null) {
                String[] env = StringUtils.split(conf.get(HboxConfiguration.HBOX_CONTAINER_ENV), "|");
                for (String envPair : env) {
                    String[] userEnvPair = StringUtils.split(envPair, "=");
                    if (userEnvPair.length != 2) {
                        LOG.error(envPair + " is not correct");
                    } else {
                        schedulerEnv.add(envPair);
                        userEnv.put(userEnvPair[0], userEnvPair[1]);
                    }
                }
            }
            if (userEnv.containsKey("PATH")) {
                schedulerEnv.add(
                        "PATH=" + userEnv.get("PATH") + System.getProperty("path.separator") + System.getenv("PATH"));
            } else {
                schedulerEnv.add("PATH=" + System.getenv("PATH"));
            }
            schedulerEnv.add("JAVA_HOME=" + System.getenv("JAVA_HOME"));
            schedulerEnv.add("HADOOP_HOME=" + System.getenv("HADOOP_HOME"));
            schedulerEnv.add("HADOOP_HDFS_HOME=" + System.getenv("HADOOP_HDFS_HOME"));
            if (userEnv.containsKey("LD_LIBRARY_PATH")) {
                schedulerEnv.add("LD_LIBRARY_PATH=" + "./:" + userEnv.get("LD_LIBRARY_PATH")
                        + System.getProperty("path.separator") + System.getenv("LD_LIBRARY_PATH") + ":"
                        + System.getenv("JAVA_HOME") + "/jre/lib/amd64/server:" + System.getenv("HADOOP_HOME")
                        + "/lib/native");
            } else {
                schedulerEnv.add(
                        "LD_LIBRARY_PATH=" + "./:" + System.getenv("LD_LIBRARY_PATH") + ":" + System.getenv("JAVA_HOME")
                                + "/jre/lib/amd64/server:" + System.getenv("HADOOP_HOME") + "/lib/native");
            }
            if (userEnv.containsKey("CLASSPATH")) {
                schedulerEnv.add("CLASSPATH=" + "./:" + userEnv.get("CLASSPATH") + System.getProperty("path.separator")
                        + System.getenv("CLASSPATH") + ":" + System.getProperty("java.class.path"));
            } else {
                schedulerEnv.add("CLASSPATH=" + "./:" + System.getenv("CLASSPATH") + ":"
                        + System.getProperty("java.class.path"));
            }
            // xdl zk
            conf.set("DOCKER_CONTAINER_NETWORK", "");
            conf.set("DOCKER_PORT", "2181");
            conf.set("RESERVED_PORT", String.valueOf(xdlZkPort));
            conf.set("DOCKER_CONTAINER_USER", "root");
            try {
                schedulerReservedSocket.close();
                final Process xdlZKProcess = processLaunch.exec(
                        "sh /usr/share/zookeeper/bin/start.sh " + "/" + applicationAttemptID.toString(),
                        schedulerEnv.toArray(new String[schedulerEnv.size()]),
                        this.envs,
                        null);
                LOG.info("Starting thread to redirect stdout of xdl zk process");
                Thread xdlZKRedirectThread = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            BufferedReader reader;
                            reader = new BufferedReader(new InputStreamReader(xdlZKProcess.getInputStream()));
                            String xdlZKStdoutLog;
                            while ((xdlZKStdoutLog = reader.readLine()) != null) {
                                LOG.info(xdlZKStdoutLog);
                                if (conf.getBoolean(
                                        HboxConfiguration.HBOX_CONTAINER_RUNNING_LOG_ENABLE,
                                        HboxConfiguration.DEFAULT_HBOX_CONTAINER_RUNNING_LOG_ENABLE)) {
                                    amContainerStdOut.append(xdlZKStdoutLog);
                                }
                            }
                        } catch (Exception e) {
                            LOG.warn("Exception in thread xdlZKRedirectThread");
                            e.printStackTrace();
                        }
                    }
                });
                xdlZKRedirectThread.start();

                LOG.info("Starting thread to redirect stderr of xdl zk process");
                Thread xdlZKStderrRedirectThread = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            BufferedReader reader;
                            reader = new BufferedReader(new InputStreamReader(xdlZKProcess.getErrorStream()));
                            String xdlZKStderrLog;
                            while ((xdlZKStderrLog = reader.readLine()) != null) {
                                LOG.info(xdlZKStderrLog);
                                if (conf.getBoolean(
                                        HboxConfiguration.HBOX_CONTAINER_RUNNING_LOG_ENABLE,
                                        HboxConfiguration.DEFAULT_HBOX_CONTAINER_RUNNING_LOG_ENABLE)) {
                                    amContainerStdErr.append(xdlZKStderrLog);
                                }
                            }
                        } catch (Exception e) {
                            LOG.warn("Error in thread xdlZKStderrRedirectThread");
                            e.printStackTrace();
                        }
                    }
                });
                xdlZKStderrRedirectThread.start();
            } catch (Exception e) {
                throw new HboxExecException("start xdl zk error " + e);
            }
        }

        if (conf.getBoolean(HboxConfiguration.HBOX_INPUT_STREAM, HboxConfiguration.DEFAULT_HBOX_INPUT_STREAM)
                || conf.get(HboxConfiguration.HBOX_INPUT_STRATEGY, HboxConfiguration.DEFAULT_HBOX_INPUT_STRATEGY)
                        .equals("STREAM")) {
            if (inputFileSplits != null) {
                allocateInputStreamSplits();
            } else {
                LOG.info("Don't have the input splits to allocated");
            }
        } else {
            if (conf.get(HboxConfiguration.HBOX_INPUT_STRATEGY, HboxConfiguration.DEFAULT_HBOX_INPUT_STRATEGY)
                            .equals("PLACEHOLDER")
                    && conf.getBoolean(
                            HboxConfiguration.HBOX_PLACEHOLDER_WHOLE_ENABLE,
                            HboxConfiguration.DEFAULT_HBOX_PLACEHOLDER_WHOLE_ENABLE)) {
                allocateWholeInput();
                allocateWholeS3Input();
            } else {
                if (conf.getBoolean(
                        HboxConfiguration.HBOX_TF_INPUT_PS_ENABLE, HboxConfiguration.DEFAULT_HBOX_TF_INPUT_PS_ENABLE)) {
                    allocateInputSplitsInlcudePs();
                } else {
                    allocateInputSplits();
                    allocateS3InputSplits();
                }
            }
        }
        buildOutputLocations();
        buildContainerLocalResource();
        Map<String, String> workerContainerEnv = buildContainerEnv(HboxConstants.WORKER);
        Map<String, String> psContainerEnv = buildContainerEnv(HboxConstants.PS);
        Map<String, String> schedulerContainerEnv = buildContainerEnv(HboxConstants.SCHEDULER);
        List<String> workerContainerLaunchcommands = buildContainerLaunchCommand(HboxConstants.WORKER, workerMemory);
        List<String> psContainerLaunchcommands = buildContainerLaunchCommand(HboxConstants.PS, psMemory);

        boolean jobExecTimeoutFlag = false;
        long startJobExecTimeStamp = Long.MIN_VALUE;
        long jobExecTimeout =
                conf.getLong(HboxConfiguration.HBOX_JOB_EXEC_TIMEOUT, HboxConfiguration.DEFAULT_HBOX_JOB_EXEC_TIME);
        String failMessage = "Job running timeout after " + jobExecTimeout / 1000 + " seconds. ";
        if (jobExecTimeout > 0) {
            jobExecTimeoutFlag = true;
            startJobExecTimeStamp = System.currentTimeMillis();
        }

        LOG.info("Launching containers");
        if (hboxAppType.equals("XDL") && psNum > 0) {
            schedulerContainerId = acquiredPsContainers.get(0).getId().toString();
        }
        int index = 0;
        for (Container container : acquiredPsContainers) {
            LOG.info("Launching ps container " + container.getId() + " on "
                    + container.getNodeId().getHost() + ":"
                    + container.getNodeId().getPort());

            // TODO launch container in special thread take with fault-tolerant
            if (container.getId().toString().equals(schedulerContainerId)) {
                launchContainer(containerLocalResource, schedulerContainerEnv, psContainerLaunchcommands, container, 0);
            } else {
                launchContainer(containerLocalResource, psContainerEnv, psContainerLaunchcommands, container, index++);
            }
            containerListener.registerContainer(new HboxContainerId(container.getId()), HboxConstants.PS);
        }
        if (hboxAppType.equals("MPI") || hboxAppType.equals("TENSORNET") || hboxAppType.equals("HOROVOD")) {
            index = 1;
        } else {
            index = 0;
        }

        if (chiefWorker) {
            chiefWorkerContainerId =
                    acquiredChiefWorkerContainers.get(0).getId().toString();
        }
        if (tfEvaluator) {
            tfEvaluatorContainerId =
                    acquiredEvaluatorWorkerContainers.get(0).getId().toString();
        }

        for (Container container : acquiredWorkerContainers) {
            LOG.info("Launching worker container " + container.getId() + " on "
                    + container.getNodeId().getHost() + ":"
                    + container.getNodeId().getPort());

            // TODO launch container in special thread take with fault-tolerant
            if (chiefWorker && container.getId().toString().equals(chiefWorkerContainerId)) {
                List<String> chiefWorkerContainerLaunchcommands =
                        buildContainerLaunchCommand(HboxConstants.CHIEF, chiefWorkerMemory);
                launchContainer(
                        containerLocalResource,
                        workerContainerEnv,
                        chiefWorkerContainerLaunchcommands,
                        container,
                        index++);
            } else if (tfEvaluator && container.getId().toString().equals(tfEvaluatorContainerId)) {
                Map<String, String> evaluatorWorkerContainerEnv = buildContainerEnv(HboxConstants.EVALUATOR);
                List<String> evaluatorWorkerContainerLaunchcommands =
                        buildContainerLaunchCommand(HboxConstants.EVALUATOR, evaluatorWorkerMemory);
                launchContainer(
                        containerLocalResource,
                        evaluatorWorkerContainerEnv,
                        evaluatorWorkerContainerLaunchcommands,
                        container,
                        0);
            } else {
                launchContainer(
                        containerLocalResource, workerContainerEnv, workerContainerLaunchcommands, container, index++);
            }
            containerListener.registerContainer(new HboxContainerId(container.getId()), HboxConstants.WORKER);
        }

        if (hboxAppType.equals("MPI") || hboxAppType.equals("TENSORNET") || hboxAppType.equals("HOROVOD")) {
            while (!containerListener.isAllContainerStarted()) {
                Utilities.sleep(statusUpdateInterval);
                if (jobExecTimeoutFlag && (System.currentTimeMillis() - startJobExecTimeStamp) > jobExecTimeout) {
                    throw new HboxExecException(failMessage);
                }
            }

            this.appendMessage(new Message(LogType.STDERR, "All containers are launched successfully"));
            span.addEvent("all_container_started");

            while (mpiExitCode == -1) {
                Utilities.sleep(statusUpdateInterval);
                try {
                    mpiExitCode = mpiExecProcess.exitValue();
                } catch (IllegalThreadStateException e) {
                    LOG.debug(hboxAppType.toLowerCase() + " exec process is running");
                }
                if (jobExecTimeoutFlag && (System.currentTimeMillis() - startJobExecTimeStamp) > jobExecTimeout) {
                    throw new HboxExecException(failMessage);
                }
            }
            appendMessage(new Message(LogType.STDERR, "finish mpiexec with code " + mpiExitCode));
        } else {
            span.addEvent("all_container_started");
        }

        String diagnostics = "";
        boolean finalSuccess;

        if (this.applicationContext.getOutputs().size() > 0) {
            final Thread saveInnerModelMonitor = new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        try {
                            Boolean startSaved = applicationContext.getStartSavingStatus();
                            containerListener.setSaveInnerModel(startSaved);
                            if (savingInterval > 0 && !startSaved) {
                                if ((containerListener.interResultTimeStamp() < 0
                                                && containerListener.allContainerStartTime() > 0
                                                && (System.currentTimeMillis()
                                                                - containerListener.allContainerStartTime())
                                                        > savingInterval)
                                        || ((System.currentTimeMillis() - containerListener.interResultTimeStamp())
                                                > savingInterval)) {
                                    applicationContext.startSavingModelStatus(true);
                                    containerListener.setSaveInnerModel(true);
                                    startSaved = true;
                                }
                            }
                            while (startSaved) {
                                if (containerListener.interResultCompletedNum(containerListener.interResultTimeStamp())
                                        == containerListener.getInnerSavingContainerNum()) {
                                    lastSavingStatus = true;
                                    if (!savingModelList.contains(containerListener.interResultTimeStamp())) {
                                        savingModelList.add(containerListener.interResultTimeStamp());
                                    }
                                    break;
                                }
                                Utilities.sleep(conf.getInt(
                                        HboxConfiguration.HBOX_CONTAINER_HEARTBEAT_INTERVAL,
                                        HboxConfiguration.DEFAULT_HBOX_CONTAINER_HEARTBEAT_INTERVAL));
                            }
                        } catch (Exception e) {
                            LOG.info("Monitor the InnerModel saving error: " + e);
                        }
                        Utilities.sleep(conf.getInt(
                                HboxConfiguration.HBOX_CONTAINER_HEARTBEAT_INTERVAL,
                                HboxConfiguration.DEFAULT_HBOX_CONTAINER_HEARTBEAT_INTERVAL));
                    }
                }
            });
            saveInnerModelMonitor.start();
        }

        containerStarted = true;

        try {
            boolean flag = true;
            boolean digitsFlag = true;
            if (!hboxAppType.equals("MPI") && !hboxAppType.equals("TENSORNET") && !hboxAppType.equals("HOROVOD")) {
                LOG.info("Waiting for train completed");
                Map<HboxContainerId, HboxContainerStatus> lastWorkerContainerStatus = new ConcurrentHashMap<>();
                Map<HboxContainerId, HboxContainerStatus> lastPsContainerStatus = new ConcurrentHashMap<>();
                while (!containerListener.isTrainCompleted()) {
                    // report progress to client
                    if (conf.getBoolean(
                                    HboxConfiguration.HBOX_REPORT_CONTAINER_STATUS,
                                    HboxConfiguration.DEFAULT_HBOX_REPORT_CONTAINER_STATUS)
                            && !hboxAppType.equals("MPI")
                            && !hboxAppType.equals("TENSORNET")
                            && !hboxAppType.equals("HOROVOD")) {
                        List<Container> workerContainersStatus = applicationContext.getWorkerContainers();
                        List<Container> psContainersStatus = applicationContext.getPsContainers();
                        for (Container container : workerContainersStatus) {
                            if (!lastWorkerContainerStatus.containsKey(new HboxContainerId(container.getId()))) {
                                lastWorkerContainerStatus.put(
                                        new HboxContainerId(container.getId()), HboxContainerStatus.STARTED);
                            }
                            if (!applicationContext
                                    .getContainerStatus(new HboxContainerId(container.getId()))
                                    .equals(lastWorkerContainerStatus.get(new HboxContainerId(container.getId())))) {
                                this.appendMessage(
                                        "container " + container.getId().toString() + " status is "
                                                + applicationContext.getContainerStatus(
                                                        new HboxContainerId(container.getId())),
                                        false);
                                lastWorkerContainerStatus.put(
                                        new HboxContainerId(container.getId()),
                                        applicationContext.getContainerStatus(new HboxContainerId(container.getId())));
                            }
                        }
                        for (Container container : psContainersStatus) {
                            if (!lastPsContainerStatus.containsKey(new HboxContainerId(container.getId()))) {
                                lastPsContainerStatus.put(
                                        new HboxContainerId(container.getId()), HboxContainerStatus.STARTED);
                            }
                            if (!applicationContext
                                    .getContainerStatus(new HboxContainerId(container.getId()))
                                    .equals(lastPsContainerStatus.get(new HboxContainerId(container.getId())))) {
                                this.appendMessage(
                                        "container " + container.getId().toString() + " status is "
                                                + applicationContext.getContainerStatus(
                                                        new HboxContainerId(container.getId())),
                                        false);
                                lastPsContainerStatus.put(
                                        new HboxContainerId(container.getId()),
                                        applicationContext.getContainerStatus(new HboxContainerId(container.getId())));
                            }
                        }
                    }

                    String containerType = conf.get(
                                    HboxConfiguration.CONTAINER_EXECUTOR_TYPE,
                                    HboxConfiguration.DEFAULT_CONTAINER_EXECUTOR_TYPE)
                            .toUpperCase();
                    List<Container> workerContainers = applicationContext.getWorkerContainers();
                    List<Container> psContainers = applicationContext.getPsContainers();
                    if (hboxAppType.equals("VPC") || hboxAppType.equals("DIGITS") || containerType.equals("DOCKER")) {
                        if (flag) {
                            Map<HboxContainerId, String> vpcCommandAndPasswdMap =
                                    applicationContext.getVPCCommandAndPasswdMap();
                            if (vpcCommandAndPasswdMap.size() == (workerNum + psNum)) {
                                for (Container container : workerContainers) {
                                    String commandCombine =
                                            vpcCommandAndPasswdMap.get(new HboxContainerId(container.getId()));
                                    String[] splits = commandCombine.split(":");
                                    this.appendMessage(
                                            "Received vpc login command and password from "
                                                    + container.getId().toString(),
                                            true);
                                    this.appendMessage("Login command:ssh " + splits[0], true);
                                    this.appendMessage("Password:" + splits[1], true);
                                }
                                for (Container container : psContainers) {
                                    String commandCombine =
                                            vpcCommandAndPasswdMap.get(new HboxContainerId(container.getId()));
                                    String[] splits = commandCombine.split(":");
                                    this.appendMessage(
                                            "Received vpc login command and password from "
                                                    + container.getId().toString(),
                                            true);
                                    this.appendMessage("Login command:ssh " + splits[0], true);
                                    this.appendMessage("Password:" + splits[1], true);
                                }
                                flag = false;
                            } else {
                                this.appendMessage("Waiting for vpc login command and password...", false);
                            }
                        }
                        if (hboxAppType.equals("DIGITS")) {
                            if (digitsFlag) {
                                Map<HboxContainerId, String> digitsUrlMap = applicationContext.getDigitsUrlMap();
                                if (digitsUrlMap.size() == (workerNum + psNum)) {
                                    for (Container container : workerContainers) {
                                        String url = digitsUrlMap.get(new HboxContainerId(container.getId()));
                                        this.appendMessage(
                                                "Received digits server url from "
                                                        + container.getId().toString(),
                                                true);
                                        this.appendMessage("digits server url: " + url, true);
                                    }
                                    for (Container container : psContainers) {
                                        String url = digitsUrlMap.get(new HboxContainerId(container.getId()));
                                        this.appendMessage(
                                                "Received digits server url from "
                                                        + container.getId().toString(),
                                                true);
                                        this.appendMessage("digits server url: " + url, true);
                                    }
                                    digitsFlag = false;
                                } else {
                                    this.appendMessage("Waiting for digits server url...", false);
                                }
                            }
                        }
                    }

                    // check the gpu device assigned whether correct
                    if (conf.getInt(HboxConfiguration.HBOX_PS_GPU, HboxConfiguration.DEFAULT_HBOX_PS_GPU) > 0) {
                        for (Container container : psContainers) {
                            HboxContainerId containerId = new HboxContainerId(container.getId());
                            if (applicationContext.getContainerGPUDevice(containerId) != null) {
                                String gpus = applicationContext.getContainerGPUDevice(containerId);
                                if (gpus.equals("")
                                        || gpus.equals(null)
                                        || gpus.split(",").length
                                                != conf.getInt(
                                                        HboxConfiguration.HBOX_PS_GPU,
                                                        HboxConfiguration.DEFAULT_HBOX_PS_GPU))
                                    throw new RuntimeException("ps container " + containerId.toString()
                                            + " is not assigned the correct gpu device. Now assigned info is " + gpus);
                            }
                        }
                    }

                    if (conf.getInt(HboxConfiguration.HBOX_WORKER_GPU, HboxConfiguration.DEFAULT_HBOX_WORKER_GPU) > 0) {
                        for (Container container : workerContainers) {
                            HboxContainerId containerId = new HboxContainerId(container.getId());
                            if (applicationContext.getContainerGPUDevice(containerId) != null) {
                                String gpus = applicationContext.getContainerGPUDevice(containerId);
                                if (gpus.equals("")
                                        || gpus.equals(null)
                                        || gpus.split(",").length
                                                != conf.getInt(
                                                        HboxConfiguration.HBOX_WORKER_GPU,
                                                        HboxConfiguration.DEFAULT_HBOX_WORKER_GPU))
                                    throw new RuntimeException("worker container " + containerId.toString()
                                            + " is not assigned the correct gpu device. Now assigned info is " + gpus);
                            }
                        }
                    }

                    Map<HboxContainerId, String> clientProgress = applicationContext.getReporterProgress();
                    float total = 0.0f;
                    for (Container container : workerContainers) {
                        String progressLog = clientProgress.get(new HboxContainerId(container.getId()));
                        if (progressLog != null && !progressLog.equals("")) {
                            String[] progress = progressLog.toString().split(":");
                            if (progress.length != 2) {
                                this.appendMessage("progress log format error", false);
                            } else {
                                try {
                                    Float percentProgress = Float.parseFloat(progress[1]);
                                    if (percentProgress < 0.0 || percentProgress > 1.0) {
                                        this.appendMessage("progress log format error", false);
                                    } else {
                                        total += Float.parseFloat(progress[1]);
                                    }
                                } catch (Exception e) {
                                    this.appendMessage("progress log format error", false);
                                }
                            }
                        }
                    }
                    if (total > 0.0f) {
                        float finalProgress = total / workerContainers.size();
                        DecimalFormat df = new DecimalFormat("0.00");
                        df.setRoundingMode(RoundingMode.HALF_UP);
                        // this.appendMessage("reporter progress:" + Float.toString(finalProgress*100) + "%", false);
                        this.appendMessage("reporter progress:" + df.format(finalProgress * 100) + "%", false);
                        rmCallbackHandler.setProgress(finalProgress);
                    }
                    if (jobExecTimeoutFlag && (System.currentTimeMillis() - startJobExecTimeStamp) > jobExecTimeout) {
                        throw new HboxExecException(failMessage);
                    }
                    Utilities.sleep(statusUpdateInterval);
                }
                LOG.info("Train completed");
                containerListener.setTrainFinished();

                if ((("TENSORFLOW".equals(hboxAppType)
                                        || "TENSOR2TENSOR".equals(hboxAppType)
                                        || "MXNET".equals(hboxAppType)
                                        || "XDL".equals(hboxAppType))
                                && psNum > 0)
                        || "DISTLIGHTLDA".equals(hboxAppType)
                        || "XFLOW".equals(hboxAppType)) {
                    LOG.info("Waiting all ps containers completed");
                    while (!containerListener.isAllPsContainersFinished()) {
                        Utilities.sleep(statusUpdateInterval);
                    }
                    LOG.info("All ps containers completed");
                }
                finalSuccess = containerListener.isAllWorkerContainersSucceeded();
                span.addEvent("all_container_completed");
            } else {
                containerListener.setAMFinished();
                LOG.info("Waiting all containers completed");
                finalSuccess = mpiExitCode == 0;
                while (!containerListener.isTrainCompleted()) {
                    Utilities.sleep(statusUpdateInterval);
                    if (jobExecTimeoutFlag && (System.currentTimeMillis() - startJobExecTimeStamp) > jobExecTimeout) {
                        throw new HboxExecException(failMessage);
                    }
                }
                LOG.info("All containers completed");
                span.addEvent("all_container_completed", Attributes.of(AttributeKey.longKey("hbox.mpiexec.exit"), (long)
                        mpiExitCode));
            }

            boolean uploadWhenFailed = this.conf.getBoolean(
                    HboxConfiguration.HBOX_FAILED_UPLOAD, HboxConfiguration.DEFAULT_HBOX_FAILED_UPLOAD);
            if (uploadWhenFailed) LOG.info("This application allow failed container to upload output files!");
            if (finalSuccess || uploadWhenFailed) {
                if ((conf.getBoolean(HboxConfiguration.HBOX_OUTPUT_STREAM, HboxConfiguration.DEFAULT_HBOX_OUTPUT_STREAM)
                                || conf.get(
                                                HboxConfiguration.HBOX_OUTPUT_STRATEGY,
                                                HboxConfiguration.DEFAULT_HBOX_OUTPUT_STRATEGY)
                                        .equals("STREAM"))
                        && outputInfos.size() > 0) {
                    LOG.info("HBOX_OUTPUT_STRATEGY is STREAM, AM handling the final result...");
                    FileSystem fs = new Path(outputInfos.get(0).getDfsLocation()).getFileSystem(conf);
                    Map<HboxContainerId, String> mapPath = applicationContext.getMapedTaskID();
                    for (Container finishedContainer : acquiredWorkerContainers) {
                        String taskID = mapPath.get(new HboxContainerId(finishedContainer.getId()));
                        Path tmpResultPath = new Path(outputInfos.get(0).getDfsLocation() + "/_temporary/"
                                + finishedContainer.getId().toString() + "/_temporary/0/_temporary/" + taskID);
                        LOG.info("tmpResultPath is " + tmpResultPath.toString());
                        Path finalResultPath = new Path(outputInfos.get(0).getDfsLocation() + "/"
                                + finishedContainer.getId().toString());
                        LOG.info("finalResultPath is " + finalResultPath.toString());
                        if (fs.exists(tmpResultPath)) {
                            LOG.info("Move from " + tmpResultPath.toString() + " to " + finalResultPath.toString());
                            fs.rename(tmpResultPath, finalResultPath);
                        }
                    }
                    Path tmpPath = new Path(outputInfos.get(0).getDfsLocation() + "/_temporary/");
                    if (fs.exists(tmpPath)) {
                        fs.delete(tmpPath, true);
                    }
                    fs.createNewFile(new Path(outputInfos.get(0).getDfsLocation() + "/_SUCCESS"));
                } else {
                    for (OutputInfo outputInfo : outputInfos) {
                        if (outputInfo.getOutputType().equals(HboxConstants.S3)) {
                            continue;
                        }
                        FileSystem fs = new Path(outputInfo.getDfsLocation()).getFileSystem(conf);
                        Path finalResultPath = new Path(outputInfo.getDfsLocation());
                        // spec one container to upload
                        if (outputIndex >= 0) {
                            Path tmpResultPath = new Path(
                                    outputInfo.getDfsLocation() + "/_temporary/" + outputInfo.getLocalLocation());
                            if (fs.exists(tmpResultPath)) {
                                LOG.debug("Move from " + tmpResultPath.toString() + " to " + finalResultPath);
                                fs.rename(tmpResultPath, finalResultPath);
                            }
                        } else {
                            for (Container finishedContainer : acquiredWorkerContainers) {
                                Path tmpResultPath = new Path(outputInfo.getDfsLocation() + "/_temporary/"
                                        + finishedContainer.getId().toString());
                                if (workerNum == 1
                                        && !conf.getBoolean(
                                                HboxConfiguration.HBOX_CREATE_CONTAINERID_DIR,
                                                HboxConfiguration.DEFAULT_HBOX_CREATE_CONTAINERID_DIR)) {
                                    tmpResultPath = new Path(outputInfo.getDfsLocation() + "/_temporary/"
                                            + outputInfo.getLocalLocation());
                                }
                                if (fs.exists(tmpResultPath)) {
                                    LOG.debug("Move from " + tmpResultPath.toString() + " to " + finalResultPath);
                                    fs.rename(tmpResultPath, finalResultPath);
                                }
                            }
                            if (psNum > 0
                                    && (hboxAppType.equals("DISTLIGHTLDA")
                                            || hboxAppType.equals("TENSORFLOW")
                                            || "TENSOR2TENSOR".equals(hboxAppType)
                                            || hboxAppType.equals("XDL"))) {
                                for (Container finishedContainer : acquiredPsContainers) {
                                    Path tmpResultPath = new Path(outputInfo.getDfsLocation() + "/_temporary/"
                                            + finishedContainer.getId().toString());
                                    if (fs.exists(tmpResultPath)) {
                                        LOG.debug("Move from " + tmpResultPath.toString() + " to " + finalResultPath);
                                        fs.rename(tmpResultPath, finalResultPath);
                                    }
                                }
                            }
                        }

                        Path tmpPath = new Path(outputInfo.getDfsLocation() + "/_temporary/");
                        if (fs.exists(tmpPath)) {
                            fs.delete(tmpPath, true);
                        }
                        fs.createNewFile(new Path(outputInfo.getDfsLocation() + "/_SUCCESS"));
                    }
                }
                span.addEvent("upload_done");
            }
        } catch (Exception e) {
            finalSuccess = false;
            this.appendMessage("Some error occurs " + org.apache.hadoop.util.StringUtils.stringifyException(e), true);
            diagnostics = e.getMessage();
            span.recordException(e);
        }

        int appAttempts =
                conf.getInt(HboxConfiguration.HBOX_APP_MAX_ATTEMPTS, HboxConfiguration.DEFAULT_HBOX_APP_MAX_ATTEMPTS);
        if (appAttempts
                > conf.getInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, YarnConfiguration.DEFAULT_RM_AM_MAX_ATTEMPTS)) {
            appAttempts =
                    conf.getInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, YarnConfiguration.DEFAULT_RM_AM_MAX_ATTEMPTS);
        }

        if (!finalSuccess && applicationAttemptID.getAttemptId() < appAttempts) {
            Runtime.getRuntime().removeShutdownHook(cleanApplication);
            throw new RuntimeException("Application Failed, retry starting. Note that container memory auto scale");
        }

        this.appendMessage("Flush log and Unregister Application", true);
        unregisterApp(finalSuccess ? FinalApplicationStatus.SUCCEEDED : FinalApplicationStatus.FAILED, diagnostics);

        return finalSuccess;
    }
Async Method telling NMClientAsync to launch specific container

@param container the container which should be launched
@return is launched success
core/src/main/java/net/qihoo/hbox/AM/ApplicationMaster.java