This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 bbde250914 [Fix][CDC] Reconstruct missing snapshot completion
watermark and always wait for a checkpoint (#12162)
bbde250914 is described below
commit bbde250914346068f50131e7907bef2c9ecbc448
Author: Jast <[email protected]>
AuthorDate: Fri Sep 11 05:15:03 2026 +0000
[Fix][CDC] Reconstruct missing snapshot completion watermark and always
wait for a checkpoint (#12162)
Co-authored-by: zhangshenghang <[email protected]>
---
.../source/enumerator/SnapshotSplitAssigner.java | 53 +++++++++++-----
.../enumerator/SnapshotSplitAssignerTest.java | 73 +++++++++++++++++++---
2 files changed, 105 insertions(+), 21 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 cac8c1d128..d5d132e727 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
@@ -202,15 +202,25 @@ public class SnapshotSplitAssigner<C extends
SourceConfig> implements SplitAssig
completedSplitWatermarks.forEach(
watermark ->
this.splitCompletedOffsets.put(watermark.getSplitId(), watermark));
if (allSplitsCompleted()) {
- // Skip the waiting checkpoint when current parallelism is 1 which
means we do not need
- // to care about the global output data order of snapshot splits
and incremental split.
if (currentParallelism == 1) {
+ // A single-reader job completes immediately. Zeta disables
checkpointing
+ // entirely for batch jobs without 'checkpoint.interval', so
waiting for
+ // notifyCheckpointComplete would hang such a job forever. The
failover risk
+ // of skipping the checkpoint wait is covered by the durable
finished-unacked
+ // splits in the reader's own checkpoint, whose re-report on
restore and the
+ // back-fill in restoreCompletedSnapshotSplit reconstruct the
completion state
+ // without replaying the splits.
assignerCompleted = true;
LOG.info(
- "Snapshot split assigner received all splits completed
and the job parallelism is 1, snapshot split assigner is turn into completed
status.");
+ "Snapshot split assigner received all splits completed
at parallelism 1, snapshot split assigner is turn into completed status.");
} else {
+ // Multi-reader jobs must wait for a complete checkpoint
before switching to
+ // the incremental phase, so that all records of snapshot
splits are completely
+ // processed in the pipeline and no incremental record can
overtake a snapshot
+ // record of the same key.
LOG.info(
- "Snapshot split assigner received all splits
completed, waiting for a complete checkpoint to mark the assigner completed.");
+ "Snapshot split assigner received all splits completed
at parallelism {}, waiting for a complete checkpoint to mark the assigner
completed.",
+ currentParallelism);
}
}
}
@@ -219,9 +229,9 @@ public class SnapshotSplitAssigner<C extends SourceConfig>
implements SplitAssig
public void addSplits(Collection<SourceSplitBase> splits) {
for (SourceSplitBase split : splits) {
SnapshotSplit snapshotSplit = split.asSnapshotSplit();
- if (hasCheckpointedCompletionState(snapshotSplit)) {
+ if (restoreCompletedSnapshotSplit(snapshotSplit)) {
LOG.info(
- "Ignore add-back for completed snapshot split {}, keep
checkpointed completion state",
+ "Restore completed snapshot split {} from checkpoint
without replaying it",
snapshotSplit.splitId());
continue;
}
@@ -291,16 +301,31 @@ public class SnapshotSplitAssigner<C extends
SourceConfig> implements SplitAssig
}
/**
- * Returns whether the restored split already has durable completion state
in the checkpoint.
+ * Returns whether the restored split already has durable completion state
in the checkpoint,
+ * and back-fills any missing completion watermark from the split itself.
*
- * <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.
+ * <p>A finished reader that never reported its watermark before failover
(for example because
+ * the reader crashed immediately after marking the split as
snapshot-read-finished but before
+ * its next CompletedSnapshotSplitsReportEvent went out) can show up after
restore as a finished
+ * split whose watermark has not yet been checkpointed. Without back-fill,
the enumerator would
+ * re-enqueue the split and the snapshot phase would never finish.
+ *
+ * <p>The split is skipped on add-back in that case, and the missing
watermark is reconstructed
+ * from the split's own low/high watermark. If the split was not finished
before the failover we
+ * still re-enqueue it so the reader can replay it from its persisted
state.
*/
- private boolean hasCheckpointedCompletionState(SnapshotSplit
snapshotSplit) {
- return snapshotSplit.isSnapshotReadFinished()
- && assignedSplits.containsKey(snapshotSplit.splitId())
- && splitCompletedOffsets.containsKey(snapshotSplit.splitId());
+ private boolean restoreCompletedSnapshotSplit(SnapshotSplit snapshotSplit)
{
+ if (!snapshotSplit.isSnapshotReadFinished()
+ || !assignedSplits.containsKey(snapshotSplit.splitId())) {
+ return false;
+ }
+ splitCompletedOffsets.putIfAbsent(
+ snapshotSplit.splitId(),
+ new SnapshotSplitWatermark(
+ snapshotSplit.splitId(),
+ snapshotSplit.getLowWatermark(),
+ snapshotSplit.getHighWatermark()));
+ return true;
}
@VisibleForTesting
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
index 6e7ac75862..8cd030a542 100644
---
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
@@ -65,7 +65,7 @@ public class SnapshotSplitAssignerTest {
}
@Test
- public void
testAddSplitsShouldReplayFinishedSplitWithoutCompletedWatermark() {
+ public void
testAddSplitsShouldRestoreFinishedSplitWithoutCompletedWatermark() {
SnapshotSplit finishedSplit =
createFinishedSnapshotSplit("db1.table1.1");
Map<String, SnapshotSplit> assignedSplits = new HashMap<>();
assignedSplits.put(finishedSplit.splitId(), finishedSplit);
@@ -76,12 +76,17 @@ public class SnapshotSplitAssignerTest {
splitAssigner.addSplits(Collections.singletonList(finishedSplit));
SnapshotPhaseState state = splitAssigner.snapshotState(12L);
- Assertions.assertEquals(1, state.getRemainingSplits().size());
+ // The split was already finished in the reader before the failover
but its
+ // completed-watermark was never checkpointed. The assigner must
reconstruct the
+ // watermark from the split itself and skip add-back, otherwise the
snapshot phase
+ // would never finish.
+ Assertions.assertTrue(state.getRemainingSplits().isEmpty());
+ Assertions.assertEquals(
+ Collections.singleton(finishedSplit.splitId()),
state.getAssignedSplits().keySet());
Assertions.assertEquals(
- finishedSplit.splitId(),
state.getRemainingSplits().get(0).splitId());
- Assertions.assertTrue(state.getAssignedSplits().isEmpty());
- Assertions.assertTrue(state.getSplitCompletedOffsets().isEmpty());
- Assertions.assertTrue(splitAssigner.waitingForCompletedSplits());
+ Collections.singleton(finishedSplit.splitId()),
+ state.getSplitCompletedOffsets().keySet());
+ Assertions.assertFalse(splitAssigner.waitingForCompletedSplits());
}
@Test
@@ -114,9 +119,63 @@ public class SnapshotSplitAssignerTest {
Assertions.assertFalse(restoredAssigner.waitingForCompletedSplits());
}
+ @Test
+ public void
testSingleParallelismShouldCompleteImmediatelyWhenAllSplitsCompleted() {
+ // A single-reader batch job may run with checkpointing disabled (no
+ // 'checkpoint.interval' in env), so the assigner can never rely on
+ // notifyCheckpointComplete to flip the completed flag. It must
complete immediately
+ // once all snapshot splits have reported their watermarks, otherwise
such jobs hang
+ // forever between the snapshot and incremental phases.
+ SnapshotSplit finishedSplit =
createFinishedSnapshotSplit("db1.table1.1");
+ Map<String, SnapshotSplit> assignedSplits = new HashMap<>();
+ assignedSplits.put(finishedSplit.splitId(), finishedSplit);
+
+ SnapshotSplitAssigner<?> splitAssigner =
+ createRestoredSnapshotSplitAssigner(assignedSplits, new
HashMap<>(), 1);
+
+ Assertions.assertFalse(splitAssigner.isCompleted());
+
+
splitAssigner.onCompletedSplits(Collections.singletonList(createWatermark(finishedSplit)));
+
+ Assertions.assertTrue(splitAssigner.isCompleted());
+ Assertions.assertFalse(splitAssigner.waitingForCompletedSplits());
+ }
+
+ @Test
+ public void testMultiParallelismShouldWaitForCheckpointBeforeCompleted() {
+ SnapshotSplit finishedSplit =
createFinishedSnapshotSplit("db1.table1.1");
+ Map<String, SnapshotSplit> assignedSplits = new HashMap<>();
+ assignedSplits.put(finishedSplit.splitId(), finishedSplit);
+
+ SnapshotSplitAssigner<?> splitAssigner =
+ createRestoredSnapshotSplitAssigner(assignedSplits, new
HashMap<>(), 10);
+
+
splitAssigner.onCompletedSplits(Collections.singletonList(createWatermark(finishedSplit)));
+
+ // multi-parallelism jobs must not complete before a checkpoint pinned
the completion
+ // state, otherwise incremental splits could overtake snapshot records
of the same key
+ Assertions.assertFalse(splitAssigner.isCompleted());
+
+ splitAssigner.snapshotState(21L);
+ Assertions.assertFalse(splitAssigner.isCompleted());
+
+ splitAssigner.notifyCheckpointComplete(20L);
+ Assertions.assertFalse(splitAssigner.isCompleted());
+
+ splitAssigner.notifyCheckpointComplete(21L);
+ Assertions.assertTrue(splitAssigner.isCompleted());
+ }
+
private SnapshotSplitAssigner<?> createRestoredSnapshotSplitAssigner(
Map<String, SnapshotSplit> assignedSplits,
Map<String, SnapshotSplitWatermark> completedOffsets) {
+ return createRestoredSnapshotSplitAssigner(assignedSplits,
completedOffsets, 10);
+ }
+
+ private SnapshotSplitAssigner<?> createRestoredSnapshotSplitAssigner(
+ Map<String, SnapshotSplit> assignedSplits,
+ Map<String, SnapshotSplitWatermark> completedOffsets,
+ int currentParallelism) {
SnapshotPhaseState checkpointState =
new SnapshotPhaseState(
Collections.emptyList(),
@@ -133,7 +192,7 @@ public class SnapshotSplitAssignerTest {
Collections.singleton(TableId.parse("db1.table1")),
checkpointState.getAssignedSplits(),
checkpointState.getSplitCompletedOffsets());
- return new SnapshotSplitAssigner<>(context, 10, checkpointState, null);
+ return new SnapshotSplitAssigner<>(context, currentParallelism,
checkpointState, null);
}
private SnapshotSplit createFinishedSnapshotSplit(String splitId) {
