[FLINK-40697][cdc-base] Release finished snapshot split metadata after entering the stream phase - #4550
Open
spoorthibasu wants to merge 1 commit into
Open
[FLINK-40697][cdc-base] Release finished snapshot split metadata after entering the stream phase#4550spoorthibasu wants to merge 1 commit into
spoorthibasu wants to merge 1 commit into
Conversation
…r entering the stream phase
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
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.
What is the purpose of this pull request?
The incremental source framework (flink-cdc-base) keeps the full snapshot-split metadata (assigned splits, finished offsets, table schemas) in the split enumerator for the whole lifetime of the job and re-serializes it on every checkpoint. On a large multi-table job this keeps JobManager memory high and inflates checkpoint size.
FLINK-39775 fixed this for the MySQL connector. This PR generalizes the same mechanism into flink-cdc-base so the JDBC incremental dialects (Postgres, SqlServer, Oracle, Db2) get it, rather than each connector solving it separately.
The release is opt-in behind a new option
scan.incremental.snapshot.metadata.release.enabled(defaultfalse), so existing jobs are unaffected. Once the stream split is assigned and the reader reports its metadata assembled, the enumerator schedules the release at the current checkpoint and clears it innotifyCheckpointComplete, so the assignment is always covered by a completed checkpoint. An assignment-generation guard rejects a stale assembled report from a reader that failed over. The checkpoint serializer is bumped to version 9 only for jobs that enable the option, so default-off jobs keep writing version 8 and stay restorable by an older build.Brief change log
scan.incremental.snapshot.metadata.release.enabled(defaultfalse) to the base source options and configsnapshotState, execute innotifyCheckpointComplete), with the assignment-generation guardPendingSplitsStateSerializerwrites version 9 only when the option is enabled, else version 8scan.newly-added-table.enabled, and on restoring an already-released state with the option offVerifying this change
This change added tests and can be verified as follows:
Documentation
The Chinese doc entries mirror the English and are a best-effort translation. Please correct the wording where needed.
Was generative AI tooling used to co-author this PR?