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)
