Skip to content

[FLINK-40697][cdc-base] Release finished snapshot split metadata after entering the stream phase - #4550

Open
spoorthibasu wants to merge 1 commit into
apache:masterfrom
spoorthibasu:phase1/cdc-base-metadata-release
Open

spoorthibasu wants to merge 1 commit into
apache:masterfrom
spoorthibasu:phase1/cdc-base-metadata-release

Conversation

@spoorthibasu

Copy link
Copy Markdown
Contributor

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 (default false), 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 in notifyCheckpointComplete, 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

  • Add scan.incremental.snapshot.metadata.release.enabled (default false) to the base source options and config
  • Add the release primitive to the snapshot assigner and thread it through the split and hybrid assigners
  • Add the reader-to-enumerator assembled event and the enumerator release orchestration (arm in snapshotState, execute in notifyCheckpointComplete), with the assignment-generation guard
  • Persist the released flag; PendingSplitsStateSerializer writes version 9 only when the option is enabled, else version 8
  • Fail fast on release together with scan.newly-added-table.enabled, and on restoring an already-released state with the option off
  • Expose the option through the DataStream builders and Flink SQL for Postgres, SqlServer, Oracle, and Db2
  • Document the option (EN and ZH) for the four connectors

Verifying this change

This change added tests and can be verified as follows:

  • Unit tests in flink-cdc-base for the release protocol, the generation guard, the conditional serializer versioning (version 8 vs 9 by the option), and the fail-fast paths
  • Integration tests on real Postgres and SqlServer databases covering the group-fetch release path plus JobManager and TaskManager failover in the stream phase, verifying no snapshot or stream data is lost through the release and restore

Documentation

  • Does this pull request introduce a new feature? yes
  • If yes, how is the feature documented? docs
    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?
  • Yes (please specify the tool below)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant