This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 7dd7b499fe [Fix][CDC] Avoid replaying completed snapshot splits after
restore (#10859)
7dd7b499fe is described below
commit 7dd7b499fe68a98c8af1a1fe0b48c77a79e45817
Author: Daniel <[email protected]>
AuthorDate: Wed May 27 23:01:36 2026 +0800
[Fix][CDC] Avoid replaying completed snapshot splits after restore (#10859)
---
.../source/enumerator/SnapshotSplitAssigner.java | 26 +++-
.../enumerator/SnapshotSplitAssignerTest.java | 171 +++++++++++++++++++++
2 files changed, 194 insertions(+), 3 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/SnapshotSplitAssigner.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/SnapshotSplitAssigner.java
index c16dd81102..cac8c1d128 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/SnapshotSplitAssigner.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/SnapshotSplitAssigner.java
@@ -218,11 +218,18 @@ public class SnapshotSplitAssigner<C extends
SourceConfig> implements SplitAssig
@Override
public void addSplits(Collection<SourceSplitBase> splits) {
for (SourceSplitBase split : splits) {
- remainingSplits.add(split.asSnapshotSplit());
+ SnapshotSplit snapshotSplit = split.asSnapshotSplit();
+ if (hasCheckpointedCompletionState(snapshotSplit)) {
+ LOG.info(
+ "Ignore add-back for completed snapshot split {}, keep
checkpointed completion state",
+ snapshotSplit.splitId());
+ continue;
+ }
+ remainingSplits.add(snapshotSplit);
// we should remove the add-backed splits from the assigned list,
because they are
// failed
- assignedSplits.remove(split.splitId());
- splitCompletedOffsets.remove(split.splitId());
+ assignedSplits.remove(snapshotSplit.splitId());
+ splitCompletedOffsets.remove(snapshotSplit.splitId());
}
}
@@ -283,6 +290,19 @@ public class SnapshotSplitAssigner<C extends SourceConfig>
implements SplitAssig
return noMoreSplits() && assignedSplits.size() ==
splitCompletedOffsets.size();
}
+ /**
+ * Returns whether the restored split already has durable completion state
in the checkpoint.
+ *
+ * <p>We can safely skip add-back only when the checkpoint persisted both
the original
+ * assignment and the finished watermark. If either side is missing, the
split still needs to be
+ * replayed after restore so the enumerator can rebuild the missing
completion state.
+ */
+ private boolean hasCheckpointedCompletionState(SnapshotSplit
snapshotSplit) {
+ return snapshotSplit.isSnapshotReadFinished()
+ && assignedSplits.containsKey(snapshotSplit.splitId())
+ && splitCompletedOffsets.containsKey(snapshotSplit.splitId());
+ }
+
@VisibleForTesting
Map<String, SnapshotSplit> getAssignedSplits() {
return assignedSplits;
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/SnapshotSplitAssignerTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/SnapshotSplitAssignerTest.java
new file mode 100644
index 0000000000..6e7ac75862
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/SnapshotSplitAssignerTest.java
@@ -0,0 +1,171 @@
+/*
+ * 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.seatunnel.connectors.cdc.base.source.enumerator;
+
+import
org.apache.seatunnel.connectors.cdc.base.source.enumerator.state.SnapshotPhaseState;
+import
org.apache.seatunnel.connectors.cdc.base.source.event.SnapshotSplitWatermark;
+import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset;
+import org.apache.seatunnel.connectors.cdc.base.source.split.SnapshotSplit;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import io.debezium.relational.TableId;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class SnapshotSplitAssignerTest {
+
+ @Test
+ public void
testAddSplitsShouldKeepCompletedFinishedSplitOutOfRemainingQueue() {
+ SnapshotSplit finishedSplit =
createFinishedSnapshotSplit("db1.table1.1");
+ Map<String, SnapshotSplit> assignedSplits = new HashMap<>();
+ assignedSplits.put(finishedSplit.splitId(), finishedSplit);
+ Map<String, SnapshotSplitWatermark> completedOffsets = new HashMap<>();
+ completedOffsets.put(
+ finishedSplit.splitId(),
+ new SnapshotSplitWatermark(
+ finishedSplit.splitId(),
+ finishedSplit.getLowWatermark(),
+ finishedSplit.getHighWatermark()));
+
+ SnapshotSplitAssigner<?> splitAssigner =
+ createRestoredSnapshotSplitAssigner(assignedSplits,
completedOffsets);
+
+ splitAssigner.addSplits(Collections.singletonList(finishedSplit));
+
+ SnapshotPhaseState state = splitAssigner.snapshotState(11L);
+ Assertions.assertTrue(state.getRemainingSplits().isEmpty());
+ Assertions.assertEquals(
+ Collections.singleton(finishedSplit.splitId()),
state.getAssignedSplits().keySet());
+ Assertions.assertEquals(
+ Collections.singleton(finishedSplit.splitId()),
+ state.getSplitCompletedOffsets().keySet());
+ Assertions.assertFalse(splitAssigner.waitingForCompletedSplits());
+
+ splitAssigner.notifyCheckpointComplete(11L);
+ Assertions.assertTrue(splitAssigner.isCompleted());
+ }
+
+ @Test
+ public void
testAddSplitsShouldReplayFinishedSplitWithoutCompletedWatermark() {
+ SnapshotSplit finishedSplit =
createFinishedSnapshotSplit("db1.table1.1");
+ Map<String, SnapshotSplit> assignedSplits = new HashMap<>();
+ assignedSplits.put(finishedSplit.splitId(), finishedSplit);
+
+ SnapshotSplitAssigner<?> splitAssigner =
+ createRestoredSnapshotSplitAssigner(assignedSplits, new
HashMap<>());
+
+ splitAssigner.addSplits(Collections.singletonList(finishedSplit));
+
+ SnapshotPhaseState state = splitAssigner.snapshotState(12L);
+ Assertions.assertEquals(1, state.getRemainingSplits().size());
+ Assertions.assertEquals(
+ finishedSplit.splitId(),
state.getRemainingSplits().get(0).splitId());
+ Assertions.assertTrue(state.getAssignedSplits().isEmpty());
+ Assertions.assertTrue(state.getSplitCompletedOffsets().isEmpty());
+ Assertions.assertTrue(splitAssigner.waitingForCompletedSplits());
+ }
+
+ @Test
+ public void
testRestoreAfterCheckpointedCompletionShouldKeepFinishedSplitOutOfReplayQueue()
{
+ SnapshotSplit finishedSplit =
createFinishedSnapshotSplit("db1.table1.2");
+ Map<String, SnapshotSplit> assignedSplits = new HashMap<>();
+ assignedSplits.put(finishedSplit.splitId(), finishedSplit);
+ SnapshotSplitAssigner<?> runningAssigner =
+ createRestoredSnapshotSplitAssigner(assignedSplits, new
HashMap<>());
+
+ runningAssigner.onCompletedSplits(
+ Collections.singletonList(createWatermark(finishedSplit)));
+ SnapshotPhaseState checkpointState =
runningAssigner.snapshotState(13L);
+
+ SnapshotSplitAssigner<?> restoredAssigner =
+ createRestoredSnapshotSplitAssigner(
+ checkpointState.getAssignedSplits(),
+ checkpointState.getSplitCompletedOffsets());
+
+ restoredAssigner.addSplits(Collections.singletonList(finishedSplit));
+
+ SnapshotPhaseState restoredState = restoredAssigner.snapshotState(14L);
+ Assertions.assertTrue(restoredState.getRemainingSplits().isEmpty());
+ Assertions.assertEquals(
+ Collections.singleton(finishedSplit.splitId()),
+ restoredState.getAssignedSplits().keySet());
+ Assertions.assertEquals(
+ Collections.singleton(finishedSplit.splitId()),
+ restoredState.getSplitCompletedOffsets().keySet());
+ Assertions.assertFalse(restoredAssigner.waitingForCompletedSplits());
+ }
+
+ private SnapshotSplitAssigner<?> createRestoredSnapshotSplitAssigner(
+ Map<String, SnapshotSplit> assignedSplits,
+ Map<String, SnapshotSplitWatermark> completedOffsets) {
+ SnapshotPhaseState checkpointState =
+ new SnapshotPhaseState(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ assignedSplits,
+ completedOffsets,
+ false,
+ Collections.emptyList(),
+ false,
+ true);
+ SplitAssigner.Context<?> context =
+ new SplitAssigner.Context<>(
+ null,
+ Collections.singleton(TableId.parse("db1.table1")),
+ checkpointState.getAssignedSplits(),
+ checkpointState.getSplitCompletedOffsets());
+ return new SnapshotSplitAssigner<>(context, 10, checkpointState, null);
+ }
+
+ private SnapshotSplit createFinishedSnapshotSplit(String splitId) {
+ return new SnapshotSplit(
+ splitId,
+ TableId.parse("db1.table1"),
+ null,
+ null,
+ null,
+ new TestOffset(1L),
+ new TestOffset(2L));
+ }
+
+ private SnapshotSplitWatermark createWatermark(SnapshotSplit
finishedSplit) {
+ return new SnapshotSplitWatermark(
+ finishedSplit.splitId(),
+ finishedSplit.getLowWatermark(),
+ finishedSplit.getHighWatermark());
+ }
+
+ private static final class TestOffset extends Offset {
+ private static final long serialVersionUID = 1L;
+
+ private TestOffset(long value) {
+ this.offset = Collections.singletonMap("pos",
String.valueOf(value));
+ }
+
+ @Override
+ public int compareTo(Offset other) {
+ return Long.compare(
+ Long.parseLong(this.offset.get("pos")),
+ Long.parseLong(other.getOffset().get("pos")));
+ }
+ }
+}
