[FLINK-26516][streaming] Recover GlobalCommittables with Sink V1 GlobalCommittable serializer #18805

fapaul · 2022-02-16T14:27:21Z

What is the purpose of the change

With SinkV2 the committer and global committer work very similar and
they only write committables into state. SinkV1's GlobalCommitter on the
other hand used to write GlobalCommittables into state so this commits
adds an migration path.

Brief change log

c725569 enabled the GlobalCommitter IT cases unfortunately we missed to enable them after FLINK-25726 was merged
8549f8f Pass the global committer and serializer to the global committer operator to restore Sink V1 state

Verifying this change

Reenabled tests pass
Added an upgrade test to restore a savepoint from Sink V1.

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)

flinkbot · 2022-02-16T14:32:34Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 8ab3d80 (Wed Feb 16 14:32:33 UTC 2022)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!
Invalid pull request title: No valid Jira ID provided

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2022-02-16T14:34:06Z

CI report:

617d6de Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

fapaul · 2022-02-28T15:45:12Z

@gaoyunhaii I think this should fix the upgrade path from Sink V1 to Sink V2.

fapaul · 2022-02-28T16:13:32Z

@flinkbot run azure

fapaul · 2022-02-28T16:39:03Z

@flinkbot run azure

gaoyunhaii · 2022-03-01T06:53:24Z

Thanks @fapaul for the PR! I'll have a look~

...java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java

gaoyunhaii · 2022-03-03T17:07:35Z

...reaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SinkV1Adapter.java

@@ -447,5 +451,13 @@ public void commit(Collection<CommitRequest<CommT>> committables)
                committables.forEach(CommitRequest::retryLater);


Here seems should be failures.forEach?

This is not easily possible because only committable are retriable at this point. I would need to add a special logic when this happens. I rely here on the fact that I assume all commit calls are idempotent because it might happen that global committables are committed again.

Thanks for the clarification! I did not notice the difference of types. Might we add some comments for this behavior?

Good idea, I add a comment.

gaoyunhaii · 2022-03-03T17:13:29Z

Very thanks @fapaul for the PR!

One main issue from my side is that it seems the behavior of the current GlobalCommitterOperator is still not compatible with the legacy one? Currently once the legacy state is restored, in the following execution the behavior of combine, filterRecoveredCommittables and endOfInput is still not called~?

fapaul · 2022-03-04T08:36:53Z

Very thanks @fapaul for the PR!

One main issue from my side is that it seems the behavior of the current GlobalCommitterOperator is still not compatible with the legacy one? Currently once the legacy state is restored, in the following execution the behavior of combine, filterRecoveredCommittables and endOfInput is still not called~?

I would have though combine and filterRecoveredCommittables are now equal or very closely to what was done before.

To be honest I never really understood the idea behind exposing endOfInput of the GlobalCommitter. I also have never seen a sink really using it. Can you maybe provide more context here? I would probably make a big warning in the release notes that this endOfInput is not used anymore.

gaoyunhaii · 2022-03-04T10:15:08Z

Hi Fabian~

I would have though combine and filterRecoveredCommittables are now equal or very closely to what was done before.

It seems currently filterRecoveredCommittables and combine are used only on the legacy state~? For example, suppose we recovered from sink v1, then on restoring we would call filterRecoveredCommittables, then if the job continue running and take another several checkpoints, then after failover all committables are stored in the CommittableCollector and on restoring filterRecoveredCommittables seems would not be called~?

For endOfInput it is initially designed to write metas, like write partition to the hive metastore or write _SUCCESS file under the directory. I'm also agree with we could change it if no users are using it since it is marked as Experimental before.

fapaul · 2022-03-04T14:49:33Z

It seems currently filterRecoveredCommittables and combine are used only on the legacy state~? For example, suppose we recovered from sink v1, then on restoring we would call filterRecoveredCommittables, then if the job continue running and take another several checkpoints, then after failover all committables are stored in the CommittableCollector and on restoring filterRecoveredCommittables seems would not be called~?

