dev-donghwan opened a new pull request, #4489: URL: https://github.com/apache/flink-cdc/pull/4489
Jira: https://issues.apache.org/jira/browse/FLINK-40216 ### Problem On restore, `SnapshotSplitAssigner` re-sorts `assignedSplits` lexicographically 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, ...`). Since `IncrementalSourceEnumerator` partitions `getFinishedSplitInfos()` 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 `FinishedSnapshotSplitInfo` are received twice, others never. `StreamSplit#isCompletedSplit` only compares sizes, so the corrupted split is considered complete, and `IncrementalSourceStreamFetcher#shouldEmit` silently 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 `SnapshotSplitAssigner` from `flink-cdc-base` are prone to the same issue. This PR ports that fix to `flink-cdc-base`, covering all connectors on the base incremental framework (Postgres, MongoDB, Oracle, SqlServer, Db2). ### Changes * `PendingSplitsStateSerializer` / `SnapshotPendingSplitsState`: deserialize `assignedSplits` (and related maps) into a `LinkedHashMap` — 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; build `finishedSnapshotSplitInfos` in 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`), mirroring `MySqlBinlogSplit`. ### 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 and `chunk-meta.group.size=2`, the reader ends up with `t:4`/`t:5` duplicated and `t:10`/`t:11` never delivered); with this fix both pass. This is the base-side counterpart of `MySqlSnapshotSplitAssignerTest#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-base` is possible follow-up work, out of scope here. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
