[Fix](job) fix drop db error and duplicate fragment_id when retry (#57307)

### What problem does this PR solve?
Fixed the following issues:
1. In a cloud environment, if the database has been deleted, attempting
to drop a job would result in an error.
2. In a cloud environment, MS offsets were not successfully put, and
statistics were inaccurate.
3. When a StreamingTask timed out and retried, a duplicate framement
error would be reported.

Author: JNSimba

fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java

@@ -89,7 +89,7 @@
 @Log4j2
 public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, Map<Object, Object>> implements
         TxnStateChangeCallback, GsonPostProcessable {
-    private final long dbId;
+    private long dbId;
     private StreamingJobStatistic jobStatistic = new StreamingJobStatistic();
     @Getter
     @Setter
@@ -129,7 +129,6 @@ public StreamingInsertJob(String jobName,
             Map<String, String> properties) {
         super(Env.getCurrentEnv().getNextId(), jobName, jobStatus, dbName, comment, createUser,
                 jobConfig, createTimeMs, executeSql);
-        this.dbId = ConnectContext.get().getCurrentDbId();
         this.properties = properties;
         init();
     }
@@ -246,11 +245,12 @@ public void updateJobStatus(JobStatus status) throws JobException {
         try {
             super.updateJobStatus(status);
             if (JobStatus.PAUSED.equals(getJobStatus())) {
-                clearRunningStreamTask();
+                clearRunningStreamTask(status);
             }
             if (isFinalStatus()) {
                 Env.getCurrentGlobalTransactionMgr().getCallbackFactory().removeCallback(getJobId());
             }
+            log.info("Streaming insert job {} update status to {}", getJobId(), getJobStatus());
         } finally {
             lock.writeLock().unlock();
         }
@@ -296,6 +296,8 @@ protected StreamingInsertTask createStreamingInsertTask() {
                 offsetProvider, getCurrentDbName(), jobProperties, originTvfProps, getCreateUser());
         Env.getCurrentEnv().getJobManager().getStreamingTaskManager().registerTask(runningStreamTask);
         this.runningStreamTask.setStatus(TaskStatus.PENDING);
+        log.info("create new streaming insert task for job {}, task {} ",
+                getJobId(), runningStreamTask.getTaskId());
         return runningStreamTask;
     }
 
@@ -314,18 +316,19 @@ public boolean needScheduleTask() {
         return (getJobStatus().equals(JobStatus.RUNNING) || getJobStatus().equals(JobStatus.PENDING));
     }
 
-    public void clearRunningStreamTask() {
+    public void clearRunningStreamTask(JobStatus newJobStatus) {
         if (runningStreamTask != null) {
+            log.info("clear running streaming insert task for job {}, task {}, status {} ",
+                    getJobId(), runningStreamTask.getTaskId(), runningStreamTask.getStatus());
+            runningStreamTask.cancel(JobStatus.STOPPED.equals(newJobStatus) ? false : true);
             runningStreamTask.closeOrReleaseResources();
-            runningStreamTask = null;
         }
     }
 
     public boolean hasMoreDataToConsume() {
         return offsetProvider.hasMoreDataToConsume();
     }
 
-
     @Override
     public void onTaskFail(StreamingJobSchedulerTask task) throws JobException {
         if (task.getErrMsg() != null) {
@@ -358,6 +361,8 @@ public void onStreamTaskSuccess(StreamingInsertTask task) {
             Env.getCurrentEnv().getJobManager().getStreamingTaskManager().removeRunningTask(task);
             StreamingInsertTask nextTask = createStreamingInsertTask();
             this.runningStreamTask = nextTask;
+            log.info("Streaming insert job {} create next streaming insert task {} after task {} success",
+                    getJobId(), nextTask.getTaskId(), task.getTaskId());
         } finally {
             writeUnlock();
         }
@@ -374,6 +379,17 @@ private void updateJobStatisticAndOffset(StreamingTaskTxnCommitAttachment attach
         offsetProvider.updateOffset(offsetProvider.deserializeOffset(attachment.getOffset()));
     }
 
+    private void updateCloudJobStatisticAndOffset(StreamingTaskTxnCommitAttachment attachment) {
+        if (this.jobStatistic == null) {
+            this.jobStatistic = new StreamingJobStatistic();
+        }
+        this.jobStatistic.setScannedRows(attachment.getScannedRows());
+        this.jobStatistic.setLoadBytes(attachment.getLoadBytes());
+        this.jobStatistic.setFileNumber(attachment.getNumFiles());
+        this.jobStatistic.setFileSize(attachment.getFileBytes());
+        offsetProvider.updateOffset(offsetProvider.deserializeOffset(attachment.getOffset()));
+    }
+
     @Override
     public void onRegister() throws JobException {
         Env.getCurrentGlobalTransactionMgr().getCallbackFactory().addCallback(this);
@@ -555,6 +571,12 @@ public void beforeCommitted(TransactionState txnState) throws TransactionExcepti
         boolean shouldReleaseLock = false;
         writeLock();
         try {
+            if (runningStreamTask.getIsCanceled().get()) {
+                log.info("streaming insert job {} task {} is canceled, skip beforeCommitted",
+                        getJobId(), runningStreamTask.getTaskId());
+                return;
+            }
+
             ArrayList<Long> taskIds = new ArrayList<>();
             taskIds.add(runningStreamTask.getTaskId());
             // todo: Check whether the taskid of runningtask is consistent with the taskid associated with txn
@@ -601,14 +623,26 @@ public void replayOnCommitted(TransactionState txnState) {
         succeedTaskCount.incrementAndGet();
     }
 
+    public long getDbId() {
+        if (dbId <= 0) {
+            try {
+                this.dbId = Env.getCurrentInternalCatalog().getDbOrAnalysisException(getCurrentDbName()).getId();
+            } catch (AnalysisException e) {
+                log.warn("failed to get db id for streaming insert job {}, db name: {}, msg: {}",
+                        getJobId(), getCurrentDbName(), e.getMessage());
+            }
+        }
+        return dbId;
+    }
+
     public void replayOnCloudMode() throws JobException {
         Cloud.GetStreamingTaskCommitAttachRequest.Builder builder =
                 Cloud.GetStreamingTaskCommitAttachRequest.newBuilder();
         builder.setCloudUniqueId(Config.cloud_unique_id);
-        builder.setDbId(dbId);
+        builder.setDbId(getDbId());
         builder.setJobId(getJobId());
 
-        Cloud.GetStreamingTaskCommitAttachResponse response;
+        Cloud.GetStreamingTaskCommitAttachResponse response = null;
         try {
             response = MetaServiceProxy.getInstance().getStreamingTaskCommitAttach(builder.build());
             if (response.getStatus().getCode() != Cloud.MetaServiceCode.OK) {
@@ -621,13 +655,13 @@ public void replayOnCloudMode() throws JobException {
                 }
             }
         } catch (RpcException e) {
-            log.info("failed to get streaming task commit attach {}", e);
+            log.info("failed to get streaming task commit attach {}", response, e);
             throw new JobException(e.getMessage());
         }
 
         StreamingTaskTxnCommitAttachment commitAttach =
                 new StreamingTaskTxnCommitAttachment(response.getCommitAttach());
-        updateJobStatisticAndOffset(commitAttach);
+        updateCloudJobStatisticAndOffset(commitAttach);
     }
 
     @Override

fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java

@@ -27,6 +27,7 @@
 import org.apache.doris.job.extensions.insert.InsertTask;
 import org.apache.doris.job.offset.Offset;
 import org.apache.doris.job.offset.SourceOffsetProvider;
+import org.apache.doris.job.offset.s3.S3Offset;
 import org.apache.doris.nereids.StatementContext;
 import org.apache.doris.nereids.glue.LogicalPlanAdapter;
 import org.apache.doris.nereids.parser.NereidsParser;
@@ -67,10 +68,12 @@ public class StreamingInsertTask {
     private UserIdentity userIdentity;
     private ConnectContext ctx;
     private Offset runningOffset;
+    @Getter
     private AtomicBoolean isCanceled = new AtomicBoolean(false);
     private StreamingJobProperties jobProperties;
     private Map<String, String> originTvfProps;
     SourceOffsetProvider offsetProvider;
+    private int retryCount = 0;
 
     public StreamingInsertTask(long jobId,
                                long taskId,
@@ -93,22 +96,32 @@ public StreamingInsertTask(long jobId,
     }
 
     public void execute() throws JobException {
-        try {
-            before();
-            run();
-            onSuccess();
-        } catch (Exception e) {
-            if (TaskStatus.CANCELED.equals(status)) {
+        while (retryCount <= MAX_RETRY) {
+            try {
+                before();
+                run();
+                onSuccess();
                 return;
-            }
-            log.warn("execute task error, job id is {}, task id is {}", jobId, taskId, e);
-            onFail(e.getMessage());
-        } finally {
-            // The cancel logic will call the closeOrReleased Resources method by itself.
-            // If it is also called here,
-            // it may result in the inability to obtain relevant information when canceling the task
-            if (!TaskStatus.CANCELED.equals(status)) {
-                closeOrReleaseResources();
+            } catch (Exception e) {
+                if (TaskStatus.CANCELED.equals(status)) {
+                    return;
+                }
+                this.errMsg = e.getMessage();
+                retryCount++;
+                if (retryCount > MAX_RETRY) {
+                    log.error("Task execution failed after {} retries.", MAX_RETRY, e);
+                    onFail(e.getMessage());
+                    return;
+                }
+                log.warn("execute streaming task error, job id is {}, task id is {}, retrying {}/{}: {}",
+                        jobId, taskId, retryCount, MAX_RETRY, e.getMessage());
+            } finally {
+                // The cancel logic will call the closeOrReleased Resources method by itself.
+                // If it is also called here,
+                // it may result in the inability to obtain relevant information when canceling the task
+                if (!TaskStatus.CANCELED.equals(status)) {
+                    closeOrReleaseResources();
+                }
             }
         }
     }
@@ -118,7 +131,8 @@ private void before() throws Exception {
         this.startTimeMs = System.currentTimeMillis();
 
         if (isCanceled.get()) {
-            throw new JobException("Streaming insert task has been canceled, task id: {}", getTaskId());
+            log.info("streaming insert task has been canceled, task id is {}", getTaskId());
+            return;
         }
         ctx = InsertTask.makeConnectContext(userIdentity, currentDb);
         ctx.setSessionVariable(jobProperties.getSessionVariable());
@@ -135,42 +149,36 @@ private void before() throws Exception {
             throw new JobException("Can not get Parsed plan");
         }
         this.taskCommand = offsetProvider.rewriteTvfParams(baseCommand, runningOffset);
-        this.taskCommand.setLabelName(Optional.of(getJobId() + LABEL_SPLITTER + getTaskId()));
+        this.taskCommand.setLabelName(Optional.of(labelName));
         this.stmtExecutor = new StmtExecutor(ctx, new LogicalPlanAdapter(taskCommand, ctx.getStatementContext()));
     }
 
     private void run() throws JobException {
+        StreamingInsertJob job =
+                (StreamingInsertJob) Env.getCurrentEnv().getJobManager().getJob(getJobId());
+        StreamingInsertTask runningStreamTask = job.getRunningStreamTask();
+        log.info("current running stream task id is {} for job id {}",
+                runningStreamTask == null ? -1 : runningStreamTask.getTaskId(), getJobId());
+        if (isCanceled.get()) {
+            log.info("task has been canceled, task id is {}", getTaskId());
+            return;
+        }
+        log.info("start to run streaming insert task, label {}, offset is {}, filepath {}",
+                labelName, runningOffset.toString(), ((S3Offset) runningOffset).getFileLists());
         String errMsg = null;
-        int retry = 0;
-        while (retry <= MAX_RETRY) {
-            try {
-                if (isCanceled.get()) {
-                    log.info("task has been canceled, task id is {}", getTaskId());
-                    return;
-                }
-                taskCommand.run(ctx, stmtExecutor);
-                if (ctx.getState().getStateType() == QueryState.MysqlStateType.OK) {
-                    return;
-                } else {
-                    errMsg = ctx.getState().getErrorMessage();
-                }
-                log.error(
-                        "streaming insert failed with {}, reason {}, to retry",
-                        taskCommand.getLabelName(),
-                        errMsg);
-                if (retry == MAX_RETRY) {
-                    errMsg = "reached max retry times, failed with " + errMsg;
-                }
-            } catch (Exception e) {
-                log.warn("execute insert task error, label is {},offset is {}", taskCommand.getLabelName(),
-                         runningOffset.toString(), e);
-                errMsg = Util.getRootCauseMessage(e);
+        try {
+            taskCommand.run(ctx, stmtExecutor);
+            if (ctx.getState().getStateType() == QueryState.MysqlStateType.OK) {
+                return;
+            } else {
+                errMsg = ctx.getState().getErrorMessage();
             }
-            retry++;
+            throw new JobException(errMsg);
+        } catch (Exception e) {
+            log.warn("execute insert task error, label is {},offset is {}", taskCommand.getLabelName(),
+                    runningOffset.toString(), e);
+            throw new JobException(Util.getRootCauseMessage(e));
         }
-        log.error("streaming insert task failed, job id is {}, task id is {}, offset is {}, errMsg is {}",
-                getJobId(), getTaskId(), runningOffset.toString(), errMsg);
-        throw new JobException(errMsg);
     }
 
     public boolean onSuccess() throws JobException {
@@ -218,6 +226,8 @@ public void cancel(boolean needWaitCancelComplete) {
         }
         isCanceled.getAndSet(true);
         if (null != stmtExecutor) {
+            log.info("cancelling streaming insert task, job id is {}, task id is {}",
+                    getJobId(), getTaskId());
             stmtExecutor.cancel(new Status(TStatusCode.CANCELLED, "streaming insert task cancelled"),
                     needWaitCancelComplete);
         }

fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java

@@ -30,11 +30,13 @@
 import org.apache.doris.thrift.TCell;
 import org.apache.doris.thrift.TRow;
 
+import lombok.extern.log4j.Log4j2;
 import org.apache.commons.lang3.StringUtils;
 
 import java.util.Arrays;
 import java.util.List;
 
+@Log4j2
 public class StreamingJobSchedulerTask extends AbstractTask {
     private static final long BACK_OFF_BASIC_TIME_SEC = 10L;
     private static final long MAX_BACK_OFF_TIME_SEC = 60 * 5;
@@ -108,6 +110,7 @@ protected void closeOrReleaseResources() {
 
     @Override
     protected void executeCancelLogic(boolean needWaitCancelComplete) throws Exception {
+        log.info("cancelling streaming insert job scheduler task for job id {}", streamingInsertJob.getJobId());
         if (streamingInsertJob.getRunningStreamTask() != null) {
             streamingInsertJob.getRunningStreamTask().cancel(needWaitCancelComplete);
         }
@@ -119,6 +122,10 @@ public TRow getTvfInfo(String jobName) {
         if (runningTask == null) {
             return null;
         }
+        if (!streamingInsertJob.needScheduleTask()) {
+            //todo: should list history task
+            return null;
+        }
         TRow trow = new TRow();
         trow.addToColumnValue(new TCell().setStringVal(String.valueOf(runningTask.getTaskId())));
         trow.addToColumnValue(new TCell().setStringVal(String.valueOf(runningTask.getJobId())));

fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java

@@ -51,6 +51,7 @@
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.plans.commands.AlterJobCommand;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.rpc.RpcException;
 
 import com.google.common.collect.Lists;
 import lombok.Getter;
@@ -241,25 +242,25 @@ public void dropJobInternal(T job, boolean isReplay) throws JobException {
     }
 
     private void deleteStremingJob(AbstractJob<?, C> job) throws JobException {
-        boolean isStreamingJob = job.getJobConfig().getExecuteType().equals(JobExecuteType.STREAMING);
-        if (!(Config.isCloudMode() && isStreamingJob)) {
+        if (!(Config.isCloudMode() && job instanceof StreamingInsertJob)) {
             return;
         }
+        StreamingInsertJob streamingJob = (StreamingInsertJob) job;
+        Cloud.DeleteStreamingJobResponse resp = null;
         try {
-            long dbId = Env.getCurrentInternalCatalog().getDbOrDdlException(job.getCurrentDbName()).getId();
             Cloud.DeleteStreamingJobRequest req = Cloud.DeleteStreamingJobRequest.newBuilder()
                     .setCloudUniqueId(Config.cloud_unique_id)
-                    .setDbId(dbId)
+                    .setDbId(streamingJob.getDbId())
                     .setJobId(job.getJobId())
                     .build();
-            Cloud.DeleteStreamingJobResponse resp = MetaServiceProxy.getInstance().deleteStreamingJob(req);
+            resp = MetaServiceProxy.getInstance().deleteStreamingJob(req);
             if (resp.getStatus().getCode() != Cloud.MetaServiceCode.OK) {
-                throw new JobException("deleteJobKey failed for jobId={}, dbId={}, status={}",
-                        job.getJobId(), dbId, resp.getStatus());
+                log.warn("failed to delete streaming job, response: {}", resp);
+                throw new JobException("deleteJobKey failed for jobId=%s, dbId=%s, status=%s",
+                        job.getJobId(), job.getJobId(), resp.getStatus());
             }
-        } catch (Exception e) {
-            throw new JobException("deleteJobKey exception for jobId={}, dbName={}",
-                    job.getJobId(), job.getCurrentDbName(), e);
+        } catch (RpcException e) {
+            log.warn("failed to delete streaming job {}", resp, e);
         }
     }