I thought about this for a longer time and think the behaviour of combine should be fine since it is always used when calling commit. The semantic of filterRecoveredCommittables is definitely currently unclear although I would also question the purpose of this method.
In general, I think all commit calls be for the committer of global committer need to be idempotent and handling already committed committables needs to happen in the commit method either way. I do not see an immediate benefit of calling filterRecoveredCommittables.

I am not really what to do about that. Do you have an idea?

gaoyunhaii · 2022-03-05T08:29:31Z

Hi Fabian~ thanks for the clarification!

For the filterRecoveredCommittables, it should be initiated by some data lake systems like iceberg, if there are global committables from more than one checkpoint not committed before failover, then it should be able to merge them into one global committable on restoring, thus the restoring process could be accelerated [1].

For the combine, sorry I overlooked the issue and the current implementation, currently it is indeed called before committed. The issue is said that the committables should be combined on snapshotState, and the resulted global committables should be persisted, thus if there are failovers and re-commit, the combine does need to be executed again.

If we want to call filterRecoveredCommittables for each snapshot, we might need to add a parameter isRestored to the CommittableManager#commit. If we want to also keep the same behavior for the combine method, we might need to do some more modification to the CommittableManager or not reuse the current implementation.

However, currently the sink implementation in the iceberg, hudi and flink-table-store are indeed using the customized operators / legacy sink API instead of the sink API v1. In addition, the two issues should mainly affect he performance instead of the correctness. At last, users are still be able to add their own global committer implementation with topology API. Thus I think it is still useful to solve the two issues, but if we do not have capacity it would be also ok we postpone solving them.

If so, perhaps we could create a new jira issue for state compatibility with the sink v1 and attach this PR to that issue, and dowgrade the FLINK-26173 and leaves it open? Since we have not fully fix the issues listed in the FLINK-26173~

[1] https://lists.apache.org/thread/3tqgnhclbr8ptb69b2ro74535dtbd608

…alCommittable serializer With SinkV2 the committer and global committer work very similar and they only write committables into state. SinkV1's GlobalCommitter on the other hand used to write GlobalCommittables into state so this commits adds an migration path.

fapaul · 2022-03-07T13:23:26Z

If so, perhaps we could create a new jira issue for state compatibility with the sink v1 and attach this PR to that issue, and dowgrade the FLINK-26173 and leaves it open? Since we have not fully fix the issues listed in the FLINK-26173~

Thanks for the suggestion. I created a new ticket https://issues.apache.org/jira/browse/FLINK-26516 to fix the state incompatibility and lowered the priority of the original ticket.

gaoyunhaii · 2022-03-08T09:38:19Z

I added a fix for the failed azure test: previously if there are no committables the combine and commit would be skipped, otherwise there might create an empty global committable. The test failed due to this reason.

gaoyunhaii

Thanks @fapaul for the PR! LGTM

…alCommittable serializer With SinkV2 the committer and global committer work very similar and they only write committables into state. SinkV1's GlobalCommitter on the other hand used to write GlobalCommittables into state so this commits adds an migration path. This closes apache#18805.

fapaul changed the title ~~Flink 26173~~ [FLINK-26173][streaming] Call GlobalCommitter#filterRecoveredCommittable during commit to ensure sane behaviour after migration Feb 17, 2022

rmetzger added the component=API/DataStream label Feb 17, 2022

fapaul force-pushed the FLINK-26173 branch from 8ab3d80 to 8549f8f Compare February 28, 2022 15:35

fapaul changed the title ~~[FLINK-26173][streaming] Call GlobalCommitter#filterRecoveredCommittable during commit to ensure sane behaviour after migration~~ [FLINK-26173][streaming] Recover GlobalCommittables with Sink V1 GlobalCommittable serializer Feb 28, 2022

fapaul force-pushed the FLINK-26173 branch from 8549f8f to f8417e0 Compare February 28, 2022 15:56

fapaul force-pushed the FLINK-26173 branch 4 times, most recently from ab6319c to 438259a Compare March 3, 2022 09:02

gaoyunhaii reviewed Mar 3, 2022

