Skip to content

[FLINK-38765] Fix persisted metadata handling in sink #27301

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Merged
dawidwys merged 2 commits into from
Dec 15, 2025
Merged

[FLINK-38765] Fix persisted metadata handling in sink #27301

dawidwys merged 2 commits into from
Dec 15, 2025

Conversation

@dawidwys
Copy link
Contributor

@dawidwys dawidwys commented Dec 2, 2025

What is the purpose of the change

Fixes a bug where an incorrect type is used in sink for

  • shuffling
  • sink upsert materializer
  • constraint validation

Verifying this change

Added a test in SinkTestPrograms

Does this pull request potentially affect one of the following parts:

  • Dependencies (does it add or upgrade a dependency): (yes / no)
  • The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
  • The serializers: (yes / no / don't know)
  • The runtime per-record code paths (performance sensitive): (yes / no / don't know)
  • Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
  • The S3 file system connector: (yes / no / don't know)

Documentation

  • Does this pull request introduce a new feature? (yes / no)
  • If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

@dawidwys dawidwys requested a review from twalthr December 2, 2025 14:32
@flinkbot
Copy link
Collaborator

flinkbot commented Dec 2, 2025 •
edited
Loading

CI report:

Bot commands The @flinkbot bot supports the following commands:
  • @flinkbot run azure re-run the last Azure build

twalthr
twalthr reviewed Dec 2, 2025
View reviewed changes
twalthr
twalthr approved these changes Dec 3, 2025
View reviewed changes
Copy link
Contributor

@twalthr twalthr left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @dawidwys. +1 for this. Please update the release notes and let people know about the past inconsistency and that there is a potential restoring issue but the state was corrupted anyways.

twalthr
twalthr reviewed Dec 11, 2025
View reviewed changes
Copy link
Contributor

@twalthr twalthr left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thank you for digging deeper and fixing this bug in a backwards compatible way. I left some last comments.

...lanner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecSink.java Outdated
// We introduced a new version, because statements that were never rolling back to a value from
// state could run succesfully. We allow those jobs to be upgraded. Without a new versions such jobs
// would fail on restore, because the state serializer would differ
@ExecNodeMetadata(
Copy link
Contributor

@twalthr twalthr Dec 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Batch version can stay at v1. There is no upgrade story that would interrupt a running batch job. I did it similarly for scan_v2, where I just updated v1 to the new behavior.

Copy link
Contributor Author

@dawidwys dawidwys Dec 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point. Overlooked that in your commit

...rc/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSinkRestoreTest.java Outdated
@Override
protected Stream<String> getSavepointPaths(
TableTestProgram program, ExecNodeMetadata metadata) {
// disable the writable metadata test for sink node with version 1. it fails after the
Copy link
Contributor

@twalthr twalthr Dec 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we generalize this into a method in RestoreTestBase that TableSinkRestoreTest can implement for ignoring certain programs with versions?

Copy link
Contributor Author

@dawidwys dawidwys Dec 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure if that's what you had in mind, but I refactored it a bit.

twalthr
twalthr approved these changes Dec 15, 2025
View reviewed changes
Copy link
Contributor

@twalthr twalthr left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM, thanks for leveraging the version infrastructure and ensuring backwards compatibility.

@dawidwys
Copy link
Contributor Author

dawidwys commented Dec 15, 2025

Thank you. I'll just squash the commits to run the CI on a final version that I can merge 👍

@dawidwys dawidwys merged commit c028790 into apache:master Dec 15, 2025
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@twalthr twalthr twalthr approved these changes

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@dawidwys @flinkbot @twalthr