public void updateEtlStatus() throws Exception {
        if (!checkState(JobState.ETL)) {
            return;
        }

        // get etl status
        SparkEtlJobHandler handler = new SparkEtlJobHandler();
        EtlStatus status =
                handler.getEtlJobStatus(sparkLoadAppHandle, appId, id, etlOutputPath, sparkResource, brokerDesc);
        writeLock();
        try {
            switch (status.getState()) {
                case RUNNING:
                    unprotectedUpdateEtlStatusInternal(status);
                    break;
                case FINISHED:
                    unprotectedProcessEtlFinish(status, handler);
                    break;
                case CANCELLED:
                    throw new LoadException("spark etl job failed. msg: " + status.getFailMsg());
                default:
                    LOG.warn("unknown etl state: {}", status.getState().name());
                    break;
            }
        } finally {
            writeUnlock();
        }

        if (checkState(JobState.LOADING)) {
            // create and send push tasks
            submitPushTasks();
        }
    }
@Test(expected = LoadException.class)
    public void testUpdateEtlStatusCancelled(@Mocked GlobalStateMgr globalStateMgr, @Injectable String originStmt,
                                             @Mocked SparkEtlJobHandler handler) throws Exception {
        EtlStatus status = new EtlStatus();
        status.setState(TEtlState.CANCELLED);

        new Expectations() {
            {
                handler.getEtlJobStatus((SparkLoadAppHandle) any, appId, anyLong, etlOutputPath,
                        (SparkResource) any, (BrokerDesc) any);
                result = status;
            }
        };

        SparkLoadJob job = getEtlStateJob(originStmt);
        job.updateEtlStatus();
    }