[FLINK-40216][cdc-base] Preserve assignment order of splits across restore to keep meta groups consistent#4489
Open
dev-donghwan wants to merge 1 commit into
Conversation
…store to keep meta groups consistent The restore constructor of SnapshotSplitAssigner re-sorted the assigned splits lexicographically by split id, while at runtime splits are kept in assignment order. Since the enumerator partitions getFinishedSplitInfos() into meta groups, the groups served before and after a restore differed, so a reader resuming an incomplete meta synchronization received duplicated split infos while never receiving others, leading to silently dropped change events in the stream phase. Port the FLINK-38218 fix from the MySQL connector to flink-cdc-base: - deserialize assignedSplits into a LinkedHashMap and keep the checkpointed assignment order instead of re-sorting on restore - remove the lexicographic sort in HybridSplitAssigner#createStreamSplit - replace the order-dependent deduplication in IncrementalSourceReader with the order-agnostic prefix-discard approach - reject duplicated split infos in StreamSplit
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Jira: https://issues.apache.org/jira/browse/FLINK-40216
Problem
On restore,
SnapshotSplitAssignerre-sortsassignedSplitslexicographically by split id (t:0, t:1, t:10, t:11, t:2, ...), while during a normal run the map holds them in assignment order (t:0, t:1, t:2, ...). SinceIncrementalSourceEnumeratorpartitionsgetFinishedSplitInfos()into meta groups positionally (Lists.partition), the same group id maps to different splits before and after a restore.A reader that had synchronized only part of the meta groups before the restart resumes requesting from its next group id against the re-partitioned list: some
FinishedSnapshotSplitInfoare received twice, others never.StreamSplit#isCompletedSplitonly compares sizes, so the corrupted split is considered complete, andIncrementalSourceStreamFetcher#shouldEmitsilently drops change events in the key ranges of the missing chunks — permanent data loss. Depending on timing it can instead surface as the"invalid request meta group id"error.The MySQL connector had the identical defect and fixed it in FLINK-38218; as noted there, other connectors using
SnapshotSplitAssignerfromflink-cdc-baseare prone to the same issue. This PR ports that fix toflink-cdc-base, covering all connectors on the base incremental framework (Postgres, MongoDB, Oracle, SqlServer, Db2).Changes
PendingSplitsStateSerializer/SnapshotPendingSplitsState: deserializeassignedSplits(and related maps) into aLinkedHashMap— the serialized form already preserves iteration order — and declare the order contract on the fields.SnapshotSplitAssigner: remove the lexicographic re-sort in the restore constructor.HybridSplitAssigner#createStreamSplit: remove the lexicographic sort; buildfinishedSnapshotSplitInfosin assignment order.IncrementalSourceReader: replace the order-dependent last-group deduplication (getExistedSplitsOfLastGroup) with the order-agnostic prefix-discard approach from FLINK-38218.StreamSplit: guard against duplicated split infos (ensureNoDuplicates), mirroringMySqlBinlogSplit.Tests
MetaGroupOrderingTest(new): restores an assigner from a checkpoint taken mid meta-group synchronization and asserts (a)getFinishedSplitInfos()preserves assignment order after restore, and (b) a resuming reader receives every split exactly once. On current master both fail deterministically (with 12 chunks andchunk-meta.group.size=2, the reader ends up witht:4/t:5duplicated andt:10/t:11never delivered); with this fix both pass. This is the base-side counterpart ofMySqlSnapshotSplitAssignerTest#testFinishedSnapshotSplitInfosAreInOrderOfAssignment.PendingSplitsStateSerializerTest: serde round-trip preserves assignment order.StreamSplitTest: duplicate split infos are rejected.Note
This is independent of the metadata-transmission protocol redesign discussed in FLINK-38270. Unifying the MySQL connector's own copy of this logic with
flink-cdc-baseis possible follow-up work, out of scope here.