This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new c83166309e4 [fix](fe) Keep CDC end offset consistent with current
progress (#65688)
c83166309e4 is described below
commit c83166309e4444bddd71cb79553ca12399ba4786
Author: wudi <[email protected]>
AuthorDate: Mon Jul 20 16:19:38 2026 +0800
[fix](fe) Keep CDC end offset consistent with current progress (#65688)
### What problem does this PR solve?
Streaming CDC jobs periodically fetch the source end offset, while a
successful task commits its current offset independently. A task can
advance beyond the previously fetched end offset. The FE previously
converted the offset comparison into a boolean, so it could stop
scheduling but could not distinguish equality from the current offset
being ahead, leaving CurrentOffset greater than EndOffset in job output.
This change preserves the three-way comparison result and advances
EndOffset to CurrentOffset when the fetched end is behind. End-offset
publication is synchronized so an obsolete comparison result cannot
overwrite a newer fetched end. The PostgreSQL latest-offset credential
regression case now waits for the first task to commit the resolved
latest offset before inserting new rows.
---
.../job/offset/jdbc/JdbcSourceOffsetProvider.java | 58 ++++++---
.../offset/jdbc/JdbcTvfSourceOffsetProvider.java | 15 ++-
.../jdbc/JdbcSourceOffsetProviderOffsetTest.java | 139 +++++++++++++++++++++
...ob_cdc_stream_postgres_latest_alter_cred.groovy | 11 +-
4 files changed, 198 insertions(+), 25 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java
index 0a9e8df0921..13874a35101 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java
+++
b/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 class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
}
}
} 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 class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
}
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 class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
return true;
}
+ JdbcOffset currentOffsetAtCompare;
+ Map<String, String> endOffsetAtCompare;
synchronized (splitsLock) {
if (currentOffset.snapshotSplit()) {
if (!remainingSplits.isEmpty()) {
@@ -328,19 +340,33 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
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 class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
}
}
- 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 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
}
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(),
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
index 6cbd2e63728..2376d3bd65b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
@@ -288,12 +288,17 @@ public class JdbcTvfSourceOffsetProvider extends
JdbcSourceOffsetProvider {
}
}
} 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;
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderOffsetTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderOffsetTest.java
new file mode 100644
index 00000000000..6efb9959748
--- /dev/null
+++
b/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;
+ }
+ }
+}
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_postgres_latest_alter_cred.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_postgres_latest_alter_cred.groovy
index f1cd5e13f91..87171eb369f 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_postgres_latest_alter_cred.groovy
+++
b/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}'"""))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
