From c18baa34969390986a7ba0c5087bb4fdf5ac67a5 Mon Sep 17 00:00:00 2001
From: Maksim Ryzhukhin <eumaxpl@gmail.com>
Date: Fri, 7 Aug 2026 01:28:15 +0200
Subject: [PATCH] [fix](streaming) avoid infinite retry when cloud mode job
 progress is not found
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

In compute-storage-decoupled (cloud) mode, replayOnCloudMode() asks MetaService
for a streaming job's persisted progress. When MetaService answers
STREAMING_JOB_PROGRESS_NOT_FOUND, the method logs a warning and returns void.
The caller cannot tell 'no progress exists' from 'progress loaded successfully',
so it repeats the RPC on every scheduler tick indefinitely, flooding the log
and leaving the job stuck in PENDING.

Fix:
- Change replayOnCloudMode() to return boolean (false on NOT_FOUND).
- Add a transient cloudProgressMissing flag that short-circuits subsequent
  attempts so the pointless RPC is not re-issued on every tick or journal entry.
- Clear the flag in afterCommitted() when the job actually persists progress,
  enabling recovery after MetaService later stores valid state.

The flag is intentionally transient (not serialized): it resets on FE restart,
which gives MetaService another chance if the issue was temporary. This is the
correct behavior — a permanent NOT_FOUND is expected for newly created jobs
that have not yet committed their first transaction.
---
 .../insert/streaming/StreamingInsertJob.java  | 41 +++++++++++++++----
 .../streaming/StreamingJobSchedulerTask.java  |  6 ++-
 2 files changed, 39 insertions(+), 8 deletions(-)

diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
index affb6d6c5992fd..d74bca6ee68cd2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
@@ -196,6 +196,12 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M
     @Setter
     private transient volatile boolean needRebuildReader = true;
 
+    // Set to true when MetaService returns STREAMING_JOB_PROGRESS_NOT_FOUND during replay.
+    // Prevents repeated RPC calls on subsequent scheduler ticks for the same job.
+    // The job will proceed using its configured offset instead of cloud-persisted progress.
+    // Intentionally transient: resets on FE restart so replay can re-attempt after MetaService recovery.
+    private transient volatile boolean cloudProgressMissing = false;
+
     // The sampling window starts at the beginning of the sampling window.
     // If the error rate exceeds `max_filter_ratio` within the window, the sampling fails.
     @Setter
@@ -1035,8 +1041,9 @@ public void replayOnUpdated(StreamingInsertJob replayJob) {
         }
         try {
             modifyPropertiesInternal(replayJob.getProperties());
-            // When the pause state is restarted, it also needs to be updated
-            if (Config.isCloudMode()) {
+            // When the pause state is restarted, it also needs to be updated.
+            // Skip if we already know progress is missing — avoid repeated pointless RPCs during replay.
+            if (Config.isCloudMode() && !cloudProgressMissing) {
                 replayOnCloudMode();
             }
         } catch (Exception e) {
@@ -1387,6 +1394,8 @@ public void afterCommitted(TransactionState txnState, boolean txnOperated) throw
         StreamingTaskTxnCommitAttachment attachment =
                 (StreamingTaskTxnCommitAttachment) txnState.getTxnCommitAttachment();
         updateJobStatisticAndOffset(attachment, false);
+        // Progress now exists in MetaService; clear the missing flag so future replays can recover.
+        cloudProgressMissing = false;
     }
 
     @Override
@@ -1411,7 +1420,21 @@ public long getDbId() {
         return dbId;
     }
 
-    public void replayOnCloudMode() throws JobException {
+    /**
+     * Recover job statistics and offset from MetaService (cloud mode).
+     *
+     * @return true if cloud progress was successfully recovered; false if progress was not found
+     *         in MetaService (STREAMING_JOB_PROGRESS_NOT_FOUND). When false, the job should
+     *         proceed using its configured offset property rather than cloud-persisted progress.
+     * @throws JobException on transient MetaService errors (network, unknown codes) that warrant retry
+     */
+    public boolean replayOnCloudMode() throws JobException {
+        // Once we know progress is missing, skip further RPC attempts — the answer won't change
+        // within the same FE lifetime unless the job successfully commits a new task.
+        if (cloudProgressMissing) {
+            return false;
+        }
+
         Cloud.GetStreamingTaskCommitAttachRequest.Builder builder =
                 Cloud.GetStreamingTaskCommitAttachRequest.newBuilder()
                         .setRequestIp(FrontendOptions.getLocalHostAddressCached());
@@ -1423,22 +1446,26 @@ public void replayOnCloudMode() throws JobException {
         try {
             response = MetaServiceProxy.getInstance().getStreamingTaskCommitAttach(builder.build());
             if (response.getStatus().getCode() != Cloud.MetaServiceCode.OK) {
-                log.warn("failed to get streaming task commit attach, response: {}", response);
                 if (response.getStatus().getCode() == Cloud.MetaServiceCode.STREAMING_JOB_PROGRESS_NOT_FOUND) {
-                    log.warn("not found streaming job progress, response: {}", response);
-                    return;
+                    cloudProgressMissing = true;
+                    log.warn("Cloud progress not found for streaming job {} (db_id={}). "
+                            + "Job will start from configured offset instead of cloud-persisted progress. "
+                            + "MetaService response: {}", getJobId(), getDbId(), response.getStatus().getMsg());
+                    return false;
                 } else {
+                    log.warn("failed to get streaming task commit attach, response: {}", response);
                     throw new JobException(response.getStatus().getMsg());
                 }
             }
         } catch (RpcException e) {
-            log.info("failed to get streaming task commit attach {}", response, e);
+            log.warn("RPC failed when getting streaming task commit attach for job {}", getJobId(), e);
             throw new JobException(e.getMessage());
         }
 
         StreamingTaskTxnCommitAttachment commitAttach =
                 new StreamingTaskTxnCommitAttachment(response.getCommitAttach());
         updateCloudJobStatisticAndOffset(commitAttach, true);
+        return true;
     }
 
     @Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java
index 030c3bb2b8bd6c..4bcd91eb04493d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java
@@ -57,7 +57,11 @@ public void run() throws JobException {
     private void handlePendingState() throws JobException {
         if (Config.isCloudMode()) {
             try {
-                streamingInsertJob.replayOnCloudMode();
+                boolean progressRecovered = streamingInsertJob.replayOnCloudMode();
+                if (!progressRecovered) {
+                    log.info("Cloud progress missing for job {}; proceeding with configured offset",
+                            streamingInsertJob.getJobId());
+                }
             } catch (JobException e) {
                 streamingInsertJob.setFailureReason(
                     new FailureReason(InternalErrorCode.INTERNAL_ERR, e.getMessage()));