View reviewed changes

[hotfix][streaming] Enabling GlobalCommitter SinkITCases

263ae3d

fapaul force-pushed the FLINK-26173 branch from 438259a to 6395fcb Compare March 7, 2022 09:44

fapaul changed the title ~~[FLINK-26173][streaming] Recover GlobalCommittables with Sink V1 GlobalCommittable serializer~~ [FLINK-26516][streaming] Recover GlobalCommittables with Sink V1 GlobalCommittable serializer Mar 7, 2022

fapaul force-pushed the FLINK-26173 branch from 6395fcb to e2e075c Compare March 7, 2022 09:45

rmetzger added component=Connectors/Common and removed component=API/DataStream labels Mar 7, 2022

fapaul force-pushed the FLINK-26173 branch from e2e075c to e0cc9d1 Compare March 7, 2022 13:21

[fixup] not call combine if there are no committables

617d6de

gaoyunhaii approved these changes Mar 8, 2022

View reviewed changes

gaoyunhaii closed this in 955e5ff Mar 8, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-26516][streaming] Recover GlobalCommittables with Sink V1 GlobalCommittable serializer #18805

[FLINK-26516][streaming] Recover GlobalCommittables with Sink V1 GlobalCommittable serializer #18805

fapaul commented Feb 16, 2022 •

edited

Loading

flinkbot commented Feb 16, 2022

flinkbot commented Feb 16, 2022 •

edited

Loading

fapaul commented Feb 28, 2022

fapaul commented Feb 28, 2022

fapaul commented Feb 28, 2022

gaoyunhaii commented Mar 1, 2022

gaoyunhaii Mar 3, 2022

fapaul Mar 4, 2022 •

edited

Loading

gaoyunhaii Mar 5, 2022 •

edited

Loading

fapaul Mar 7, 2022

gaoyunhaii commented Mar 3, 2022

fapaul commented Mar 4, 2022

gaoyunhaii commented Mar 4, 2022

fapaul commented Mar 4, 2022

gaoyunhaii commented Mar 5, 2022 •

edited

Loading

fapaul commented Mar 7, 2022

gaoyunhaii commented Mar 8, 2022

gaoyunhaii left a comment

		@@ -447,5 +451,13 @@ public void commit(Collection<CommitRequest<CommT>> committables)
		committables.forEach(CommitRequest::retryLater);

[FLINK-26516][streaming] Recover GlobalCommittables with Sink V1 GlobalCommittable serializer #18805

[FLINK-26516][streaming] Recover GlobalCommittables with Sink V1 GlobalCommittable serializer #18805

Conversation

fapaul commented Feb 16, 2022 • edited Loading

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

flinkbot commented Feb 16, 2022

Automated Checks

Review Progress

flinkbot commented Feb 16, 2022 • edited Loading

CI report:

fapaul commented Feb 28, 2022

fapaul commented Feb 28, 2022

fapaul commented Feb 28, 2022

gaoyunhaii commented Mar 1, 2022

gaoyunhaii Mar 3, 2022

Choose a reason for hiding this comment

fapaul Mar 4, 2022 • edited Loading

Choose a reason for hiding this comment

gaoyunhaii Mar 5, 2022 • edited Loading

Choose a reason for hiding this comment

fapaul Mar 7, 2022

Choose a reason for hiding this comment

gaoyunhaii commented Mar 3, 2022

fapaul commented Mar 4, 2022

gaoyunhaii commented Mar 4, 2022

fapaul commented Mar 4, 2022

gaoyunhaii commented Mar 5, 2022 • edited Loading

fapaul commented Mar 7, 2022

gaoyunhaii commented Mar 8, 2022

gaoyunhaii left a comment

Choose a reason for hiding this comment

fapaul commented Feb 16, 2022 •

edited

Loading

flinkbot commented Feb 16, 2022 •

edited

Loading

fapaul Mar 4, 2022 •

edited

Loading

gaoyunhaii Mar 5, 2022 •

edited

Loading

gaoyunhaii commented Mar 5, 2022 •

edited

Loading