spoorthibasu commented on code in PR #4418:
URL: https://github.com/apache/flink-cdc/pull/4418#discussion_r3574706431
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java:
##########
@@ -161,20 +179,75 @@ public void addSplits(Collection<MySqlSplit> splits) {
} else {
// we don't store the split, but will re-create binlog split
later
isBinlogSplitAssigned = false;
+ // re-creating the binlog split: the reader must re-assemble
and re-report
+ // before the snapshot metadata can be released again. Bumping
the generation
+ // invalidates any assembled event still in flight from the
failed attempt.
+ binlogSplitMetaAssembled = false;
+ checkpointIdToReleaseMeta = null;
+ binlogAssignmentGeneration++;
}
}
snapshotSplitAssigner.addSplits(snapshotSplits);
}
@Override
public PendingSplitsState snapshotState(long checkpointId) {
+ // Schedule releasing the snapshot metadata once the binlog split is
assigned and the reader
+ // has assembled it; the release happens in notifyCheckpointComplete.
Skipped when
+ // newly-added-table scan is on, since that flow may still need the
metadata.
+ if (isBinlogSplitAssigned
+ && binlogSplitMetaAssembled
+ && checkpointIdToReleaseMeta == null
+ && !snapshotSplitAssigner.isSnapshotMetaReleased()
+ && !sourceConfig.isScanNewlyAddedTableEnabled()) {
Review Comment:
Agreed, this shouldn't change the default behavior for existing jobs. I've
made the whole optimization opt-in behind a new option
`scan.incremental.snapshot.metadata.release.enabled` (default false). With it
off, nothing is released, so the metadata stays in state and
`scan.newly-added-table.enabled` can be turned on later just like today. Only a
job that explicitly enables the option releases the metadata.
For the cross-restart case you're pointing at, a job that released and then
restarts with `scan.newly-added-table.enabled=true`, I added a fail-fast on
restore. If a released state is restored together with newly-added-table
enabled, the assigner throws a clear error instead of silently building an
inconsistent binlog split. That released state is detected from the existing
checkpoint fields (finished snapshot, emptied assigned-splits map, tables
already processed), so there is no checkpoint or serializer format change and
old checkpoints still restore. Both paths are tested,
`testNoReleaseWhenReleaseMetadataDisabled` for the default-off behavior and
`testFailFastWhenNewlyAddedEnabledOnReleasedRestore` for the cross-restart
guard.
Since this is opt-in, a user hitting the OOM might not realize the flag
exists. If you think it's worth it, I can add a warning log when the
coordinator is still holding a large snapshot state after the binlog phase, so
it surfaces in the JobManager logs where they'd be looking. Happy to add it or
leave it out.
Does this direction and the option name look right to you?
--
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]
