@Override
    protected void runPendingJob() throws AlterCancelException {
        Preconditions.checkState(jobState == JobState.PENDING, jobState);

        LOG.info("begin to send create temp partitions. job: {}", jobId);
        Database db = GlobalStateMgr.getCurrentState().getDb(dbId);
        if (db == null) {
            throw new AlterCancelException("Database " + dbId + " does not exist");
        }

        if (!checkTableStable(db)) {
            return;
        }

        if (optimizeClause == null) {
            throw new AlterCancelException("optimize clause is null since FE restart, job: " + jobId);
        }
        if (optimizeClause.isTableOptimize()) {
            allPartitionOptimized = true;
        }

        // 1. create temp partitions
        for (int i = 0; i < optimizeClause.getSourcePartitionIds().size(); ++i) {
            tmpPartitionIds.add(GlobalStateMgr.getCurrentState().getNextId());
        }

        long createPartitionStartTimestamp = System.currentTimeMillis();
        OlapTable targetTable = checkAndGetTable(db, tableId);
        try {
            PartitionUtils.createAndAddTempPartitionsForTable(db, targetTable, postfix,
                        optimizeClause.getSourcePartitionIds(), getTmpPartitionIds(), optimizeClause.getDistributionDesc(),
                        warehouseId);
            LOG.debug("create temp partitions {} success. job: {}", getTmpPartitionIds(), jobId);
        } catch (Exception e) {
            LOG.warn("create temp partitions failed", e);
            throw new AlterCancelException("create temp partitions failed " + e);
        }
        long createPartitionElapse = System.currentTimeMillis() - createPartitionStartTimestamp;

        // wait previous transactions finished
        this.jobState = JobState.WAITING_TXN;
        this.optimizeOperation = optimizeClause.toString();
        span.setAttribute("createPartitionElapse", createPartitionElapse);
        span.setAttribute("watershedTxnId", this.watershedTxnId);
        span.addEvent("setWaitingTxn");

        // write edit log
        GlobalStateMgr.getCurrentState().getEditLog().logAlterJob(this);
        LOG.info("transfer optimize job {} state to {}, watershed txn_id: {}", jobId, this.jobState, watershedTxnId);
    }
@Test
    public void testSchemaChangeWhileTabletNotStable() throws Exception {
        SchemaChangeHandler schemaChangeHandler = GlobalStateMgr.getCurrentState().getSchemaChangeHandler();
        Database db = GlobalStateMgr.getCurrentState().getDb(GlobalStateMgrTestUtil.testDb1);
        OlapTable olapTable = (OlapTable) db.getTable(GlobalStateMgrTestUtil.testTable7);
        Partition testPartition = olapTable.getPartition(GlobalStateMgrTestUtil.testTable7);

        schemaChangeHandler.process(alterTableStmt.getAlterClauseList(), db, olapTable);
        Map<Long, AlterJobV2> alterJobsV2 = schemaChangeHandler.getAlterJobsV2();
        Assert.assertEquals(1, alterJobsV2.size());
        OnlineOptimizeJobV2 optimizeJob = (OnlineOptimizeJobV2) alterJobsV2.values().stream().findAny().get();

        MaterializedIndex baseIndex = testPartition.getBaseIndex();
        LocalTablet baseTablet = (LocalTablet) baseIndex.getTablets().get(0);
        List<Replica> replicas = baseTablet.getImmutableReplicas();
        Replica replica1 = replicas.get(0);

        // runPendingJob
        replica1.setState(Replica.ReplicaState.DECOMMISSION);
        optimizeJob.runPendingJob();
        Assert.assertEquals(JobState.PENDING, optimizeJob.getJobState());

        // table is stable runPendingJob again
        replica1.setState(Replica.ReplicaState.NORMAL);
        optimizeJob.runPendingJob();
        Assert.assertEquals(JobState.WAITING_TXN, optimizeJob.getJobState());
    }