private void onPendingTaskFinished(SparkPendingTaskAttachment attachment) {
        writeLock();
        try {
            // check if job has been cancelled
            if (isTxnDone()) {
                LOG.warn(new LogBuilder(LogKey.LOAD_JOB, id)
                        .add("state", state)
                        .add("error_msg", "this task will be ignored when job is: " + state)
                        .build());
                return;
            }

            if (finishedTaskIds.contains(attachment.getTaskId())) {
                LOG.warn(new LogBuilder(LogKey.LOAD_JOB, id)
                        .add("task_id", attachment.getTaskId())
                        .add("error_msg", "this is a duplicated callback of pending task "
                                + "when broker already has loading task")
                        .build());
                return;
            }

            // add task id into finishedTaskIds
            finishedTaskIds.add(attachment.getTaskId());

            sparkLoadAppHandle = attachment.getHandle();
            appId = attachment.getAppId();
            etlOutputPath = attachment.getOutputPath();

            executeEtl();
            // log etl state
            unprotectedLogUpdateStateInfo();
        } finally {
            writeUnlock();
        }
    }
@Test
    public void testOnPendingTaskFinished(@Mocked GlobalStateMgr globalStateMgr, @Injectable String originStmt)
            throws MetaNotFoundException {
        ResourceDesc resourceDesc = new ResourceDesc(resourceName, Maps.newHashMap());
        SparkLoadJob job = new SparkLoadJob(dbId, label, resourceDesc, new OriginStatement(originStmt, 0));
        SparkPendingTaskAttachment attachment = new SparkPendingTaskAttachment(pendingTaskId);
        attachment.setAppId(appId);
        attachment.setOutputPath(etlOutputPath);
        job.onTaskFinished(attachment);

        // check pending task finish
        Assert.assertTrue(job.finishedTaskIds.contains(pendingTaskId));
        Assert.assertEquals(appId, Deencapsulation.getField(job, "appId"));
        Assert.assertEquals(etlOutputPath, Deencapsulation.getField(job, "etlOutputPath"));
        Assert.assertEquals(JobState.ETL, job.getState());
    }