[Enhancement] Drop abandoned agent tasks from AgentTaskQueue on leader demotion

The demotion drain previously only failed the in-flight agent tasks' latches; the tasks
themselves stayed in AgentTaskQueue. A demoting/follower leader no longer processes the BE
reports that normally remove them, so they would leak (including latch-less tasks such as
drop/clone) across a demote/re-elect cycle. Clear the queue as part of the drain, mirroring
how the old System.exit model wiped it via process restart.

Rename AgentTaskQueue.failAllPendingWaiters -> abandonInFlightTasks (and the GlobalStateMgr
demotion stage failInFlightAgentTasks -> abandonInFlightAgentTasks) to reflect that it now
both fails latch waiters and drops all tasks. Safe to clear the whole queue because addTask()
rejects new tasks while demoting, so nothing live is discarded. AgentTaskTest asserts the
queue is emptied.

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/server/GlobalStateMgr.java

@@ -2149,7 +2149,7 @@ void executeLeaderDemotionStages(FrontendNodeType targetType) {
         LOG.info("leader demotion to {} starting", targetType);
         long startMs = System.currentTimeMillis();
         runDemotionStage("beginLeaderDemotion", () -> beginLeaderDemotion(targetType));
-        runDemotionStage("failInFlightAgentTasks", this::failInFlightAgentTasks);
+        runDemotionStage("abandonInFlightAgentTasks", this::abandonInFlightAgentTasks);
         runDemotionStage("sealJournalWriter", this::sealJournalWriter);
         runDemotionStage("stopLeaderOnlyDaemonThreads", this::stopLeaderOnlyDaemonThreads);
         runDemotionStage("flushLeaderSessionState", this::flushLeaderSessionState);
@@ -2175,11 +2175,12 @@ private void runDemotionStage(String stageName, Runnable stage) {
      * Abandon every in-flight BE agent task the (demoting) leader issued: fail their completion
      * latches so waiters (e.g. create-tablet in TabletTaskExecutor.waitForFinished, which can run on
      * a user connection thread that demotion never interrupts) unblock immediately and release their
-     * db locks, instead of waiting out tablet_create_timeout for an operation whose journal write is
-     * already fenced. AgentTaskQueue.addTask() rejects new tasks once demoting, closing the TOCTOU.
+     * db locks instead of waiting out tablet_create_timeout for an operation whose journal write is
+     * already fenced, and drop the tasks from the queue so they do not leak across a demote/re-elect
+     * cycle. AgentTaskQueue.addTask() rejects new tasks once demoting, closing the TOCTOU.
      */
-    private void failInFlightAgentTasks() {
-        com.starrocks.task.AgentTaskQueue.failAllPendingWaiters(
+    private void abandonInFlightAgentTasks() {
+        com.starrocks.task.AgentTaskQueue.abandonInFlightTasks(
                 new com.starrocks.common.Status(TStatusCode.CANCELLED, "leader is demoting"));
     }
 

fe/fe-core/src/main/java/com/starrocks/task/AgentTask.java

@@ -172,7 +172,7 @@ public void setFailed(boolean isFailed) {
      * Release any waiter blocked on this task's completion latch, failing it with {@code status}.
      * Default no-op: most agent tasks have no waiter. Latch-holding subclasses (create-replica,
      * push, tablet-metadata-update, drop-auto-increment-map) override this so a leader demotion can
-     * abandon all in-flight agent tasks at once (see {@link AgentTaskQueue#failAllPendingWaiters})
+     * abandon all in-flight agent tasks at once (see {@link AgentTaskQueue#abandonInFlightTasks})
      * and unblock their waiters immediately, instead of each wait site polling for demotion.
      * Distinct from {@link #setFailed(boolean)} (per-task status flag consumed by ALTER jobs) and
      * from the per-mark countdown (per-replica success) - overriding those would break retry/quorum.

fe/fe-core/src/main/java/com/starrocks/task/AgentTaskQueue.java

@@ -77,7 +77,7 @@ public static synchronized void addTaskList(List<AgentTask> taskList) {
 
     public static synchronized boolean addTask(AgentTask task) {
         // Source guard for leader demotion: once demoting, refuse to enqueue new agent tasks.
-        // Together with failAllPendingWaiters() (which drains what is already queued) this closes the
+        // Together with abandonInFlightTasks() (which drains what is already queued) this closes the
         // TOCTOU where a create-tablet started just after the drain would otherwise wait out its
         // timeout. The throw fails the doomed operation fast (its journal write would be fenced anyway).
         if (GlobalStateMgr.getCurrentState().isLeaderDemoting()) {
@@ -272,15 +272,17 @@ public static synchronized void clearAllTasks() {
     }
 
     /**
-     * Leader-demotion drain: fail the completion latch of every in-flight agent task so any waiter
-     * (e.g. TabletTaskExecutor.waitForFinished waiting on a create-tablet latch) unblocks at once
-     * with {@code status} instead of waiting out its timeout. Snapshots under the queue lock, then
-     * cancels outside it, so we do not hold the global queue monitor across N latch operations
-     * (which would stall concurrent BE-response processing). Tasks without a latch are no-ops.
-     * Pairs with the addTask() demoting guard, which prevents new tasks from being enqueued after
-     * this runs.
+     * Leader-demotion drain: abandon every in-flight agent task. First fail each task's completion
+     * latch (if any) so waiters (e.g. TabletTaskExecutor.waitForFinished on a create-tablet latch)
+     * unblock at once with {@code status} and release their locks instead of waiting out a timeout;
+     * then drop all tasks from the queue so they do not leak across a demote/re-elect cycle - a
+     * demoting/follower leader no longer processes the BE reports that would normally remove them.
+     * Snapshots under the queue lock and cancels outside it, so the global queue monitor is not held
+     * across N latch operations (which would stall concurrent BE-response processing). Clearing the
+     * whole queue is safe because addTask() rejects new tasks while demoting, so nothing live is
+     * discarded here.
      */
-    public static void failAllPendingWaiters(Status status) {
+    public static void abandonInFlightTasks(Status status) {
         List<AgentTask> snapshot = Lists.newArrayList();
         synchronized (AgentTaskQueue.class) {
             for (Map<Long, AgentTask> signatureMap : tasks.values()) {
@@ -294,6 +296,10 @@ public static void failAllPendingWaiters(Status status) {
                 LOG.warn("failed to cancel pending waiter for agent task {}", task, t);
             }
         }
+        synchronized (AgentTaskQueue.class) {
+            tasks.clear();
+            taskNum = 0;
+        }
     }
 
     public static synchronized int getTaskNum() {

fe/fe-core/src/test/java/com/starrocks/task/AgentTaskTest.java

@@ -495,21 +495,21 @@ public void testCreateReplicaTaskBuilderPreservesRange() throws AnalysisExceptio
     }
 
     @Test
-    public void testFailAllPendingWaitersReleasesLatch() {
+    public void testAbandonInFlightTasksReleasesLatchAndClearsQueue() {
         AgentTaskQueue.clearAllTasks();
         MarkedCountDownLatch<Long, Long> l = new MarkedCountDownLatch<>(1);
         l.addMark(backendId1, tabletId1);
         ((CreateReplicaTask) createReplicaTask).setLatch(l);
         AgentTaskQueue.addTask(createReplicaTask);
 
-        // Leader-demotion drain: fail every in-flight agent task latch so a waiter
-        // (e.g. TabletTaskExecutor.waitForFinished) unblocks at once with the demotion reason
-        // instead of waiting out its timeout.
-        AgentTaskQueue.failAllPendingWaiters(new Status(TStatusCode.CANCELLED, "leader is demoting"));
+        // Leader-demotion drain: fail every in-flight agent task latch (so a waiter such as
+        // TabletTaskExecutor.waitForFinished unblocks with the demotion reason instead of waiting out
+        // its timeout) AND drop the tasks from the queue so they do not leak across a demote/re-elect.
+        AgentTaskQueue.abandonInFlightTasks(new Status(TStatusCode.CANCELLED, "leader is demoting"));
 
         Assertions.assertEquals(0, l.getCount());
         Assertions.assertFalse(l.getStatus().ok());
-        AgentTaskQueue.clearAllTasks();
+        Assertions.assertEquals(0, AgentTaskQueue.getTaskNum());
     }
 
     @Test