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 0a9e8df092133b..13874a35101825 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 finishedSplits = new ArrayList<>(); volatile JdbcOffset currentOffset; - Map endBinlogOffset; + volatile Map 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 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 properties) throws Exception { } Map newEndOffset = parseCdcResponseData( result.getResponse(), new TypeReference>() {}); - // 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 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 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 offsetFirst, Map offsetSecond) + protected int compareOffset(Map offsetFirst, Map offsetSecond) throws JobException { Backend backend = StreamingJobUtils.selectBackend(cloudCluster, boundBackendId); CompareOffsetRequest requestParams = @@ -375,7 +401,7 @@ private boolean compareOffset(Map offsetFirst, Map() {}); - 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 6cbd2e63728db2..2376d3bd65ba20 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 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; } 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 00000000000000..6efb9959748bcd --- /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 currentOffset = Collections.singletonMap("lsn", "100"); + Map 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 staleEndOffset = Collections.singletonMap("lsn", "100"); + Map 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 offsetFirst, Map offsetSecond) { + return compareResult; + } + } + + private static class TestJdbcTvfSourceOffsetProvider extends JdbcTvfSourceOffsetProvider { + private final int compareResult; + + TestJdbcTvfSourceOffsetProvider(int compareResult) { + this.compareResult = compareResult; + } + + @Override + protected int compareOffset(Map offsetFirst, Map offsetSecond) { + return compareResult; + } + } + + private static class RefreshingEndOffsetProvider extends JdbcSourceOffsetProvider { + @Override + protected int compareOffset(Map offsetFirst, Map offsetSecond) { + synchronized (splitsLock) { + endBinlogOffset = Collections.singletonMap("lsn", "300"); + hasMoreData = false; + } + return -1; + } + } + + private static class AlteringCurrentOffsetProvider extends JdbcSourceOffsetProvider { + @Override + protected int compareOffset(Map offsetFirst, Map 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 f1cd5e13f91c63..87171eb369fdb1 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}'"""))