@Override
    void executeTask() throws LoadException {
        LOG.info("begin to execute spark pending task. load job id: {}", loadJobId);
        submitEtlJob();
    }
@Test
    public void testExecuteTask(@Injectable SparkLoadJob sparkLoadJob,
                                @Injectable SparkResource resource,
                                @Injectable BrokerDesc brokerDesc,
                                @Mocked GlobalStateMgr globalStateMgr,
                                @Injectable SparkLoadAppHandle handle,
                                @Injectable Database database,
                                @Injectable OlapTable table) throws LoadException {
        long dbId = 0L;
        long tableId = 1L;

        // columns
        List<Column> columns = Lists.newArrayList();
        columns.add(new Column("c1", Type.BIGINT, true, null, false, null, ""));
        Map<ColumnId, Column> idToColumn = Maps.newTreeMap(ColumnId.CASE_INSENSITIVE_ORDER);
        idToColumn.put(columns.get(0).getColumnId(), columns.get(0));

        // indexes
        Map<Long, List<Column>> indexIdToSchema = Maps.newHashMap();
        long indexId = 3L;
        indexIdToSchema.put(indexId, columns);

        // partition and distribution infos
        long partitionId = 2L;
        DistributionInfo distributionInfo = new HashDistributionInfo(2, Lists.newArrayList(columns.get(0)));
        PartitionInfo partitionInfo = new SinglePartitionInfo();
        Partition partition = new Partition(partitionId, "p1", null, distributionInfo);
        List<Partition> partitions = Lists.newArrayList(partition);

        // file group
        Map<BrokerFileGroupAggInfo.FileGroupAggKey, List<BrokerFileGroup>> aggKeyToFileGroups = Maps.newHashMap();
        List<BrokerFileGroup> brokerFileGroups = Lists.newArrayList();
        DataDescription desc = new DataDescription("testTable", null, Lists.newArrayList("abc.txt"),
                null, null, null, null, false, null);
        BrokerFileGroup brokerFileGroup = new BrokerFileGroup(desc);
        brokerFileGroups.add(brokerFileGroup);
        BrokerFileGroupAggInfo.FileGroupAggKey aggKey = new BrokerFileGroupAggInfo.FileGroupAggKey(tableId, null);
        aggKeyToFileGroups.put(aggKey, brokerFileGroups);

        new Expectations() {
            {
                globalStateMgr.getDb(dbId);
                result = database;
                sparkLoadJob.getHandle();
                result = handle;
                database.getTable(tableId);
                result = table;
                table.getPartitions();
                result = partitions;
                table.getIndexIdToSchema();
                result = indexIdToSchema;
                table.getDefaultDistributionInfo();
                result = distributionInfo;
                table.getSchemaHashByIndexId(indexId);
                result = 123;
                table.getPartitionInfo();
                result = partitionInfo;
                table.getPartition(partitionId);
                result = partition;
                table.getKeysTypeByIndexId(indexId);
                result = KeysType.DUP_KEYS;
                table.getBaseIndexId();
                result = indexId;
                table.getIdToColumn();
                result = idToColumn;
            }
        };

        String appId = "application_15888888888_0088";
        new MockUp<SparkEtlJobHandler>() {
            @Mock
            public void submitEtlJob(long loadJobId, String loadLabel, EtlJobConfig etlJobConfig,
                                     SparkResource resource, BrokerDesc brokerDesc,
                                     SparkLoadAppHandle handle,
                                     SparkPendingTaskAttachment attachment,
                                     Long sparkLoadSubmitTimeout) throws LoadException {
                attachment.setAppId(appId);
            }
        };

        SparkLoadPendingTask task = new SparkLoadPendingTask(sparkLoadJob, aggKeyToFileGroups, resource, brokerDesc);
        task.init();
        SparkPendingTaskAttachment attachment = Deencapsulation.getField(task, "attachment");
        Assert.assertEquals(null, attachment.getAppId());
        task.executeTask();
        Assert.assertEquals(appId, attachment.getAppId());
    }