From bea9564e385d3b0ce01866047db08cfaf8f47cf1 Mon Sep 17 00:00:00 2001
From: "github-actions[bot]"
 <41898282+github-actions[bot]@users.noreply.github.com>
Date: Fri, 6 Dec 2024 17:01:45 +0800
Subject: [PATCH] branch-2.1: [fix](Export) Fix the problem of exporting stuck
 #44944 (#45094)

Cherry-picked from #44944

Co-authored-by: Tiewei Fang <fangtiewei@selectdb.com>
---
 .../java/org/apache/doris/load/ExportMgr.java | 30 +++++++++----------
 .../scheduler/disruptor/TaskDisruptor.java    | 11 ++++---
 .../manager/TransientTaskManager.java         |  4 +--
 3 files changed, 22 insertions(+), 23 deletions(-)

diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java b/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java
index 12791ba8b7e603..5ee68045b9f7b0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java
@@ -102,26 +102,24 @@ public void addExportJobAndRegisterTask(ExportJob job) throws Exception {
                 throw new LabelAlreadyUsedException(job.getLabel());
             }
             unprotectAddJob(job);
-            // delete existing files
-            if (Config.enable_delete_existing_files && Boolean.parseBoolean(job.getDeleteExistingFiles())) {
-                if (job.getBrokerDesc() == null) {
-                    throw new AnalysisException("Local file system does not support delete existing files");
-                }
-                String fullPath = job.getExportPath();
-                BrokerUtil.deleteDirectoryWithFileSystem(fullPath.substring(0, fullPath.lastIndexOf('/') + 1),
-                        job.getBrokerDesc());
-            }
             Env.getCurrentEnv().getEditLog().logExportCreate(job);
-            // ATTN: Must add task after edit log, otherwise the job may finish before adding job.
-            job.getCopiedTaskExecutors().forEach(executor -> {
-                Env.getCurrentEnv().getTransientTaskManager().addMemoryTask(executor);
-            });
-            LOG.info("add export job. {}", job);
-
         } finally {
             writeUnlock();
         }
-
+        // delete existing files
+        if (Config.enable_delete_existing_files && Boolean.parseBoolean(job.getDeleteExistingFiles())) {
+            if (job.getBrokerDesc() == null) {
+                throw new AnalysisException("Local file system does not support delete existing files");
+            }
+            String fullPath = job.getExportPath();
+            BrokerUtil.deleteDirectoryWithFileSystem(fullPath.substring(0, fullPath.lastIndexOf('/') + 1),
+                    job.getBrokerDesc());
+        }
+        // ATTN: Must add task after edit log, otherwise the job may finish before adding job.
+        for (int i = 0; i < job.getCopiedTaskExecutors().size(); i++) {
+            Env.getCurrentEnv().getTransientTaskManager().addMemoryTask(job.getCopiedTaskExecutors().get(i));
+        }
+        LOG.info("add export job. {}", job);
     }
 
     public void cancelExportJob(CancelExportStmt stmt) throws DdlException, AnalysisException {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/scheduler/disruptor/TaskDisruptor.java b/fe/fe-core/src/main/java/org/apache/doris/scheduler/disruptor/TaskDisruptor.java
index 345b31d6bc2537..8144ca22ea22eb 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/scheduler/disruptor/TaskDisruptor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/scheduler/disruptor/TaskDisruptor.java
@@ -20,6 +20,7 @@
 import org.apache.doris.common.Config;
 import org.apache.doris.common.CustomThreadFactory;
 import org.apache.doris.scheduler.constants.TaskType;
+import org.apache.doris.scheduler.exception.JobException;
 
 import com.lmax.disruptor.EventTranslatorThreeArg;
 import com.lmax.disruptor.LiteTimeoutBlockingWaitStrategy;
@@ -119,15 +120,17 @@ public void tryPublish(Long jobId, Long taskId, TaskType taskType) {
      *
      * @param taskId task id
      */
-    public void tryPublishTask(Long taskId) {
+    public void tryPublishTask(Long taskId) throws JobException {
         if (isClosed) {
             log.info("tryPublish failed, disruptor is closed, taskId: {}", taskId);
             return;
         }
-        try {
+        // We reserve two slots in the ring buffer
+        // to prevent it from becoming stuck due to competition between producers and consumers.
+        if (disruptor.getRingBuffer().hasAvailableCapacity(2)) {
             disruptor.publishEvent(TRANSLATOR, taskId, 0L, TaskType.TRANSIENT_TASK);
-        } catch (Exception e) {
-            log.warn("tryPublish failed, taskId: {}", taskId, e);
+        } else {
+            throw new JobException("There is not enough available capacity in the RingBuffer.");
         }
     }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/scheduler/manager/TransientTaskManager.java b/fe/fe-core/src/main/java/org/apache/doris/scheduler/manager/TransientTaskManager.java
index 7461399c8eb0c5..de501d3e0c2ffd 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/scheduler/manager/TransientTaskManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/scheduler/manager/TransientTaskManager.java
@@ -21,7 +21,6 @@
 import org.apache.doris.scheduler.exception.JobException;
 import org.apache.doris.scheduler.executor.TransientTaskExecutor;
 
-import lombok.Setter;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -41,7 +40,6 @@ public class TransientTaskManager {
      * disruptor is used to handle task
      * disruptor will start a thread pool to handle task
      */
-    @Setter
     private TaskDisruptor disruptor;
 
     public TransientTaskManager() {
@@ -56,7 +54,7 @@ public TransientTaskExecutor getMemoryTaskExecutor(Long taskId) {
         return taskExecutorMap.get(taskId);
     }
 
-    public Long addMemoryTask(TransientTaskExecutor executor) {
+    public Long addMemoryTask(TransientTaskExecutor executor) throws JobException {
         Long taskId = executor.getId();
         taskExecutorMap.put(taskId, executor);
         disruptor.tryPublishTask(taskId);