public EtlStatus getEtlJobStatus(SparkLoadAppHandle handle, String appId, long loadJobId, String etlOutputPath,
                                     SparkResource resource, BrokerDesc brokerDesc) throws UserException {
        EtlStatus status = new EtlStatus();

        Preconditions.checkState(appId != null && !appId.isEmpty());
        if (resource.isYarnMaster()) {
            // prepare yarn config
            String configDir = resource.prepareYarnConfig();
            // yarn client path
            String yarnClient = resource.getYarnClientPath();
            // command: yarn --config configDir application -status appId
            String yarnStatusCmd = String.format(YARN_STATUS_CMD, yarnClient, configDir, appId);
            LOG.info(yarnStatusCmd);
            String[] envp = {"LC_ALL=" + Config.locale, "JAVA_HOME=" + System.getProperty("java.home")};
            CommandResult result = Util.executeCommand(yarnStatusCmd, envp, EXEC_CMD_TIMEOUT_MS);
            if (result.getReturnCode() != 0) {
                String stderr = result.getStderr();
                // case application not exists
                if (stderr != null && stderr.contains("doesn't exist in RM")) {
                    LOG.warn("spark application not found. spark app id: {}, load job id: {}, stderr: {}",
                            appId, loadJobId, stderr);
                    status.setState(TEtlState.CANCELLED);
                    status.setFailMsg("spark application not found");
                    return status;
                }

                LOG.warn("yarn application status failed. spark app id: {}, load job id: {}, timeout: {}" +
                                ", return code: {}, stderr: {}, stdout: {}",
                        appId, loadJobId, EXEC_CMD_TIMEOUT_MS, result.getReturnCode(), stderr, result.getStdout());
                throw new LoadException("yarn application status failed. error: " + stderr);
            }
            ApplicationReport report = new YarnApplicationReport(result.getStdout()).getReport();
            LOG.info("yarn application -status {}. load job id: {}, output: {}, report: {}",
                    appId, loadJobId, result.getStdout(), report);
            YarnApplicationState state = report.getYarnApplicationState();
            FinalApplicationStatus faStatus = report.getFinalApplicationStatus();
            status.setState(fromYarnState(state, faStatus));
            if (status.getState() == TEtlState.CANCELLED) {
                if (state == YarnApplicationState.FINISHED) {
                    status.setFailMsg("spark app state: " + faStatus.toString());
                } else {
                    status.setFailMsg("yarn app state: " + state.toString());
                }
            }
            status.setTrackingUrl(handle.getUrl() != null ? handle.getUrl() : report.getTrackingUrl());
            status.setProgress((int) (report.getProgress() * 100));
        } else {
            // state from handle
            if (handle == null) {
                status.setFailMsg("spark app handle is null");
                status.setState(TEtlState.CANCELLED);
                return status;
            }

            State state = handle.getState();
            status.setState(fromSparkState(state));
            if (status.getState() == TEtlState.CANCELLED) {
                status.setFailMsg("spark app state: " + state.toString());
            }
            LOG.info("spark app id: {}, load job id: {}, app state: {}", appId, loadJobId, state);
        }

        if (status.getState() == TEtlState.FINISHED || status.getState() == TEtlState.CANCELLED) {
            // get dpp result
            String dppResultFilePath = EtlJobConfig.getDppResultFilePath(etlOutputPath);
            try {
                byte[] data;
                if (brokerDesc.hasBroker()) {
                    data = BrokerUtil.readFile(dppResultFilePath, brokerDesc);
                } else {
                    data = HdfsUtil.readFile(dppResultFilePath, brokerDesc);
                }
                String dppResultStr = new String(data, StandardCharsets.UTF_8);
                DppResult dppResult = new Gson().fromJson(dppResultStr, DppResult.class);
                if (dppResult != null) {
                    status.setDppResult(dppResult);
                    if (status.getState() == TEtlState.CANCELLED && !Strings.isNullOrEmpty(dppResult.failedReason)) {
                        status.setFailMsg(dppResult.failedReason);
                    }
                }
            } catch (UserException | JsonSyntaxException e) {
                LOG.warn("read broker file failed. path: {}", dppResultFilePath, e);
            }
        }

        return status;
    }
@Test(expected = TimeoutException.class)
    public void testGetEtlJobStatusTimeout(@Mocked BrokerUtil brokerUtil, @Mocked Util util,
                                           @Mocked SparkYarnConfigFiles sparkYarnConfigFiles,
                                           @Mocked SparkLoadAppHandle handle)
            throws IOException, UserException {

        new Expectations() {
            {
                sparkYarnConfigFiles.prepare();
                sparkYarnConfigFiles.getConfigDir();
                result = "./yarn_config";

                Util.executeCommand(anyString, (String[]) any, anyLong);
                minTimes = 0;
                result = new TimeoutException("get spark etl job status timeout");
            }
        };

        SparkResource resource = new SparkResource(resourceName);
        Map<String, String> sparkConfigs = resource.getSparkConfigs();
        sparkConfigs.put("spark.master", "yarn");
        sparkConfigs.put("spark.submit.deployMode", "cluster");
        sparkConfigs.put("spark.hadoop.yarn.resourcemanager.address", "127.0.0.1:9999");
        new Expectations(resource) {
            {
                resource.getYarnClientPath();
                result = Config.yarn_client_path;
            }
        };

        BrokerDesc brokerDesc = new BrokerDesc(broker, Maps.newHashMap());
        SparkEtlJobHandler handler = new SparkEtlJobHandler();
        handler.getEtlJobStatus(handle, appId, loadJobId, etlOutputPath, resource, brokerDesc);
    }