public void submitEtlJob(long loadJobId, String loadLabel, EtlJobConfig etlJobConfig, SparkResource resource,
                             BrokerDesc brokerDesc, SparkLoadAppHandle handle, SparkPendingTaskAttachment attachment,
                             Long sparkLoadSubmitTimeout)
            throws LoadException {
        // delete outputPath
        deleteEtlOutputPath(etlJobConfig.outputPath, brokerDesc);

        // init local dir
        if (!FeConstants.runningUnitTest) {
            initLocalDir();
        }

        // prepare dpp archive
        SparkRepository.SparkArchive archive = resource.prepareArchive();
        SparkRepository.SparkLibrary dppLibrary = archive.getDppLibrary();
        SparkRepository.SparkLibrary spark2xLibrary = archive.getSpark2xLibrary();

        // spark home
        String sparkHome = Config.spark_home_default_dir;
        // etl config path
        String configsHdfsDir = etlJobConfig.outputPath + "/" + JOB_CONFIG_DIR + "/";
        // etl config json path
        String jobConfigHdfsPath = configsHdfsDir + CONFIG_FILE_NAME;
        // spark submit app resource path
        String appResourceHdfsPath = dppLibrary.remotePath;
        // spark yarn archive path
        String jobArchiveHdfsPath = spark2xLibrary.remotePath;
        // spark yarn stage dir
        String jobStageHdfsPath = resource.getWorkingDir();
        // spark launcher log path
        String logFilePath = Config.spark_launcher_log_dir + "/" + String.format(LAUNCHER_LOG, loadJobId, loadLabel);

        // update archive and stage configs here
        Map<String, String> sparkConfigs = resource.getSparkConfigs();
        if (Strings.isNullOrEmpty(sparkConfigs.get("spark.yarn.archive"))) {
            sparkConfigs.put("spark.yarn.archive", jobArchiveHdfsPath);
        }
        if (Strings.isNullOrEmpty(sparkConfigs.get("spark.yarn.stage.dir"))) {
            sparkConfigs.put("spark.yarn.stage.dir", jobStageHdfsPath);
        }

        try {
            byte[] configData = etlJobConfig.configToJson().getBytes(StandardCharsets.UTF_8);
            if (brokerDesc.hasBroker()) {
                BrokerUtil.writeFile(configData, jobConfigHdfsPath, brokerDesc);
            } else {
                HdfsUtil.writeFile(configData, jobConfigHdfsPath, brokerDesc);
            }
        } catch (UserException e) {
            throw new LoadException(e.getMessage());
        }

        SparkLauncher launcher = new SparkLauncher();
        // master      |  deployMode
        // ------------|-------------
        // yarn        |  cluster
        // spark://xx  |  client
        launcher.setMaster(resource.getMaster())
                .setDeployMode(resource.getDeployMode().name().toLowerCase())
                .setAppResource(appResourceHdfsPath)
                .setMainClass(SparkEtlJob.class.getCanonicalName())
                .setAppName(String.format(ETL_JOB_NAME, loadLabel))
                .setSparkHome(sparkHome)
                .addAppArgs(jobConfigHdfsPath)
                .redirectError();

        // spark configs
        for (Map.Entry<String, String> entry : resource.getSparkConfigs().entrySet()) {
            launcher.setConf(entry.getKey(), entry.getValue());
        }

        // start app
        State state = null;
        String appId = null;
        String logPath = null;
        String errMsg = "start spark app failed. error: ";
        try {
            Process process = launcher.launch();
            handle.setProcess(process);
            if (!FeConstants.runningUnitTest) {
                SparkLauncherMonitor.LogMonitor logMonitor = SparkLauncherMonitor.createLogMonitor(handle);
                logMonitor.setSubmitTimeoutMs(sparkLoadSubmitTimeout);
                logMonitor.setRedirectLogPath(logFilePath);
                logMonitor.start();
                try {
                    logMonitor.join();
                } catch (InterruptedException e) {
                    logMonitor.interrupt();
                    throw new LoadException(errMsg + e.getMessage());
                }
            }
            appId = handle.getAppId();
            state = handle.getState();
            logPath = handle.getLogPath();
        } catch (IOException e) {
            LOG.warn(errMsg, e);
            throw new LoadException(errMsg + e.getMessage());
        }

        if (fromSparkState(state) == TEtlState.CANCELLED) {
            if (state == State.KILLED) {
                try {
                    killYarnApplication(appId, loadJobId, resource);
                } catch (UserException e) {
                    LOG.warn(errMsg, e);
                }
            }
            throw new LoadException(
                    errMsg + "spark app state: " + state.toString() + ", loadJobId:" + loadJobId + ", logPath:" +
                            logPath);
        }

        if (appId == null) {
            throw new LoadException(errMsg + "Waiting too much time to get appId from handle. spark app state: "
                    + state.toString() + ", loadJobId:" + loadJobId);
        }

        // success
        attachment.setAppId(appId);
        attachment.setHandle(handle);
    }
@Test(expected = LoadException.class)
    public void testSubmitEtlJobFailed(@Mocked BrokerUtil brokerUtil, @Mocked SparkLauncher launcher,
                                       @Injectable Process process,
                                       @Mocked SparkLoadAppHandle handle) throws IOException, LoadException {
        new Expectations() {
            {
                launcher.launch();
                result = process;
                handle.getAppId();
                result = appId;
                handle.getState();
                result = SparkLoadAppHandle.State.FAILED;
            }
        };

        EtlJobConfig etlJobConfig = new EtlJobConfig(Maps.newHashMap(), etlOutputPath, label, null);
        SparkResource resource = new SparkResource(resourceName);
        new Expectations(resource) {
            {
                resource.prepareArchive();
                result = archive;
            }
        };

        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");
        BrokerDesc brokerDesc = new BrokerDesc(broker, Maps.newHashMap());
        SparkPendingTaskAttachment attachment = new SparkPendingTaskAttachment(pendingTaskId);
        SparkEtlJobHandler handler = new SparkEtlJobHandler();
        long sparkLoadSubmitTimeout = Config.spark_load_submit_timeout_second;
        handler.submitEtlJob(loadJobId, label, etlJobConfig, resource, brokerDesc, handle, attachment, sparkLoadSubmitTimeout);
    }