branch-4.1: [fix](fe) Keep CDC end offset consistent with current progress #65688

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

@@ -89,7 +89,7 @@ public class JdbcSourceOffsetProvider implements SourceOffsetProvider {
     List<SnapshotSplit> finishedSplits = new ArrayList<>();
 
     volatile JdbcOffset currentOffset;
-    Map<String, String> endBinlogOffset;
+    volatile Map<String, String> endBinlogOffset;
 
     @SerializedName("chw")
     // tableID -> splitId -> chunk of highWatermark
@@ -117,7 +117,10 @@ public class JdbcSourceOffsetProvider implements SourceOffsetProvider {
     /** Cache of Job.syncTables, set by initSplitProgress / replayIfNeed. */
     transient List<String> cachedSyncTables;
 
-    /** Guards cdcSplitProgress/committedSplitProgress/remainingSplits/finishedSplits. */
+    /**
+     * Guards cdcSplitProgress/committedSplitProgress/remainingSplits/finishedSplits
+     * and compound binlog updates to currentOffset/endBinlogOffset/hasMoreData.
+     */
     protected final transient Object splitsLock = new Object();
 
     /**
@@ -251,9 +254,14 @@ public void updateOffset(Offset offset) {
                 }
             }
         } else {
-            BinlogSplit binlogSplit = (BinlogSplit) newOffset.getSplits().get(0);
-            binlogOffsetPersist = new HashMap<>(binlogSplit.getStartingOffset());
-            binlogOffsetPersist.put(SPLIT_ID, BinlogSplit.BINLOG_SPLIT_ID);
+            synchronized (splitsLock) {
+                BinlogSplit binlogSplit = (BinlogSplit) newOffset.getSplits().get(0);
+                binlogOffsetPersist = new HashMap<>(binlogSplit.getStartingOffset());
+                binlogOffsetPersist.put(SPLIT_ID, BinlogSplit.BINLOG_SPLIT_ID);
+                currentOffset = newOffset;
+                hasMoreData = true;
+            }
+            return;
         }
         this.currentOffset = newOffset;
     }
@@ -286,11 +294,13 @@ public void fetchRemoteMeta(Map<String, String> properties) throws Exception {
             }
             Map<String, String> newEndOffset = parseCdcResponseData(
                     result.getResponse(), new TypeReference<Map<String, String>>() {});
-            // null→value also counts as a change: upstream may have advanced while fetch was blocked.
-            if (endBinlogOffset == null || !endBinlogOffset.equals(newEndOffset)) {
-                hasMoreData = true;
+            synchronized (splitsLock) {
+                // null→value also counts as a change: upstream may have advanced while fetch was blocked.
+                if (endBinlogOffset == null || !endBinlogOffset.equals(newEndOffset)) {
+                    hasMoreData = true;
+                }
+                endBinlogOffset = newEndOffset;
             }
-            endBinlogOffset = newEndOffset;
         } catch (TimeoutException te) {
             log.warn("cdc_client RPC timeout api=/api/fetchEndOffset jobId={} backend={}:{} timeout_sec={}",
                     getJobId(), backend.getHost(), backend.getBrpcPort(),
@@ -308,6 +318,8 @@ public boolean hasMoreDataToConsume() {
             return true;
         }
 
+        JdbcOffset currentOffsetAtCompare;
+        Map<String, String> endOffsetAtCompare;
         synchronized (splitsLock) {
             if (currentOffset.snapshotSplit()) {
                 if (!remainingSplits.isEmpty()) {
@@ -328,19 +340,33 @@ public boolean hasMoreDataToConsume() {
             if (CollectionUtils.isNotEmpty(remainingSplits)) {
                 return true;
             }
+            currentOffsetAtCompare = currentOffset;
+            endOffsetAtCompare = endBinlogOffset;
         }
-        if (MapUtils.isEmpty(endBinlogOffset)) {
+        if (MapUtils.isEmpty(endOffsetAtCompare)) {
             return false;
         }
         try {
-            if (!currentOffset.snapshotSplit()) {
-                BinlogSplit binlogSplit = (BinlogSplit) currentOffset.getSplits().get(0);
+            if (!currentOffsetAtCompare.snapshotSplit()) {
+                BinlogSplit binlogSplit = (BinlogSplit) currentOffsetAtCompare.getSplits().get(0);
                 if (MapUtils.isEmpty(binlogSplit.getStartingOffset())) {
                     // snapshot to binlog phase
                     return true;
                 }
-                hasMoreData = compareOffset(endBinlogOffset, new HashMap<>(binlogSplit.getStartingOffset()));
-                return hasMoreData;
+                Map<String, String> currentBinlogOffset = new HashMap<>(binlogSplit.getStartingOffset());
+                int compareResult = compareOffset(endOffsetAtCompare, currentBinlogOffset);
+                synchronized (splitsLock) {
+                    if (currentOffset != currentOffsetAtCompare || endBinlogOffset != endOffsetAtCompare) {
+                        // The stale result cannot prove that the latest offsets have caught up.
+                        hasMoreData = true;
+                        return true;
+                    }
+                    if (compareResult < 0) {
+                        endBinlogOffset = currentBinlogOffset;
+                    }
+                    hasMoreData = compareResult > 0;
+                    return hasMoreData;
+                }
             } else {
                 // snapshot means has data to consume
                 return true;
@@ -351,7 +377,7 @@ public boolean hasMoreDataToConsume() {
         }
     }
 
-    private boolean compareOffset(Map<String, String> offsetFirst, Map<String, String> offsetSecond)
+    protected int compareOffset(Map<String, String> offsetFirst, Map<String, String> offsetSecond)
             throws JobException {
         Backend backend = StreamingJobUtils.selectBackend(cloudCluster, boundBackendId);
         CompareOffsetRequest requestParams =
@@ -375,7 +401,7 @@ private boolean compareOffset(Map<String, String> offsetFirst, Map<String, Strin
             }
             Integer cmp = parseCdcResponseData(
                     result.getResponse(), new TypeReference<Integer>() {});
-            return cmp != null && cmp > 0;
+            return cmp;
         } catch (TimeoutException te) {
             log.warn("cdc_client RPC timeout api=/api/compareOffset jobId={} backend={}:{} timeout_sec={}",
                     getJobId(), backend.getHost(), backend.getBrpcPort(),

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

@@ -288,12 +288,17 @@ public void updateOffset(Offset offset) {
                 }
             }
         } else {
-            // Mirror binlog offset into bop so it survives FE checkpoint
-            BinlogSplit bs = (BinlogSplit) newOffset.getSplits().get(0);
-            if (MapUtils.isNotEmpty(bs.getStartingOffset())) {
-                binlogOffsetPersist = new HashMap<>(bs.getStartingOffset());
-                binlogOffsetPersist.put(SPLIT_ID, BinlogSplit.BINLOG_SPLIT_ID);
+            synchronized (splitsLock) {
+                // Mirror binlog offset into bop so it survives FE checkpoint
+                BinlogSplit bs = (BinlogSplit) newOffset.getSplits().get(0);
+                if (MapUtils.isNotEmpty(bs.getStartingOffset())) {
+                    binlogOffsetPersist = new HashMap<>(bs.getStartingOffset());
+                    binlogOffsetPersist.put(SPLIT_ID, BinlogSplit.BINLOG_SPLIT_ID);
+                }
+                currentOffset = newOffset;
+                hasMoreData = true;
             }
+            return;
         }
         this.currentOffset = newOffset;
     }

fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderOffsetTest.java

@@ -0,0 +1,139 @@
+// 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.offset.jdbc;
+
+import org.apache.doris.job.cdc.split.BinlogSplit;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.Map;
+
+public class JdbcSourceOffsetProviderOffsetTest {
+
+    @Test
+    public void testEndOffsetAdvancesWhenCurrentOffsetIsAhead() {
+        assertEndOffsetAdvancesWhenCurrentOffsetIsAhead(new TestJdbcSourceOffsetProvider(-1));
+    }
+
+    @Test
+    public void testTvfEndOffsetAdvancesWhenCurrentOffsetIsAhead() {
+        assertEndOffsetAdvancesWhenCurrentOffsetIsAhead(new TestJdbcTvfSourceOffsetProvider(-1));
+    }
+
+    @Test
+    public void testEndOffsetRemainsWhenItIsAheadOfCurrentOffset() {
+        JdbcSourceOffsetProvider provider = new TestJdbcSourceOffsetProvider(1);
+        Map<String, String> currentOffset = Collections.singletonMap("lsn", "100");
+        Map<String, String> endOffset = Collections.singletonMap("lsn", "200");
+        provider.setEndBinlogOffset(endOffset);
+        provider.setHasMoreData(false);
+        provider.updateOffset(new JdbcOffset(
+                Collections.singletonList(new BinlogSplit(currentOffset))));
+
+        Assert.assertTrue(provider.hasMoreDataToConsume());
+        Assert.assertEquals(endOffset, provider.getEndBinlogOffset());
+    }
+
+    @Test
+    public void testStaleCompareDoesNotOverwriteRefreshedEndOffset() {
+        JdbcSourceOffsetProvider provider = new RefreshingEndOffsetProvider();
+        provider.setEndBinlogOffset(Collections.singletonMap("lsn", "100"));
+        provider.updateOffset(new JdbcOffset(
+                Collections.singletonList(new BinlogSplit(Collections.singletonMap("lsn", "200")))));
+
+        Assert.assertTrue(provider.hasMoreDataToConsume());
+        Assert.assertTrue(provider.hasMoreData);
+        Assert.assertEquals(Collections.singletonMap("lsn", "300"), provider.getEndBinlogOffset());
+    }
+
+    @Test
+    public void testStaleCompareDoesNotOverwriteAlteredCurrentOffsetState() {
+        JdbcSourceOffsetProvider provider = new AlteringCurrentOffsetProvider();
+        provider.setEndBinlogOffset(Collections.singletonMap("lsn", "200"));
+        provider.updateOffset(new JdbcOffset(
+                Collections.singletonList(new BinlogSplit(Collections.singletonMap("lsn", "200")))));
+
+        Assert.assertTrue(provider.hasMoreDataToConsume());
+        Assert.assertTrue(provider.hasMoreData);
+        Assert.assertEquals(Collections.singletonMap("lsn", "100"),
+                ((BinlogSplit) provider.currentOffset.getSplits().get(0)).getStartingOffset());
+    }
+
+    private static void assertEndOffsetAdvancesWhenCurrentOffsetIsAhead(JdbcSourceOffsetProvider provider) {
+        Map<String, String> staleEndOffset = Collections.singletonMap("lsn", "100");
+        Map<String, String> committedOffset = Collections.singletonMap("lsn", "200");
+        provider.setEndBinlogOffset(staleEndOffset);
+        provider.setHasMoreData(false);
+
+        provider.updateOffset(new JdbcOffset(
+                Collections.singletonList(new BinlogSplit(committedOffset))));
+
+        Assert.assertEquals(staleEndOffset, provider.getEndBinlogOffset());
+        Assert.assertFalse(provider.hasMoreDataToConsume());
+        Assert.assertEquals(committedOffset, provider.getEndBinlogOffset());
+        Assert.assertEquals("{\"lsn\":\"200\"}", provider.getShowMaxOffset());
+    }
+
+    private static class TestJdbcSourceOffsetProvider extends JdbcSourceOffsetProvider {
+        private final int compareResult;
+
+        TestJdbcSourceOffsetProvider(int compareResult) {
+            this.compareResult = compareResult;
+        }
+
+        @Override
+        protected int compareOffset(Map<String, String> offsetFirst, Map<String, String> offsetSecond) {
+            return compareResult;
+        }
+    }
+
+    private static class TestJdbcTvfSourceOffsetProvider extends JdbcTvfSourceOffsetProvider {
+        private final int compareResult;
+
+        TestJdbcTvfSourceOffsetProvider(int compareResult) {
+            this.compareResult = compareResult;
+        }
+
+        @Override
+        protected int compareOffset(Map<String, String> offsetFirst, Map<String, String> offsetSecond) {
+            return compareResult;
+        }
+    }
+
+    private static class RefreshingEndOffsetProvider extends JdbcSourceOffsetProvider {
+        @Override
+        protected int compareOffset(Map<String, String> offsetFirst, Map<String, String> offsetSecond) {
+            synchronized (splitsLock) {
+                endBinlogOffset = Collections.singletonMap("lsn", "300");
+                hasMoreData = false;
+            }
+            return -1;
+        }
+    }
+
+    private static class AlteringCurrentOffsetProvider extends JdbcSourceOffsetProvider {
+        @Override
+        protected int compareOffset(Map<String, String> offsetFirst, Map<String, String> offsetSecond) {
+            updateOffset(new JdbcOffset(
+                    Collections.singletonList(new BinlogSplit(Collections.singletonMap("lsn", "100")))));
+            return 0;
+        }
+    }
+}

regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_postgres_latest_alter_cred.groovy

@@ -104,11 +104,14 @@ suite("test_streaming_job_cdc_stream_postgres_latest_alter_cred",
             )
         """
 
-        // wait for job to reach RUNNING state (at least one task scheduled)
+        // wait for the first task to commit the resolved latest offset
         try {
-            Awaitility.await().atMost(60, SECONDS).pollInterval(1, SECONDS).until({
-                def status = sql """select status from jobs("type"="insert") where Name='${jobName}'"""
-                status.size() == 1 && status.get(0).get(0) == "RUNNING"
+            Awaitility.await().atMost(300, SECONDS).pollInterval(1, SECONDS).until({
+                def jobInfo = sql """select Status, SucceedTaskCount from jobs("type"="insert")
+                                     where Name='${jobName}' and ExecuteType='STREAMING'"""
+                log.info("job readiness for latest offset: " + jobInfo)
+                jobInfo.size() == 1 && jobInfo.get(0).get(0) == "RUNNING"
+                        && (jobInfo.get(0).get(1) as int) >= 1
             })
         } catch (Exception ex) {
             log.info("job: " + (sql """select * from jobs("type"="insert") where Name='${jobName}'"""))