branch-4.0: [fix](job) fix streaming job stuck when S3 auth error is silently ignored in fetchRemoteMeta #61284 (#61296)

Cherry-picked from #61284

Co-authored-by: wudi <[email protected]>

Author: github-actions[bot]

fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java

@@ -17,6 +17,8 @@
 
 package org.apache.doris.job.offset.s3;
 
+import org.apache.doris.backup.Status;
+import org.apache.doris.common.util.DebugPointUtil;
 import org.apache.doris.datasource.property.storage.StorageProperties;
 import org.apache.doris.fs.FileSystemFactory;
 import org.apache.doris.fs.GlobListResult;
@@ -69,6 +71,11 @@ public S3Offset getNextOffset(StreamingJobProperties jobProps, Map<String, Strin
             filePath = storageProperties.validateAndNormalizeUri(uri);
             GlobListResult globListResult = fileSystem.globListWithLimit(filePath, rfiles, startFile,
                     jobProps.getS3BatchBytes(), jobProps.getS3BatchFiles());
+            if (globListResult == null || !globListResult.getStatus().ok()) {
+                String errMsg = globListResult != null
+                        ? globListResult.getStatus().getErrMsg() : "null result";
+                throw new RuntimeException("Failed to list S3 files: " + errMsg);
+            }
 
             if (!rfiles.isEmpty()) {
                 String bucket = globListResult.getBucket();
@@ -162,11 +169,19 @@ public void fetchRemoteMeta(Map<String, String> properties) throws Exception {
             String filePath = storageProperties.validateAndNormalizeUri(uri);
             List<RemoteFile> objects = new ArrayList<>();
             GlobListResult globListResult = fileSystem.globListWithLimit(filePath, objects, startFile, 1, 1);
-            if (globListResult != null && !objects.isEmpty() && StringUtils.isNotEmpty(globListResult.getMaxFile())) {
+            // debug point: simulate globListWithLimit returning a failed status (e.g. S3 auth error)
+            if (DebugPointUtil.isEnable("S3SourceOffsetProvider.fetchRemoteMeta.error")) {
+                globListResult = new GlobListResult(new Status(Status.ErrCode.COMMON_ERROR,
+                        "debug point: simulated S3 auth error"));
+            }
+            if (globListResult == null || !globListResult.getStatus().ok()) {
+                String errMsg = globListResult != null
+                        ? globListResult.getStatus().getErrMsg() : "null result";
+                throw new Exception("Failed to list S3 files: " + errMsg);
+            }
+            if (!objects.isEmpty() && StringUtils.isNotEmpty(globListResult.getMaxFile())) {
                 maxEndFile = globListResult.getMaxFile();
             }
-        } catch (Exception e) {
-            throw e;
         }
     }
 

regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_fetch_meta_error.groovy

@@ -0,0 +1,96 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+// Verify that when fetchRemoteMeta receives a failed GlobListResult (e.g. S3 auth error),
+// the streaming job is correctly PAUSED rather than hanging indefinitely.
+suite("test_streaming_insert_job_fetch_meta_error", "nonConcurrent") {
+    def tableName = "test_streaming_insert_job_fetch_meta_error_tbl"
+    def jobName = "test_streaming_insert_job_fetch_meta_error_job"
+
+    sql """drop table if exists `${tableName}` force"""
+    sql """
+        DROP JOB IF EXISTS where jobname = '${jobName}'
+    """
+
+    sql """
+        CREATE TABLE IF NOT EXISTS ${tableName} (
+            `c1` int NULL,
+            `c2` string NULL,
+            `c3` int  NULL,
+        ) ENGINE=OLAP
+        DUPLICATE KEY(`c1`)
+        COMMENT 'OLAP'
+        DISTRIBUTED BY HASH(`c1`) BUCKETS 3
+        PROPERTIES ("replication_allocation" = "tag.location.default: 1");
+    """
+
+    GetDebugPoint().enableDebugPointForAllFEs("S3SourceOffsetProvider.fetchRemoteMeta.error")
+    try {
+        sql """
+           CREATE JOB ${jobName}
+           ON STREAMING DO INSERT INTO ${tableName}
+           SELECT * FROM S3
+            (
+                "uri" = "s3://${s3BucketName}/regression/load/data/example_[0-1].csv",
+                "format" = "csv",
+                "provider" = "${getS3Provider()}",
+                "column_separator" = ",",
+                "s3.endpoint" = "${getS3Endpoint()}",
+                "s3.region" = "${getS3Region()}",
+                "s3.access_key" = "${getS3AK()}",
+                "s3.secret_key" = "${getS3SK()}"
+            );
+        """
+
+        try {
+            Awaitility.await().atMost(120, SECONDS)
+                    .pollInterval(2, SECONDS).until(
+                    {
+                        def jobRes = sql """ select Status from jobs("type"="insert") where Name = '${jobName}' and ExecuteType='STREAMING' """
+                        log.info("jobRes: " + jobRes)
+                        jobRes.size() == 1 && 'PAUSED'.equals(jobRes.get(0).get(0))
+                    }
+            )
+        } catch (Exception ex) {
+            def showjob = sql """select * from jobs("type"="insert") where Name='${jobName}'"""
+            def showtask = sql """select * from tasks("type"="insert") where JobName='${jobName}'"""
+            log.info("show job: " + showjob)
+            log.info("show task: " + showtask)
+            throw ex
+        }
+
+        def jobStatus = sql """select Status, ErrorMsg from jobs("type"="insert") where Name='${jobName}'"""
+        assert jobStatus.get(0).get(0) == "PAUSED"
+        assert jobStatus.get(0).get(1).contains("Failed to list S3 files")
+
+        sql """
+            DROP JOB IF EXISTS where jobname = '${jobName}'
+        """
+
+        def jobCountRsp = sql """select count(1) from jobs("type"="insert") where Name='${jobName}'"""
+        assert jobCountRsp.get(0).get(0) == 0
+
+    } finally {
+        GetDebugPoint().disableDebugPointForAllFEs("S3SourceOffsetProvider.fetchRemoteMeta.error")
+        sql """DROP JOB IF EXISTS where jobname = '${jobName}'"""
+    }
+}