[fix](agenttask) failed tasks and remove them from queue if an error

happens
dataroaring · Oct 22, 2024 · 0a23f8a · 0a23f8a
1 parent df4c7f2
commit 0a23f8a
Showing 1 changed file with 10 additions and 6 deletions.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/task/AgentBatchTask.java b/fe/fe-core/src/main/java/org/apache/doris/task/AgentBatchTask.java
@@ -157,12 +157,13 @@ public void run() {
             BackendService.Client client = null;
             TNetworkAddress address = null;
             boolean ok = false;
+            Backend backend = Env.getCurrentSystemInfo().getBackend(backendId);
+            if (backend == null || !backend.isAlive()) {
+                continue;
+            }
+            List<AgentTask> tasks = this.backendIdToTasks.get(backendId).subList(0, Config.schedule_batch_size);
+
             try {
-                Backend backend = Env.getCurrentSystemInfo().getBackend(backendId);
-                if (backend == null || !backend.isAlive()) {
-                    continue;
-                }
-                List<AgentTask> tasks = this.backendIdToTasks.get(backendId);
                 // create AgentClient
                 String host = FeConstants.runningUnitTest ? "127.0.0.1" : backend.getHost();
                 address = new TNetworkAddress(host, backend.getBePort());
@@ -172,7 +173,6 @@ public void run() {
                     try {
                         agentTaskRequests.add(toAgentTaskRequest(task));
                     } catch (Exception e) {
-                        task.failed();
                         throw e;
                     }
                 }
@@ -187,6 +187,10 @@ public void run() {
                 }
                 ok = true;
             } catch (Exception e) {
+                for (AgentTask task : tasks) {
+                    task.failed();
+                    AgentTaskQueue.remove(task, );
+                }
                 LOG.warn("task exec error. backend[{}]", backendId, e);
             } finally {
                 if (ok) {