[jira] [Created] (FLINK-40216) [cdc-base] Meta groups of finished snapshot splits become inconsistent after restore due to lexicographic re-sorting of assigned splits

["kim (Jira)" <ji...@apache.org>]

kim created FLINK-40216:
---------------------------

             Summary: [cdc-base] Meta groups of finished snapshot splits become inconsistent after restore due to lexicographic re-sorting of assigned splits
                 Key: FLINK-40216
                 URL: https://issues.apache.org/jira/browse/FLINK-40216
             Project: Flink
          Issue Type: Bug
          Components: Flink CDC
    Affects Versions: cdc-3.1.0, cdc-3.6.0
            Reporter: kim


h3. Problem

In {{flink-cdc-base}}, the order of {{assignedSplits}} is not preserved across a checkpoint/savepoint restore, so the meta groups of finished snapshot splits are partitioned differently before and after a restart.

During a normal run, splits are appended to {{assignedSplits}} (a {{LinkedHashMap}}) in *assignment order* ({{SnapshotSplitAssigner#getNext}}). On restore, however:
# {{PendingSplitsStateSerializer#readAssignedSnapshotSplits}} deserializes the checkpointed splits into a plain {{HashMap}}, losing the checkpointed order.
# The restore constructor of {{SnapshotSplitAssigner}} then re-sorts the entries *lexicographically* by split id (introduced in FLINK-34634 as a workaround for the lost order):
{code:java}
this.assignedSplits =
        assignedSplits.entrySet().stream()
                .sorted(Map.Entry.comparingByKey())
                ...
{code}

Split ids are generated as {{tableId + ":" + chunkId}} ({{SnapshotSplit#generateSplitId}}), so with 10+ chunks the lexicographic order ({{t:0, t:1, t:10, t:11, t:2, ...}}) differs from the assignment order ({{t:0, t:1, t:2, ..., t:11}}). The same happens across tables when the alphabetical order of table names differs from the capture order.

h3. Impact

{{IncrementalSourceEnumerator#sendStreamMetaRequestEvent}} partitions {{splitAssigner.getFinishedSplitInfos()}} by {{chunk-meta.group.size}} ({{Lists.partition}}) and serves the group requested by the reader. A reader whose stream split had synchronized only part of the meta groups before the restart resumes requesting from its next group id, but the enumerator now serves groups computed from a differently-ordered list.

We reproduced this with a unit test against current master (single table, 12 chunks, {{chunk-meta.group.size = 2}}, reader synchronized meta groups 0..2 before the restart, group ids computed with {{IncrementalSourceReader#getNextMetaGroupId}}): after the restore the reader ends up with

{code}
[t:0, t:1, t:2, t:3, t:4, t:5, t:4, t:5, t:6, t:7, t:8, t:9]
{code}

i.e. {{t:4}}/{{t:5}} are received twice and {{t:10}}/{{t:11}} are never delivered, while the stream split is considered complete because {{StreamSplit#isCompletedSplit}} only compares sizes. Subsequently {{IncrementalSourceStreamFetcher#shouldEmit}} finds no matching {{FinishedSnapshotSplitInfo}} for change events in the key ranges of the lost splits and silently drops them — permanent data loss in the stream phase. Depending on where the restart happens, the mismatch can also surface as the "invalid request meta group id" error.

The MySQL connector carried the identical re-sort and was fixed by FLINK-38218 (cdc-3.6.0), which removed it and made the whole pipeline preserve the assignment order. As its author noted there: "Note that other connectors that use SnapshotSplitAssigner from the flink-cdc-base package may be prone to this issue as well." {{flink-cdc-base}} still carries the identical code, affecting all connectors built on the base incremental framework: Postgres, MongoDB, Oracle, SqlServer, Db2 (and their SQL/pipeline wrappers). {{MySqlSnapshotSplitAssignerTest#testFinishedSnapshotSplitInfosAreInOrderOfAssignment}} guards this property on the MySQL side; {{flink-cdc-base}} has no equivalent test.

h3. Proposed fix

Port FLINK-38218 to {{flink-cdc-base}}:
* deserialize {{assignedSplits}} into a {{LinkedHashMap}} in {{PendingSplitsStateSerializer}} (the serialized form already preserves iteration order) and declare the order contract on the field, instead of re-sorting in the restore constructor of {{SnapshotSplitAssigner}};
* remove the lexicographic sort in {{HybridSplitAssigner#createStreamSplit}};
* replace the order-dependent deduplication in {{IncrementalSourceReader}} ({{getExistedSplitsOfLastGroup}}) with the order-agnostic prefix-discard approach from FLINK-38218;
* guard {{StreamSplit}} against duplicated split infos, mirroring {{MySqlBinlogSplit#ensureNoDuplicates}}.

This fix is independent of the metadata-transmission protocol redesign discussed in FLINK-38270 and can be applied separately.




--
This message was sent by Atlassian Jira
(v8.20.10#820010)