[Enhancement] Reset alter jobs to their last durable state on in-place leader demotion

AlterJobV2 state machines deliberately keep some progress in memory only: the
WAITING_TXN -> RUNNING transitions are not journaled ("DO NOT write edit log here,
tasks will be send again if FE restart or master changed") and dispatch state lives
in transient batch tasks / futures. That convention relied on the process restart at
leader switch to reload jobs from the image and implicitly discard the divergence.
With in-place demotion the same objects survive, so on same-node re-election a job
would dispatch by its unlogged in-memory state and wait on stale leader-session
dispatch state instead of re-sending work: runRunningJob polls a batch whose queue
entries were dropped by the demotion drain (wedge until timeout-cancel), re-entering
runWaitingTxnJob APPENDS into a stale batch (duplicates never get finish callbacks),
the fast-path null-guard skips the re-dispatch entirely, and OnlineOptimizeJobV2
could consume a stale INSERT future for a task whose INSERT never ran and durably
swap in an empty temp partition.

Introduce AlterJobV2.resetOnLeaderHandoff() ("demote = logical restart"): under the
job monitor, non-final jobs cancel and drop publishVersionFuture and delegate to a
per-class resetTransientStateForHandoff() that maps in-memory-only states back to
their last durable predecessor and recreates/clears leader-session transients:

- SchemaChangeJobV2 / RollupJobV2: RUNNING -> WAITING_TXN; fresh batch; latch/flags
  reset; discard an unlogged watershed on a durable-PENDING job.
- OptimizeJobV2: RUNNING -> WAITING_TXN; restore rewriteTasks/sourcePartitionNames/
  tmpPartitionNames/allPartitionOptimized/distributionInfo to the durable snapshot;
  undo a partial tmpPartitionIds append at PENDING. optimizeClause kept (a real
  reload nulls it and self-cancels; keeping it is strictly better).
- OnlineOptimizeJobV2: additionally cancel(true) + drop the in-flight INSERT future
  (stale-result cross-task attribution is data-loss grade) and clear the
  never-journaled double-write partition routing.
- LakeTableSchemaChangeJob / LakeRollupJob: RUNNING -> WAITING_TXN; dequeue + fresh
  batch. LakeRollupJob keeps whereClause (no gsonPostProcess restore exists; a real
  reload silently loses the sync-MV filter).
- LakeTableAlterMetaJobBase: RUNNING -> PENDING (WAITING_TXN is skipped on its live
  path; replay throws on RUNNING); drop the per-dispatch batch.
- LakeTableIndexFastPathJobBase: WAITING_TXN and RUNNING are both in-memory only ->
  PENDING; dequeue tasks and null the batch so the null-guard re-dispatches.

AlterHandler.onStopped() sweeps alterJobsV2 with the reset (worker joined = run-path
quiescent) and start() repeats the idempotent sweep before the first scheduling
cycle. MaterializedViewHandler.onStopped() also clears tableRunningJobMap, fixing a
pre-existing cross-leadership slot leak (a job finished and expiry-removed by an
interim leader is removed from alterJobsV2 by replay but never from the gate,
permanently suspending new rollups on that table).

Add LeaderHandoffResetTest covering the state mapping, transient cleanup, PENDING
partial-rollback, final-state immutability, publish-future handling, and the handler
sweep. Some transient fields become package-private for these tests (no reflection).

Signed-off-by: gengjun-git <gengjun@starrocks.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: gengjun-git

fe/fe-core/src/main/java/com/starrocks/alter/AlterHandler.java

@@ -218,6 +218,10 @@ public synchronized void start() {
         if (executor == null || executor.isShutdown()) {
             executor = newExecutor();
         }
+        // Idempotent insurance for the onStopped() sweep: covers a stop path that never ran
+        // onStopped (e.g. a plain setStop() without join) and guarantees every job is reset
+        // before the first scheduling cycle can dispatch it by its in-memory state.
+        resetJobsOnLeaderHandoff();
         super.start();
     }
 
@@ -233,6 +237,21 @@ protected void onStopped() {
         if (executor != null && !executor.isShutdown()) {
             executor.shutdownNow();
         }
+        // The jobs themselves survive in memory across an in-place demote / re-elect cycle,
+        // unlike a restart which reloads them from the image/journal. Reset each non-final
+        // job to its last durable state (drop unlogged in-memory transitions and leader-
+        // session transients) so a re-elected leader resumes exactly like a restarted FE.
+        resetJobsOnLeaderHandoff();
+    }
+
+    private void resetJobsOnLeaderHandoff() {
+        for (AlterJobV2 job : alterJobsV2.values()) {
+            try {
+                job.resetOnLeaderHandoff();
+            } catch (Throwable t) {
+                LOG.warn("reset alter job {} on leader handoff failed", job.getJobId(), t);
+            }
+        }
     }
 
     /*

fe/fe-core/src/main/java/com/starrocks/alter/AlterJobV2.java

@@ -233,6 +233,39 @@ public void setFinishedTimeMs(long finishedTimeMs) {
         this.finishedTimeMs = finishedTimeMs;
     }
 
+    /**
+     * Reset this job's in-memory state to its last durable (journaled) equivalent when the
+     * leader demotes in place. Historically "FE restart or master changed" reloaded jobs from
+     * the image/journal, which implicitly discarded in-memory-only progress (the deliberately
+     * unlogged WAITING_TXN -> RUNNING transitions) and leader-session transients (batch
+     * tasks, latches, futures). In-place demotion keeps the very same objects alive, so this
+     * hook performs that normalization explicitly: on re-election the job re-enters its last
+     * durable state and re-sends its work from scratch, exactly like a restarted FE.
+     *
+     * Synchronized on the job: run() and cancel() use the same monitor, so a straggling user
+     * cancel or a finish-callback thread surviving the executor's shutdownNow() cannot
+     * interleave with the reset. Final states are left untouched. Must NOT write to the
+     * journal - it is already sealed when the demotion drain invokes this.
+     */
+    public synchronized void resetOnLeaderHandoff() {
+        if (jobState.isFinalState()) {
+            return;
+        }
+        if (publishVersionFuture != null) {
+            publishVersionFuture.cancel(false);
+            publishVersionFuture = null;
+        }
+        resetTransientStateForHandoff();
+    }
+
+    /**
+     * Subclass hook for {@link #resetOnLeaderHandoff()}: map in-memory-only job states back
+     * to their last durable predecessor and recreate/clear leader-session transients (batch
+     * tasks, latches, flags, futures). Runs under the job monitor with the journal sealed.
+     */
+    protected void resetTransientStateForHandoff() {
+    }
+
     public void setComputeResource(ComputeResource computeResource) {
         this.computeResource = computeResource;
         this.warehouseId = computeResource.getWarehouseId();

fe/fe-core/src/main/java/com/starrocks/alter/LakeRollupJob.java

@@ -126,6 +126,22 @@ public class LakeRollupJob extends LakeTableSchemaChangeJobBase {
     // save all create rollup tasks
     protected AgentBatchTask rollupBatchTask = new AgentBatchTask();
 
+    @Override
+    protected void resetTransientStateForHandoff() {
+        // WAITING_TXN -> RUNNING is deliberately not journaled; map it back so the re-elected
+        // leader re-enters runWaitingTxnJob and re-sends the tasks.
+        if (jobState == JobState.RUNNING) {
+            jobState = JobState.WAITING_TXN;
+        }
+        // Same normalization replay() performs: drop queue entries and start from an empty
+        // batch - runWaitingTxnJob appends directly to the field (double-add hazard).
+        AgentTaskQueue.removeBatchTask(rollupBatchTask, TTaskType.ALTER);
+        rollupBatchTask = new AgentBatchTask();
+        // whereClause deliberately KEPT: this class has no gsonPostProcess restore, so a real
+        // reload silently loses the sync-MV filter (pre-existing reload bug); keep the
+        // strictly-better in-memory value.
+    }
+
     public LakeRollupJob(long jobId, long dbId, long tableId, String tableName, long timeoutMs,
                          long baseIndexMetaId, long rollupIndexMetaId, String baseIndexName, String rollupIndexName,
                          int rollupSchemaVersion, List<Column> rollupSchema, Expr whereClause, int baseSchemaHash,

fe/fe-core/src/main/java/com/starrocks/alter/LakeTableAlterMetaJobBase.java

@@ -75,13 +75,27 @@ public abstract class LakeTableAlterMetaJobBase extends AlterJobV2 {
     private Table<Long, Long, MaterializedIndex> physicalPartitionIndexMap = HashBasedTable.create();
     @SerializedName(value = "commitVersionMap")
     private Map<Long, Long> commitVersionMap = new HashMap<>();
-    private AgentBatchTask batchTask = null;
+    // Package-private so same-package tests can verify the leader-handoff reset without reflection.
+    AgentBatchTask batchTask = null;
     private boolean isFileBundling = false;
 
     public LakeTableAlterMetaJobBase(JobType jobType) {
         super(jobType);
     }
 
+    @Override
+    protected void resetTransientStateForHandoff() {
+        // WAITING_TXN is skipped on the live path; RUNNING is the only in-memory-only state
+        // and its durable predecessor is PENDING - the same-state re-log in runPendingJob
+        // already persisted the watershed, and the -1 guard prevents re-allocation on re-run.
+        // replay() throws on RUNNING, so it must never leak into any persisted copy.
+        if (jobState == JobState.RUNNING) {
+            jobState = JobState.PENDING;
+        }
+        // Recreated fresh per dispatch; a stale value only skews SHOW ALTER progress.
+        batchTask = null;
+    }
+
     public LakeTableAlterMetaJobBase(long jobId, JobType jobType, long dbId, long tableId,
                                      String tableName, long timeoutMs) {
         super(jobId, jobType, dbId, tableId, tableName, timeoutMs);

fe/fe-core/src/main/java/com/starrocks/alter/LakeTableIndexFastPathJobBase.java

@@ -122,6 +122,25 @@ protected LakeTableIndexFastPathJobBase(JobType type) {
         super(type);
     }
 
+    @Override
+    protected void resetTransientStateForHandoff() {
+        // BOTH WAITING_TXN and RUNNING are in-memory only for this job family; PENDING is the
+        // only durable predecessor (its same-state re-log carries the watershed and the
+        // tablet snapshot, and the watershedTxnId != -1 guard makes the re-run idempotent).
+        if (jobState == JobState.WAITING_TXN || jobState == JobState.RUNNING) {
+            jobState = JobState.PENDING;
+        }
+        // runRunningJob dispatches ONLY behind a null guard - a stale batch would silently
+        // suppress the re-send and wedge the job until timeout. Drop queue entries (mirror
+        // cancelImpl) and null the field so the re-run re-dispatches.
+        if (batchTask != null) {
+            for (AgentTask task : batchTask.getAllTasks()) {
+                AgentTaskQueue.removeTask(task.getBackendId(), TTaskType.ALTER, task.getSignature());
+            }
+            batchTask = null;
+        }
+    }
+
     protected LakeTableIndexFastPathJobBase(long jobId, JobType jobType, long dbId, long tableId, String tableName,
                                             long timeoutMs) {
         super(jobId, jobType, dbId, tableId, tableName, timeoutMs);

fe/fe-core/src/main/java/com/starrocks/alter/LakeTableSchemaChangeJob.java

@@ -158,7 +158,8 @@ public class LakeTableSchemaChangeJob extends LakeTableSchemaChangeJobBase {
     private List<Integer> sortKeyUniqueIds;
 
     // save all schema change tasks
-    private AgentBatchTask schemaChangeBatchTask = new AgentBatchTask();
+    // Package-private so same-package tests can verify the leader-handoff reset without reflection.
+    AgentBatchTask schemaChangeBatchTask = new AgentBatchTask();
 
     // runtime variable for synchronization between cancel and runPendingJob
     private MarkedCountDownLatch<Long, Long> createReplicaLatch = null;
@@ -173,6 +174,24 @@ public LakeTableSchemaChangeJob() {
         super(JobType.SCHEMA_CHANGE);
     }
 
+    @Override
+    protected void resetTransientStateForHandoff() {
+        // WAITING_TXN -> RUNNING is deliberately not journaled; map it back so the re-elected
+        // leader re-verifies the watershed and re-sends every AlterReplicaTask.
+        if (jobState == JobState.RUNNING) {
+            jobState = JobState.WAITING_TXN;
+        }
+        // Mirror cancelImpl's queue cleanup, then start from an empty batch: the WAITING_TXN
+        // handler APPENDS to it (double-add hazard), and getInfo dereferences the field, so
+        // fresh-empty rather than null. watershedTxnId/Gtid stay - they are durable with the
+        // WAITING_TXN entry, and runPendingJob reassigns them unconditionally at PENDING.
+        AgentTaskQueue.removeBatchTask(schemaChangeBatchTask, TTaskType.ALTER);
+        schemaChangeBatchTask = new AgentBatchTask();
+        createReplicaLatch = null;
+        waitingCreatingReplica.set(false);
+        isCancelling.set(false);
+    }
+
     public LakeTableSchemaChangeJob(long jobId, long dbId, long tableId, String tableName, long timeoutMs) {
         super(jobId, JobType.SCHEMA_CHANGE, dbId, tableId, tableName, timeoutMs);
     }

fe/fe-core/src/main/java/com/starrocks/alter/MaterializedViewHandler.java

@@ -122,6 +122,21 @@ public MaterializedViewHandler() {
     // table id -> set of running job ids
     private final Map<Long, Set<Long>> tableRunningJobMap = new ConcurrentHashMap<>();
 
+    @Override
+    protected void onStopped() {
+        super.onStopped();
+        // Leader-only concurrency gate, fully repopulated by runAlterJobV2 on the next
+        // leadership. Clearing it also fixes a slot leak: a job that finished and was
+        // expiry-removed by another leader while this node was a follower is removed from
+        // alterJobsV2 by replay but never from this map, permanently consuming a per-table
+        // slot and suspending new rollups on that table.
+        // tableNotFinalStateJobMap is deliberately NOT cleared: replay maintains it, and it
+        // drives the ROLLUP -> NORMAL table-state restore for jobs finished via replay.
+        synchronized (tableRunningJobMap) {
+            tableRunningJobMap.clear();
+        }
+    }
+
     @Override
     public void addAlterJobV2(AlterJobV2 alterJob) {
         super.addAlterJobV2(alterJob);

fe/fe-core/src/main/java/com/starrocks/alter/OnlineOptimizeJobV2.java

@@ -85,7 +85,8 @@ public class OnlineOptimizeJobV2 extends AlterJobV2 implements GsonPostProcessab
 
     private static final ExecutorService EXECUTOR = Executors.newCachedThreadPool();
 
-    private Future<Constants.TaskRunState> future = null;
+    // Package-private so same-package tests can verify the leader-handoff reset without reflection.
+    Future<Constants.TaskRunState> future = null;
 
     @SerializedName(value = "tmpPartitionIds")
     private List<Long> tmpPartitionIds = Lists.newArrayList();
@@ -96,7 +97,8 @@ public class OnlineOptimizeJobV2 extends AlterJobV2 implements GsonPostProcessab
     private Map<String, String> properties = Maps.newHashMap();
 
     @SerializedName(value = "rewriteTasks")
-    private List<OptimizeTask> rewriteTasks = Lists.newArrayList();
+    // Package-private so same-package tests can verify the leader-handoff reset without reflection.
+    List<OptimizeTask> rewriteTasks = Lists.newArrayList();
     private int progress = 0;
 
     @SerializedName(value = "sourcePartitionNames")
@@ -144,6 +146,45 @@ protected OnlineOptimizeJobV2(OnlineOptimizeJobV2 job) {
         this.optimizeOperation = job.optimizeOperation;
     }
 
+    @Override
+    protected void resetTransientStateForHandoff() {
+        if (jobState == JobState.RUNNING) {
+            jobState = JobState.WAITING_TXN;
+        }
+        // The single in-flight INSERT future is attributed to "the task currently being
+        // processed" by position, not identity: a stale result from the previous leader
+        // session would be consumed by a NEW task whose INSERT never ran, and its EMPTY temp
+        // partition would be durably swapped in. Cancel hard and drop it.
+        if (future != null) {
+            future.cancel(true);
+            future = null;
+        }
+        // runWaitingTxnJob APPENDS to rewriteTasks; watershedTxnId is reallocated per task
+        // during RUNNING and its durable WAITING_TXN value is -1.
+        rewriteTasks.clear();
+        watershedTxnId = -1;
+        progress = 0;
+        if (jobState == JobState.PENDING) {
+            tmpPartitionIds.clear();
+            allPartitionOptimized = false;
+        }
+        // Drop the never-journaled double-write routing so the demoted node behaves like a
+        // freshly loaded FE (best-effort: db/table may already be gone).
+        try {
+            Database db = GlobalStateMgr.getCurrentState().getLocalMetastore().getDb(dbId);
+            if (db != null) {
+                OlapTable tbl = (OlapTable) GlobalStateMgr.getCurrentState()
+                        .getLocalMetastore().getTable(db.getId(), tableId);
+                if (tbl != null) {
+                    disableDoubleWritePartition(db, tbl);
+                }
+            }
+        } catch (Throwable t) {
+            LOG.warn("clear double-write partitions failed on leader handoff, job: {}", jobId, t);
+        }
+        // optimizeClause deliberately kept - see OptimizeJobV2.resetTransientStateForHandoff().
+    }
+
     public List<Long> getTmpPartitionIds() {
         return tmpPartitionIds;
     }

fe/fe-core/src/main/java/com/starrocks/alter/OptimizeJobV2.java

@@ -87,7 +87,8 @@ public class OptimizeJobV2 extends AlterJobV2 implements GsonPostProcessable {
     private Map<String, String> properties = Maps.newHashMap();
 
     @SerializedName(value = "rewriteTasks")
-    private List<OptimizeTask> rewriteTasks = Lists.newArrayList();
+    // Package-private so same-package tests can verify the leader-handoff reset without reflection.
+    List<OptimizeTask> rewriteTasks = Lists.newArrayList();
     private int progress = 0;
 
     @SerializedName(value = "sourcePartitionNames")
@@ -135,6 +136,35 @@ protected OptimizeJobV2(OptimizeJobV2 job) {
         this.optimizeOperation = job.optimizeOperation;
     }
 
+    @Override
+    protected void resetTransientStateForHandoff() {
+        // WAITING_TXN -> RUNNING is deliberately not journaled; map it back so the re-elected
+        // leader re-enters runWaitingTxnJob (rebuild + re-register the rewrite tasks).
+        if (jobState == JobState.RUNNING) {
+            jobState = JobState.WAITING_TXN;
+        }
+        // Restore serialized-but-diverged fields to their durable WAITING_TXN snapshot: the
+        // journal copy was written before any rewrite task or finish bookkeeping existed, and
+        // runWaitingTxnJob APPENDS to rewriteTasks (a stale list would double the tasks).
+        rewriteTasks.clear();
+        sourcePartitionNames.clear();
+        tmpPartitionNames.clear();
+        allPartitionOptimized = false;
+        distributionInfo = null;
+        progress = 0;
+        if (jobState == JobState.PENDING) {
+            // Undo a partially executed runPendingJob: a fenced attempt may have appended tmp
+            // partition ids that the durable PENDING image does not have; re-running with the
+            // stale list would pair 2N tmp partitions against N sources.
+            tmpPartitionIds.clear();
+            watershedTxnId = -1;
+        }
+        // optimizeClause is deliberately KEPT: a real reload nulls it and self-cancels the
+        // job; keeping it is safe (every derived value is recomputed on re-run) and strictly
+        // better. A mid-RUNNING handoff still ends in a clean cancel via the TaskManager task
+        // name collision - identical to genuine-restart behavior.
+    }
+
     public List<Long> getTmpPartitionIds() {
         return tmpPartitionIds;
     }

fe/fe-core/src/main/java/com/starrocks/alter/RollupJobV2.java

@@ -175,7 +175,8 @@ public class RollupJobV2 extends AlterJobV2 implements GsonPostProcessable {
     private Expr whereClause;
 
     // save all create rollup tasks
-    private AgentBatchTask rollupBatchTask = new AgentBatchTask();
+    // Package-private so same-package tests can verify the leader-handoff reset without reflection.
+    AgentBatchTask rollupBatchTask = new AgentBatchTask();
 
     // runtime variable for synchronization between cancel and runPendingJob
     private MarkedCountDownLatch<Long, Long> createReplicaLatch = null;
@@ -187,6 +188,26 @@ public RollupJobV2() {
         super(JobType.ROLLUP);
     }
 
+    @Override
+    protected void resetTransientStateForHandoff() {
+        // WAITING_TXN -> RUNNING is deliberately not journaled; map it back so the re-elected
+        // leader re-enters runWaitingTxnJob and re-sends every AlterReplicaTask.
+        if (jobState == JobState.RUNNING) {
+            jobState = JobState.WAITING_TXN;
+        }
+        // runWaitingTxnJob APPENDS to the batch - see SchemaChangeJobV2 for the double-add hazard.
+        rollupBatchTask = new AgentBatchTask();
+        createReplicaLatch = null;
+        waitingCreatingReplica.set(false);
+        isCancelling.set(false);
+        if (jobState == JobState.PENDING) {
+            watershedTxnId = -1;
+        }
+        // whereClause is deliberately KEPT: gsonPostProcess only restores it for PENDING jobs,
+        // so a true reload would silently lose the sync-MV filter (pre-existing reload bug);
+        // the surviving in-memory value is strictly better and every derived value is recomputed.
+    }
+
     public RollupJobV2(long jobId, long dbId, long tableId, String tableName, long timeoutMs,
                        long baseIndexMetaId, long rollupIndexMetaId, String baseIndexName, String rollupIndexName,
                        int rollupSchemaVersion, List<Column> rollupSchema, Expr whereClause, int baseSchemaHash,