private void init() {
        appendMessage(new Message(LogType.STDOUT, "ApplicationMaster starting services"));

        this.rmCallbackHandler = new RMCallbackHandler();
        this.amrmSync = AMRMClient.<ContainerRequest>createAMRMClient();
        this.amrmAsync = AMRMClientAsync.<ContainerRequest>createAMRMClientAsync(amrmSync, 1000, rmCallbackHandler);
        this.amrmAsync.init(conf);

        NMCallbackHandler nmAsyncHandler = new NMCallbackHandler();
        this.nmAsync = NMClientAsync.createNMClientAsync(nmAsyncHandler);
        this.nmAsync.init(conf);

        addService(this.amrmAsync);
        addService(this.nmAsync);
        addService(this.messageService);
        addService(this.webService);
        addService(this.containerListener);
        try {
            super.serviceStart();
        } catch (Exception e) {
            throw new RuntimeException("Error start application services!", e);
        }

        applicationMasterTrackingUrl = "http://" + applicationMasterHostname + ":" + this.webService.getHttpPort();
        String historyWebappAddress = conf.get(
                HboxConfiguration.HBOX_HISTORY_WEBAPP_ADDRESS, HboxConfiguration.DEFAULT_HBOX_HISTORY_WEBAPP_ADDRESS);
        String cluster = conf.get(HboxConfiguration.HBOX_CLUSTER_NAME, HboxConfiguration.DEFAULT_HBOX_CLUSTER_NAME);
        if (!cluster.equals("")) {
            String clusterHistoryWebappAddress =
                    conf.get(HboxConfiguration.HBOX_CLUSTER_HISTORY_WEBAPP_ADDRESS.replace("cluster.name", cluster));
            if (clusterHistoryWebappAddress == null || clusterHistoryWebappAddress.equals("")) {
                LOG.warn("Note that not set the cluster history webaddress.");
            } else {
                historyWebappAddress = clusterHistoryWebappAddress;
                LOG.info("History webApp address has updated! ");
            }
        }
        applicationHistoryUrl =
                "http://" + historyWebappAddress + "/jobhistory/job/" + applicationAttemptID.getApplicationId();
        LOG.info("master tracking url: " + applicationMasterTrackingUrl);
        LOG.info("history url: " + applicationHistoryUrl);

        cleanApplication = new Thread(new Runnable() {
            @Override
            public void run() {
                HboxConfiguration hboxConf = new HboxConfiguration();
                hboxConf.setBoolean("fs.hdfs.impl.disable.cache", true);
                hboxConf.setBoolean("fs.hdfsold.impl.disable.cache", true);

                if (hboxConf.getBoolean(
                        HboxConfiguration.HBOX_CLEANUP_ENABLE, HboxConfiguration.DEFAULT_HBOX_CLEANUP_ENABLE)) {
                    Path stagingDir = new Path(envs.get(HboxConstants.Environment.HBOX_STAGING_LOCATION.toString()));
                    try {
                        FileSystem stagingFS = FileSystem.get(hboxConf);
                        stagingFS.delete(stagingDir);
                        LOG.info("Deleting the staging file successed.");
                    } catch (Exception e) {
                        LOG.info("Deleting the staging file Error." + e);
                    }
                }

                try {
                    FsPermission LOG_FILE_PERMISSION = FsPermission.createImmutable((short) 0777);
                    Path logdir = new Path(conf.get(
                                    HboxConfiguration.HBOX_HISTORY_LOG_DIR,
                                    HboxConfiguration.DEFAULT_HBOX_HISTORY_LOG_DIR)
                            + "/" + applicationAttemptID.getApplicationId().toString()
                            + "/" + applicationAttemptID.getApplicationId().toString());
                    Path jobLogPath = new Path(hboxConf.get("fs.defaultFS"), logdir);
                    LOG.info("jobLogPath:" + jobLogPath.toString());
                    LOG.info("Start write the log to " + jobLogPath.toString());
                    FileSystem fs = FileSystem.get(hboxConf);
                    FSDataOutputStream out = fs.create(jobLogPath);
                    fs.setPermission(jobLogPath, new FsPermission(LOG_FILE_PERMISSION));
                    if (conf.getBoolean(
                            HboxConfiguration.HBOX_HOST_LOCAL_ENABLE,
                            HboxConfiguration.DEFAULT_HBOX_HOST_LOCAL_ENABLE)) {
                        Path hostLocaldir = new Path(conf.get(
                                        HboxConfiguration.HBOX_HISTORY_LOG_DIR,
                                        HboxConfiguration.DEFAULT_HBOX_HISTORY_LOG_DIR)
                                + "/" + conf.get("hadoop.job.ugi").split(",")[0]
                                + "/" + envs.get(HboxConstants.Environment.HBOX_APP_NAME.toString()));
                        Path hostLocalPath = new Path(hboxConf.get("fs.defaultFS"), hostLocaldir);
                        try {
                            FSDataOutputStream hostLocalOut = fs.create(hostLocalPath);
                            fs.setPermission(hostLocalPath, new FsPermission(LOG_FILE_PERMISSION));
                            hostLocalOut.writeBytes(containerHostnames
                                    .toString()
                                    .substring(1, containerHostnames.toString().length() - 1));
                            hostLocalOut.close();
                            LOG.info("host local enable is true, write " + hostLocalPath.toString() + " success");
                        } catch (Exception e) {
                            LOG.info("write host local file error, " + e);
                        }
                    }
                    Map<String, List<String>> logMessage = new HashMap<>();
                    logMessage.put("appType", Arrays.asList(hboxAppType));
                    if (!(hboxAppType.equals("VPC") || hboxAppType.equals("DIGITS"))) {
                        List<String> tensorboardInfo = new ArrayList<>();
                        if (conf.getBoolean(
                                HboxConfiguration.HBOX_TF_BOARD_ENABLE,
                                HboxConfiguration.DEFAULT_HBOX_TF_BOARD_ENABLE)) {
                            Path boardLogPath;
                            if (!conf.get(
                                            HboxConfiguration.HBOX_TF_BOARD_LOG_DIR,
                                            HboxConfiguration.DEFAULT_HBOX_TF_BOARD_LOG_DIR)
                                    .contains("hdfs://")) {
                                if (conf.get(
                                                HboxConfiguration.HBOX_TF_BOARD_HISTORY_DIR,
                                                HboxConfiguration.HBOX_TF_BOARD_HISTORY_DIR)
                                        .equals(hboxConf.get(
                                                HboxConfiguration.HBOX_TF_BOARD_HISTORY_DIR,
                                                HboxConfiguration.HBOX_TF_BOARD_HISTORY_DIR))) {
                                    boardLogPath = new Path(
                                            hboxConf.get("fs.defaultFS"),
                                            conf.get(
                                                            HboxConfiguration.HBOX_TF_BOARD_HISTORY_DIR,
                                                            HboxConfiguration.DEFAULT_HBOX_TF_BOARD_HISTORY_DIR)
                                                    + "/"
                                                    + applicationAttemptID
                                                            .getApplicationId()
                                                            .toString());
                                } else {
                                    boardLogPath = new Path(
                                            conf.get("fs.defaultFS"),
                                            conf.get(
                                                    HboxConfiguration.HBOX_TF_BOARD_HISTORY_DIR,
                                                    HboxConfiguration.DEFAULT_HBOX_TF_BOARD_HISTORY_DIR));
                                }
                            } else {
                                boardLogPath = new Path(conf.get(HboxConfiguration.HBOX_TF_BOARD_LOG_DIR));
                            }
                            tensorboardInfo.add(boardLogPath.toString());
                        } else {
                            tensorboardInfo.add("-");
                        }
                        logMessage.put("board", tensorboardInfo);
                    }

                    userName = StringUtils.split(conf.get("hadoop.job.ugi"), ',')[0];
                    List<Container> workerContainers = applicationContext.getWorkerContainers();
                    List<Container> psContainers = applicationContext.getPsContainers();
                    Map<HboxContainerId, String> reporterProgress = applicationContext.getReporterProgress();
                    Map<HboxContainerId, String> containersAppStartTime =
                            applicationContext.getContainersAppStartTime();
                    Map<HboxContainerId, String> containersAppFinishTime =
                            applicationContext.getContainersAppFinishTime();

                    // container info schema (since 1.3a b30b967d7c283e8ac542b66413fcc2bc5c5fb95c):
                    // 0: node url
                    // 1: gpu device id
                    // 2: role
                    // 3: status
                    // 4: cpu metrics
                    // 5: gpu mem metrics
                    // 6: gpu util metrics
                    // 7: start time
                    // 8: finish time
                    // 9: percent progress
                    // 10: log url
                    // 11: stats - cpu
                    // 12: stats - gpu mem
                    // 13: stats - gpu utils
                    // 14: stats - mem usage warn (if cpuStatistics.size > 0 || version >= 1.9.2)
                    // 15: rank (since 1.9.2)
                    int workerIdx = 0;
                    for (Container container : workerContainers) {
                        List<String> containerMessage = new ArrayList<>();
                        containerMessage.add(container.getNodeHttpAddress());
                        HboxContainerId currentContainerID = new HboxContainerId(container.getId());
                        String rank = "-";
                        if (applicationContext.getContainerGPUDevice(currentContainerID) != null) {
                            if (applicationContext
                                            .getContainerGPUDevice(currentContainerID)
                                            .trim()
                                            .length()
                                    != 0) {
                                containerMessage.add(applicationContext.getContainerGPUDevice(currentContainerID));
                            } else {
                                containerMessage.add("-");
                            }
                        } else {
                            containerMessage.add("-");
                        }
                        if (tfEvaluator && currentContainerID.toString().equals(tfEvaluatorContainerId)) {
                            containerMessage.add(HboxConstants.EVALUATOR);
                        } else if (chiefWorker && currentContainerID.toString().equals(chiefWorkerContainerId)) {
                            containerMessage.add(HboxConstants.CHIEF);
                        } else {
                            containerMessage.add(HboxConstants.WORKER);
                            rank = "" + workerIdx++;
                        }

                        HboxContainerStatus status = applicationContext.getContainerStatus(currentContainerID);
                        if (status != null) {
                            containerMessage.add(status.toString());
                        } else {
                            containerMessage.add("-");
                        }
                        List<String> usageStatistics = new ArrayList<>();
                        ConcurrentHashMap<String, LinkedBlockingDeque<Object>> cpuMetrics =
                                applicationContext.getContainersCpuMetrics().get(currentContainerID);
                        containerMessage.add(new Gson().toJson(cpuMetrics));
                        ConcurrentHashMap<String, List<Double>> cpuStatistics =
                                applicationContext.getContainersCpuStatistics().get(currentContainerID);
                        usageStatistics.add(new Gson().toJson(cpuStatistics));
                        if (workerGCores > 0) {
                            ConcurrentHashMap<String, LinkedBlockingDeque<List<Long>>> containersGpuMemMetrics =
                                    applicationContext
                                            .getContainersGpuMemMetrics()
                                            .get(currentContainerID);
                            ConcurrentHashMap<String, LinkedBlockingDeque<List<Long>>> containersGpuUtilMetrics =
                                    applicationContext
                                            .getContainersGpuUtilMetrics()
                                            .get(currentContainerID);
                            containerMessage.add(new Gson().toJson(containersGpuMemMetrics));
                            containerMessage.add(new Gson().toJson(containersGpuUtilMetrics));
                            ConcurrentHashMap<String, List<Double>> gpuMemStatistics = applicationContext
                                    .getContainersGpuMemStatistics()
                                    .get(currentContainerID);
                            ConcurrentHashMap<String, List<Double>> gpuUtilStatistics = applicationContext
                                    .getContainersGpuUtilStatistics()
                                    .get(currentContainerID);
                            usageStatistics.add(new Gson().toJson(gpuMemStatistics));
                            usageStatistics.add(new Gson().toJson(gpuUtilStatistics));
                        } else {
                            containerMessage.add("-");
                            containerMessage.add("-");
                            usageStatistics.add("-");
                            usageStatistics.add("-");
                        }

                        if (cpuStatistics.size() != 0) {
                            Double cpuMemUsagedMax = cpuStatistics.get("CPUMEM").get(1);
                            int currentWorkerMemory = workerMemory;
                            if (chiefWorker && container.getId().toString().equals(chiefWorkerContainerId)) {
                                currentWorkerMemory = chiefWorkerMemory;
                            }
                            if (tfEvaluator && container.getId().toString().equals(tfEvaluatorContainerId)) {
                                currentWorkerMemory = evaluatorWorkerMemory;
                            }
                            if (status != null
                                    && status.toString().equalsIgnoreCase("SUCCEEDED")
                                    && (cpuMemUsagedMax / (currentWorkerMemory / 1024.0))
                                            < conf.getDouble(
                                                    HboxConfiguration.HBOX_CONTAINER_MEM_USAGE_WARN_FRACTION,
                                                    HboxConfiguration.DEFAULT_HBOX_CONTAINER_MEM_USAGE_WARN_FRACTION)) {
                                usageStatistics.add("true");
                            } else {
                                usageStatistics.add("false");
                            }
                        } else {
                            usageStatistics.add("-"); // container info schema idx=14
                        }

                        if (containersAppStartTime.get(currentContainerID) != null
                                && !containersAppStartTime
                                        .get(currentContainerID)
                                        .equals("")) {
                            String localStartTime = containersAppStartTime.get(currentContainerID);
                            containerMessage.add(localStartTime);
                        } else {
                            containerMessage.add("N/A");
                        }
                        if (containersAppFinishTime.get(currentContainerID) != null
                                && !containersAppFinishTime
                                        .get(currentContainerID)
                                        .equals("")) {
                            String localFinishTime = containersAppFinishTime.get(currentContainerID);
                            containerMessage.add(localFinishTime);
                        } else {
                            containerMessage.add("N/A");
                        }
                        if (reporterProgress.get(currentContainerID) != null
                                && !reporterProgress.get(currentContainerID).equals("")) {
                            String progressLog = reporterProgress.get(currentContainerID);
                            String[] progress = progressLog.toString().split(":");
                            if (progress.length != 2) {
                                containerMessage.add("progress log format error");
                            } else {
                                try {
                                    Float percentProgress = Float.parseFloat(progress[1]);
                                    if (percentProgress < 0.0 || percentProgress > 1.0) {
                                        containerMessage.add("progress log format error");
                                    } else {
                                        DecimalFormat df = new DecimalFormat("0.00");
                                        df.setRoundingMode(RoundingMode.HALF_UP);
                                        containerMessage.add(df.format((Float.parseFloat(progress[1]) * 100)) + "%");
                                        // containerMessage.add(Float.toString((Float.parseFloat(progress[1])*100)) +
                                        // "%");
                                    }
                                } catch (Exception e) {
                                    containerMessage.add("progress log format error");
                                }
                            }
                            // containerMessage.add(Float.toString((Float.parseFloat(progress[1])*100)) + "%");
                        } else {
                            containerMessage.add("0.00%");
                        }
                        containerMessage.add(String.format(
                                "http://%s/node/containerlogs/%s/%s",
                                container.getNodeHttpAddress(),
                                container.getId().toString(),
                                userName));
                        containerMessage.addAll(usageStatistics);
                        containerMessage.add(rank); // container info schema idx=15
                        logMessage.put(container.getId().toString(), containerMessage);
                    }

                    int psIdx = 0;
                    for (Container container : psContainers) {
                        List<String> containerMessage = new ArrayList<>();
                        containerMessage.add(container.getNodeHttpAddress());
                        HboxContainerId currentContainerID = new HboxContainerId(container.getId());
                        String rank = "-";
                        if (applicationContext.getContainerGPUDevice(currentContainerID) != null) {
                            if (applicationContext
                                            .getContainerGPUDevice(currentContainerID)
                                            .trim()
                                            .length()
                                    != 0) {
                                containerMessage.add(applicationContext.getContainerGPUDevice(currentContainerID));
                            } else {
                                containerMessage.add("-");
                            }
                        } else {
                            containerMessage.add("-");
                        }
                        if (hboxAppType.equals("TENSORFLOW") || "TENSOR2TENSOR".equals(hboxAppType)) {
                            containerMessage.add(HboxConstants.PS);
                            rank = "" + psIdx++;
                        } else if (hboxAppType.equals("MXNET")
                                || hboxAppType.equals("DISTLIGHTLDA")
                                || hboxAppType.equals("XFLOW")) {
                            containerMessage.add(HboxConstants.SERVER);
                        } else if (hboxAppType.equals("XDL")) {
                            if (currentContainerID.toString().equals(schedulerContainerId)) {
                                containerMessage.add(HboxConstants.SCHEDULER);
                            } else {
                                containerMessage.add(HboxConstants.PS);
                                rank = "" + psIdx++;
                            }
                        } else {
                            containerMessage.add(HboxConstants.PS);
                            rank = "" + psIdx++;
                        }
                        HboxContainerStatus status = applicationContext.getContainerStatus(currentContainerID);
                        if (status != null) {
                            containerMessage.add(status.toString());
                        } else {
                            containerMessage.add("-");
                        }
                        List<String> usageStatistics = new ArrayList<>();
                        ConcurrentHashMap<String, LinkedBlockingDeque<Object>> cpuMetrics =
                                applicationContext.getContainersCpuMetrics().get(currentContainerID);
                        containerMessage.add(new Gson().toJson(cpuMetrics));
                        ConcurrentHashMap<String, List<Double>> cpuStatistics =
                                applicationContext.getContainersCpuStatistics().get(currentContainerID);
                        usageStatistics.add(new Gson().toJson(cpuStatistics));

                        if (psGCores > 0) {
                            ConcurrentHashMap<String, LinkedBlockingDeque<List<Long>>> containersGpuMemMetrics =
                                    applicationContext
                                            .getContainersGpuMemMetrics()
                                            .get(currentContainerID);
                            ConcurrentHashMap<String, LinkedBlockingDeque<List<Long>>> containersGpuUtilMetrics =
                                    applicationContext
                                            .getContainersGpuUtilMetrics()
                                            .get(currentContainerID);
                            containerMessage.add(new Gson().toJson(containersGpuMemMetrics));
                            containerMessage.add(new Gson().toJson(containersGpuUtilMetrics));
                            ConcurrentHashMap<String, List<Double>> gpuMemStatistics = applicationContext
                                    .getContainersGpuMemStatistics()
                                    .get(currentContainerID);
                            ConcurrentHashMap<String, List<Double>> gpuUtilStatistics = applicationContext
                                    .getContainersGpuUtilStatistics()
                                    .get(currentContainerID);
                            usageStatistics.add(new Gson().toJson(gpuMemStatistics));
                            usageStatistics.add(new Gson().toJson(gpuUtilStatistics));
                        } else {
                            containerMessage.add("-");
                            containerMessage.add("-");
                            usageStatistics.add("-");
                            usageStatistics.add("-");
                        }

                        if (cpuStatistics.size() != 0) {
                            Double cpuMemUsagedMax = cpuStatistics.get("CPUMEM").get(1);
                            if (status != null
                                    && status.toString().equalsIgnoreCase("SUCCEEDED")
                                    && (cpuMemUsagedMax / (psMemory / 1024.0))
                                            < conf.getDouble(
                                                    HboxConfiguration.HBOX_CONTAINER_MEM_USAGE_WARN_FRACTION,
                                                    HboxConfiguration.DEFAULT_HBOX_CONTAINER_MEM_USAGE_WARN_FRACTION)) {
                                usageStatistics.add("true");
                            } else {
                                usageStatistics.add("false");
                            }
                        } else {
                            usageStatistics.add("-"); // container info schema idx=14
                        }

                        if (containersAppStartTime.get(currentContainerID) != null
                                && !containersAppStartTime
                                        .get(currentContainerID)
                                        .equals("")) {
                            String localStartTime = containersAppStartTime.get(currentContainerID);
                            containerMessage.add(localStartTime);
                        } else {
                            containerMessage.add("N/A");
                        }
                        if (containersAppFinishTime.get(currentContainerID) != null
                                && !containersAppFinishTime
                                        .get(currentContainerID)
                                        .equals("")) {
                            String localFinishTime = containersAppFinishTime.get(currentContainerID);
                            containerMessage.add(localFinishTime);
                        } else {
                            containerMessage.add("N/A");
                        }
                        containerMessage.add("0.00%");
                        containerMessage.add(String.format(
                                "http://%s/node/containerlogs/%s/%s",
                                container.getNodeHttpAddress(),
                                container.getId().toString(),
                                userName));
                        containerMessage.addAll(usageStatistics);
                        containerMessage.add(rank); // container info schema idx=15
                        logMessage.put(container.getId().toString(), containerMessage);
                    }

                    List<String> savedTimeStamp = new ArrayList<>();
                    List<String> outputList = new ArrayList<>();
                    if (applicationContext.getOutputs().size() == 0) {
                        outputList.add("-");
                        savedTimeStamp.add("-");
                    } else {
                        for (OutputInfo output : applicationContext.getOutputs()) {
                            outputList.add(output.getDfsLocation());
                        }
                        if (applicationContext.getModelSavingList().size() == 0) {
                            savedTimeStamp.add("-");
                        } else {
                            for (int i = applicationContext.getModelSavingList().size(); i > 0; i--) {
                                savedTimeStamp.add(String.valueOf(
                                        applicationContext.getModelSavingList().get(i - 1)));
                            }
                        }
                    }

                    logMessage.put("input", inputList);
                    logMessage.put("savedTimeStamp", savedTimeStamp);
                    logMessage.put("output", outputList);
                    logMessage.put("workerGcores", Arrays.asList(String.valueOf(workerGCores)));
                    logMessage.put("workerNums", Arrays.asList(String.valueOf(workerNum)));
                    logMessage.put("workerVCores", Arrays.asList(String.valueOf(workerVCores)));
                    logMessage.put("workerMemory", Arrays.asList(String.format("%.2f", workerMemory / 1024.0)));
                    logMessage.put("psGcores", Arrays.asList(String.valueOf(psGCores)));
                    logMessage.put("psNums", Arrays.asList(String.valueOf(psNum)));
                    logMessage.put("psVCores", Arrays.asList(String.valueOf(psVCores)));
                    logMessage.put("psMemory", Arrays.asList(String.format("%.2f", psMemory / 1024.0)));
                    logMessage.put("hboxVersion", Arrays.asList(HboxVersion.VERSION));
                    logMessage.put(
                            "queue",
                            Arrays.asList(conf.get(
                                    HboxConfiguration.HBOX_APP_QUEUE, HboxConfiguration.DEFAULT_HBOX_APP_QUEUE)));
                    logMessage.put(
                            "user", Arrays.asList(conf.get("hadoop.job.ugi").split(",")[0]));
                    if (chiefWorker) {
                        logMessage.put(
                                "chiefWorkerMemory", Arrays.asList(String.format("%.2f", chiefWorkerMemory / 1024.0)));
                    }
                    if (tfEvaluator) {
                        logMessage.put(
                                "evaluatorWorkerMemory",
                                Arrays.asList(String.format("%.2f", evaluatorWorkerMemory / 1024.0)));
                    }

                    // application master logs
                    final List<String> amLogUrls = new ArrayList<>();
                    if (null != regResp && null != regResp.getContainersFromPreviousAttempts()) {
                        for (final Container container : regResp.getContainersFromPreviousAttempts()) {
                            amLogUrls.add(String.format(
                                    "http://%s/node/containerlogs/%s/%s",
                                    container.getNodeHttpAddress(),
                                    container.getId().toString(),
                                    userName));
                        }
                    }
                    if (null != applicationMasterHostname && null != applicationMasterHttpPortString) {
                        amLogUrls.add(String.format(
                                "http://%s:%s/node/containerlogs/%s/%s",
                                applicationMasterHostname, applicationMasterHttpPortString, amContainerId, userName));
                    }
                    if (amLogUrls.size() > 0) {
                        logMessage.put("applicationMaster", amLogUrls);
                    }

                    out.writeBytes(new Gson().toJson(logMessage));
                    out.close();
                    fs.close();
                    LOG.info("Writing the history log file Success.");
                } catch (Exception e) {
                    LOG.info("Writing the history log file Error." + e);
                }
            }
        });
        Runtime.getRuntime().addShutdownHook(cleanApplication);
    }
Constructor, connect to Resource Manager

@throws IOException
core/src/main/java/net/qihoo/hbox/AM/ApplicationMaster.java