[fix](streaming-job) streaming CDC reliability fixes (scheduling latch, fetch-meta reason, PG db name, flaky case)

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -1199,6 +1199,10 @@ public class Config extends ConfigBase {
     @ConfField(mutable = true, masterOnly = true)
     public static int streaming_cdc_heavy_rpc_timeout_sec = 600;
 
+    // Max byte length of a PG database name for a CDC job; raise only for a larger NAMEDATALEN build.
+    @ConfField(mutable = true, masterOnly = true)
+    public static int streaming_pg_max_identifier_length = 63;
+
     @ConfField(mutable = true, masterOnly = true)
     public static int streaming_cdc_fetch_splits_batch_size = 100;
 

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

@@ -17,6 +17,7 @@
 
 package org.apache.doris.job.extensions.insert.streaming;
 
+import org.apache.doris.common.Config;
 import org.apache.doris.datasource.jdbc.client.JdbcClient;
 import org.apache.doris.job.cdc.DataSourceConfigKeys;
 import org.apache.doris.job.common.DataSourceType;
@@ -25,6 +26,7 @@
 
 import org.apache.commons.lang3.StringUtils;
 
+import java.nio.charset.StandardCharsets;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
@@ -44,6 +46,8 @@ public class PostgresResourceValidator {
 
     public static void validate(Map<String, String> sourceProperties, String jobId, List<String> tableNames)
             throws JobException {
+        // PG truncates an over-long db name, so the slot lookup never matches it; reject up front.
+        checkDatabaseNameLength(sourceProperties.get(DataSourceConfigKeys.DATABASE));
         String slotName = resolveSlotName(sourceProperties, jobId);
         String publicationName = resolvePublicationName(sourceProperties, jobId);
         // Pattern-match ownership: name equals the default = Doris-owned (auto); otherwise user.
@@ -123,6 +127,19 @@ private static String resolvePublicationName(Map<String, String> config, String
         return StringUtils.isNotBlank(name) ? name : DataSourceConfigKeys.defaultPublicationName(jobId);
     }
 
+    private static void checkDatabaseNameLength(String database) throws JobException {
+        if (StringUtils.isBlank(database)) {
+            return;
+        }
+        // PG measures the identifier limit in bytes (NAMEDATALEN-1), so compare encoded bytes.
+        int bytes = database.getBytes(StandardCharsets.UTF_8).length;
+        if (bytes > Config.streaming_pg_max_identifier_length) {
+            throw new JobException("database name '" + database + "' is " + bytes + " bytes, exceeding "
+                    + Config.streaming_pg_max_identifier_length + "; PostgreSQL truncates it and the"
+                    + " replication-slot lookup would fail.");
+        }
+    }
+
     private static boolean publicationExists(Connection conn, String publicationName) throws Exception {
         try (PreparedStatement ps = conn.prepareStatement("SELECT 1 FROM pg_publication WHERE pubname = ?")) {
             ps.setString(1, publicationName);

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

@@ -709,12 +709,12 @@ protected void fetchMeta() throws JobException {
                     || !InternalErrorCode.MANUAL_PAUSE_ERR.equals(this.getFailureReason().getCode())) {
                 // When a job is manually paused, it does not need to be set again,
                 // otherwise, it may be woken up by auto resume.
+                // Pause before setting the reason: updateJobStatus's writeLock orders this after any
+                // task-success callback that clears failureReason, so a success can't wipe the reason.
+                this.updateJobStatus(JobStatus.PAUSED);
                 this.setFailureReason(
                         new FailureReason(InternalErrorCode.GET_REMOTE_DATA_ERROR,
                                 "Failed to fetch meta, " + ex.getMessage()));
-                // If fetching meta fails, the job is paused
-                // and auto resume will automatically wake it up.
-                this.updateJobStatus(JobStatus.PAUSED);
 
                 if (MetricRepo.isInit) {
                     MetricRepo.COUNTER_STREAMING_JOB_GET_META_FAIL_COUNT.increase(1L);

fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java

@@ -924,13 +924,6 @@ protected long extractEventTimeMs(Map<String, String> offsetMap) {
         return -1;
     }
 
-    @Override
-    public void onTaskCommitted(long scannedRows, long loadBytes) {
-        if (scannedRows == 0 && loadBytes == 0) {
-            hasMoreData = false;
-        }
-    }
-
     @Override
     public boolean hasReachedEnd() {
         if (!isSnapshotOnlyMode()) {

fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/PostgresResourceValidatorTest.java

@@ -0,0 +1,45 @@
+// 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.
+
+package org.apache.doris.job.extensions.insert.streaming;
+
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.exception.JobException;
+
+import org.apache.commons.lang3.StringUtils;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class PostgresResourceValidatorTest {
+
+    // A 22-char CJK database name is length()==22 but 66 bytes in UTF-8; PG truncates it to 63 bytes.
+    // The byte-based check must reject it before connecting (validate fails on the very first line).
+    @Test
+    public void testRejectMultibyteOverLongDatabaseName() {
+        String dbName = StringUtils.repeat("库", 22);
+        Assert.assertEquals(22, dbName.length());
+        Map<String, String> props = new HashMap<>();
+        props.put(DataSourceConfigKeys.DATABASE, dbName);
+        JobException e = Assert.assertThrows(JobException.class,
+                () -> PostgresResourceValidator.validate(props, "1", Collections.emptyList()));
+        Assert.assertTrue(e.getMessage(), e.getMessage().contains("bytes"));
+    }
+}

regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_special_offset.groovy

@@ -143,6 +143,37 @@ suite("test_streaming_postgres_job_special_offset", "p0,external,pg,external_doc
             def jobStatus = sql """select status from jobs("type"="insert") where Name='${jobName}'"""
             return jobStatus[0][0] == "PAUSED"
         })
+
+        // PAUSE cancels the FE task but does NOT stop the in-flight cdc_client reader started before
+        // the pause: it keeps polling up to max_interval (default 10s) and would stream-load whatever
+        // rows we insert next, defeating the ALTER-offset reposition (this is the historical source of
+        // flakiness). No new reader is dispatched while PAUSED, so drain the lingering one
+        // deterministically: insert a disposable probe row, then wait until the row count stays stable
+        // for a window longer than max_interval. An active reader would have loaded the probe within
+        // max_interval, so a stable window proves no reader is consuming anymore. Finally delete the
+        // probe from both sides so it never pollutes the before/after-mark assertions or the .out.
+        def probeId = 10
+        connect("${pgUser}", "${pgPassword}", "jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
+            sql """INSERT INTO ${pgDB}.${pgSchema}.${table1} VALUES (${probeId}, 'drain_probe')"""
+        }
+        int stablePolls = 0
+        long lastCnt = -1L
+        Awaitility.await().atMost(120, SECONDS).pollInterval(3, SECONDS).until({
+            long c = sql("""SELECT count(*) FROM ${currentDb}.${table1}""")[0][0] as long
+            if (c == lastCnt) {
+                stablePolls++
+            } else {
+                stablePolls = 0
+                lastCnt = c
+            }
+            // 5 * 3s = 15s stable > max_interval (10s) => the lingering reader has fully drained.
+            return stablePolls >= 5
+        })
+        connect("${pgUser}", "${pgPassword}", "jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
+            sql """DELETE FROM ${pgDB}.${pgSchema}.${table1} WHERE id = ${probeId}"""
+        }
+        sql """DELETE FROM ${currentDb}.${table1} WHERE id = ${probeId}"""
+
         def alterLsn = ""
         connect("${pgUser}", "${pgPassword}", "jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
             // insert data BEFORE the LSN mark