branch-2.1: [fix](agent) cancel agent task when it is rejected by agent-task-pool #51138

fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java

@@ -606,22 +606,14 @@ private static void taskReport(long backendId, Map<TTaskType, Set<Long>> running
         AgentBatchTask batchTask = new AgentBatchTask(Config.report_resend_batch_task_num_per_rpc);
         long taskReportTime = System.currentTimeMillis();
         for (AgentTask task : diffTasks) {
-            // these tasks no need to do diff
-            // 1. CREATE
-            // 2. SYNC DELETE
-            // 3. CHECK_CONSISTENCY
-            // 4. STORAGE_MDEIUM_MIGRATE
-            if (task.getTaskType() == TTaskType.CREATE
-                    || task.getTaskType() == TTaskType.CHECK_CONSISTENCY
-                    || task.getTaskType() == TTaskType.STORAGE_MEDIUM_MIGRATE) {
+            if (!task.isNeedResendType()) {
                 continue;
             }
 
             // to escape sending duplicate agent task to be
             if (task.shouldResend(taskReportTime)) {
                 batchTask.addTask(task);
             }
-
         }
 
         if (LOG.isDebugEnabled()) {

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

@@ -144,6 +144,17 @@ public boolean isFinished() {
         return isFinished;
     }
 
+    public boolean isNeedResendType() {
+        // these tasks no need to do diff
+        // 1. CREATE
+        // 2. SYNC DELETE
+        // 3. CHECK_CONSISTENCY
+        // 4. STORAGE_MEDIUM_MIGRATE
+        return !(taskType == TTaskType.CREATE
+                || taskType == TTaskType.CHECK_CONSISTENCY
+                || taskType == TTaskType.STORAGE_MEDIUM_MIGRATE);
+    }
+
     public boolean shouldResend(long currentTimeMillis) {
         return createTime == -1 || currentTimeMillis - createTime > Config.agent_task_resend_wait_time_ms;
     }

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

@@ -21,21 +21,34 @@
 import org.apache.doris.common.ThreadPoolManager;
 
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
 
 public class AgentTaskExecutor {
 
-    private static final ExecutorService EXECUTOR = ThreadPoolManager.newDaemonCacheThreadPool(
+    private static final ExecutorService EXECUTOR = ThreadPoolManager.newDaemonCacheThreadPoolThrowException(
             Config.max_agent_task_threads_num, "agent-task-pool", true);
 
     public AgentTaskExecutor() {
-
     }
 
     public static void submit(AgentBatchTask task) {
         if (task == null) {
             return;
         }
-        EXECUTOR.submit(task);
+        try {
+            EXECUTOR.submit(task);
+        } catch (RejectedExecutionException e) {
+            String msg = "Task is rejected, because the agent-task-pool is full, "
+                    + "consider increasing the max_agent_task_threads_num config";
+            for (AgentTask t : task.getAllTasks()) {
+                // Skip the task if it is a resend type and already exists in the queue, because it will be
+                // re-submit to the executor later.
+                if (t.isNeedResendType() && AgentTaskQueue.contains(t)) {
+                    continue;
+                }
+                t.failedWithMsg(msg);
+            }
+        }
     }
 
 }

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

@@ -132,6 +132,19 @@ public static synchronized void removeTaskOfType(TTaskType type, long signature)
         }
     }
 
+    public static synchronized boolean contains(AgentTask task) {
+        long backendId = task.getBackendId();
+        TTaskType type = task.getTaskType();
+        long signature = task.getSignature();
+
+        if (!tasks.contains(backendId, type)) {
+            return false;
+        }
+
+        Map<Long, AgentTask> signatureMap = tasks.get(backendId, type);
+        return signatureMap.containsKey(signature);
+    }
+
     public static synchronized AgentTask getTask(long backendId, TTaskType type, long signature) {
         if (!tasks.contains(backendId, type)) {
             return null;