This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12162-85ce19515b374cd55d3b213873c848b40c0d964d in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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) {
