[opt](agent-task) Add a daemon thread to clean up agent tasks on dead BEs

Author: TangSiyang2001

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -3651,4 +3651,6 @@ public static int metaServiceRpcRetryTimes() {
 
     @ConfField(mutable = true)
     public static String aws_credentials_provider_version = "v2";
+
+    public static long agent_task_health_check_intervals_ms = 5 * 60 * 1000L; // 5 min
 }

fe/fe-core/src/main/java/org/apache/doris/task/AgentTaskCleanupDaemon.java

@@ -0,0 +1,56 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.task;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.Status;
+import org.apache.doris.common.util.MasterDaemon;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.TStatusCode;
+
+public class AgentTaskCleanupDaemon extends MasterDaemon {
+    public AgentTaskCleanupDaemon() {
+        super("agent-task-cleanup", Config.agent_task_health_check_intervals_ms);
+    }
+
+    @Override
+    protected void runAfterCatalogReady() {
+        SystemInfoService infoService = Env.getCurrentSystemInfo();
+        infoService.getAllClusterBackends(false)
+                .stream()
+                .filter(backend -> !backend.isAlive())
+                .map(Backend::getId)
+                .forEach(this::removeInactiveBeAgentTasks);
+    }
+
+    private void removeInactiveBeAgentTasks(Long beId) {
+        AgentTaskQueue.removeTask(beId, (agentTask -> {
+            long tabletId = agentTask.getTabletId();
+            String errMsg = "BE down, agent task is aborted. BE=" + beId + ", tablet=" + tabletId;
+            if (agentTask instanceof PushTask) {
+                PushTask task = ((PushTask) agentTask);
+                task.countDownLatchWithStatus(beId, tabletId, new Status(TStatusCode.ABORTED, errMsg));
+            }
+            agentTask.setFinished(true);
+            agentTask.setErrorCode(TStatusCode.ABORTED);
+            agentTask.setErrorMsg(errMsg);
+        }));
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/task/AgentTaskQueue.java

@@ -33,6 +33,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Consumer;
 
 /**
  * Task queue
@@ -96,6 +97,16 @@ public static synchronized void removeTask(long backendId, TTaskType type, long
         --taskNum;
     }
 
+    public static synchronized void removeTask(long backendId, Consumer<AgentTask> onTaskRemoved) {
+        Map<TTaskType, Map<Long, AgentTask>> tasks = AgentTaskQueue.tasks.row(backendId);
+        tasks.forEach((type, taskSet) -> {
+            taskSet.forEach((signature, task) -> {
+                removeTask(backendId, type, signature);
+                onTaskRemoved.accept(task);
+            });
+        });
+    }
+
     /*
      * we cannot define a push task with only 'backendId', 'signature' and 'TTaskType'
      * add version and TPushType to help

fe/fe-core/src/main/java/org/apache/doris/task/PushTask.java

@@ -58,7 +58,7 @@ public class PushTask extends AgentTask {
     private TPushType pushType;
     private List<Predicate> conditions;
     // for synchronous delete
-    private MarkedCountDownLatch latch;
+    private MarkedCountDownLatch<Long, Long> latch;
 
     // lzop decompress or not
     private boolean needDecompress;
@@ -212,6 +212,18 @@ public void countDownLatch(long backendId, long tabletId) {
         }
     }
 
+    public void countDownLatchWithStatus(long backendId, long tabletId, Status st) {
+        if (this.latch == null) {
+            return;
+        }
+        if (latch.markedCountDownWithStatus(backendId, tabletId, st)) {
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("pushTask current latch count with status: {}. backend: {}, tablet:{}, st::{}",
+                        latch.getCount(), backendId, tabletId, st);
+            }
+        }
+    }
+
     // call this always means one of tasks is failed. count down to zero to finish entire task
     public void countDownToZero(TStatusCode code, String errMsg) {
         if (this.latch != null) {

regression-test/suites/fault_injection_p0/test_sc_stuck_when_be_down.groovy

@@ -0,0 +1,61 @@
+import org.apache.doris.regression.suite.ClusterOptions
+
+suite("test_sc_stuck_when_be_down", "docker") {
+    def options = new ClusterOptions()
+    options.cloudMode = false
+    options.beNum = 3
+    options.feNum = 2
+    options.enableDebugPoints()
+    options.feConfigs += ["agent_task_health_check_intervals_ms=5000"]
+
+    docker(options) {
+        GetDebugPoint().clearDebugPointsForAllBEs()
+
+        def tblName = "test_sc_stuck_when_be_down"
+        sql """ DROP TABLE IF EXISTS ${tblName} """
+        sql """
+                CREATE TABLE IF NOT EXISTS ${tblName} (
+                    `k` int NOT NULL,
+                    `v0` int NOT NULL,
+                    `v1` int NOT NULL
+                ) 
+                DUPLICATE KEY(`k`)
+                DISTRIBUTED BY HASH(`k`) BUCKETS 24
+                PROPERTIES (
+                    "replication_allocation" = "tag.location.default: 3"
+                )
+        """
+        sql """ INSERT INTO ${tblName} SELECT number, number, number from numbers("number" = "1024") """
+
+        GetDebugPoint().enableDebugPointForAllBEs("SchemaChangeJob::_do_process_alter_tablet.block")
+        try {
+            sql """ ALTER TABLE ${tblName} MODIFY COLUMN v0 VARCHAR(100) """
+            sleep(3000)
+            cluster.stopBackends(1)
+            GetDebugPoint().clearDebugPointsForAllBEs()
+            waitForSchemaChangeDone {
+                sql """ SHOW ALTER TABLE COLUMN WHERE TableName='${tblName}' ORDER BY createtime DESC LIMIT 1 """
+                time 600
+            }
+        } finally {
+            GetDebugPoint().clearDebugPointsForAllBEs()
+        }
+
+        GetDebugPoint().enableDebugPointForAllBEs("SchemaChangeJob::_do_process_alter_tablet.block")
+        try {
+            sql """ ALTER TABLE ${tblName} MODIFY COLUMN v1 VARCHAR(100) """
+            sleep(3000)
+            cluster.stopBackends(1, 2)
+            GetDebugPoint().clearDebugPointsForAllBEs()
+            waitForSchemaChangeDone {
+                sql """ SHOW ALTER TABLE COLUMN WHERE TableName='${tblName}' ORDER BY createtime DESC LIMIT 1 """
+                time 600
+            }
+            assertTrue(false)
+        } catch (Exception ignore) {
+            // do nothing
+        } finally {
+            GetDebugPoint().clearDebugPointsForAllBEs()
+        }
+    }
+}