[FLINK-22367][streaming] Reset syncSavepointId only if it is equal to…

[akalash]

… checkpoint id from event

What is the purpose of the change

PR fixes the bug with resetting syncSavepointId at the wrong time which led to problem with terminateWithSavepointWithoutComplicationsShouldSucceedAndLeadJobToFinished

Brief change log

• *Right now, syncSavepointId can be reset only if it is equal to the checkpoint id from event *

Verifying this change

This change added tests and can be verified as follows:

• *Added new test which checks that the abort before completing checkpoint doesn't lead to the problem with finishing task. *

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]

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 84e6a95 (Mon Aug 30 15:18:28 UTC 2021)

Warnings:

• No documentation files were touched! Remember to keep the Flink docs up to date!

_{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]

CI report:

• 510c0e1 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run travis re-run the last Travis build
• @flinkbot run azure re-run the last Azure build

[dawidwys]

How can this happen? You cannot complete a checkpoint that has not been triggered yet. The order of notifyCheckpointCompleteAsync and triggerCheckpointOnBarrier is wrong here.

[akalash]

I took this order from the test terminateWithSavepointWithoutComplicationsShouldSucceedAndLeadJobToFinished. Usually, it indeed happens in normal order, but sometimes I see that the abort happens before the complete.

[dawidwys]

I meant the order of notifyCheckpointCompleteAsync and triggerCheckpointOnBarrier.

The barrier must arrive first before we can say the checkpoint completed. It's not possible to receive a notifyCheckpointCompleteAsync before we've seen a barrier for the checkpoint.

In your testcase you're first sending the notifyCheckpointCompleteAsync(2) and after that you're sending a barrier: triggerCheckpointOnBarrier(...(2)...)

For the complete vs abort, you're right, I can see that happen.

[akalash]

notifyCheckpointCompleteAsync and notifyCheckpointAbortAsync actually do nothing but add the task to mailbox. When triggerCheckpointOnBarrier is called it first of all does its own job and only then it directly execute all waiting mails until the checkpoint is complete. So in fact the triggerCheckpointOnBarrier is always first.

[dawidwys]

Still this is based on the internal knowledge of the component. It is still true that those methods can not be called in that order in the production code. I find the current order both confusing and prone to changes in the implementation. I find it also distracting from the real scenario that's being tested here.

I might be wrong, but I truly believe I am not the only one that would be confused by this order.

[akalash]

Do I understand correctly, that you are talking about order in the test? Do you mean that we should explicitly call triggerCheckpointOnBarrier first? If so please, check the my last commit I a little rewrote the test

[dawidwys]

Yes, I am talking about the order in the test.

Why wasn't it necessary in the previous version to call the triggerCheckpointOnBarrier from within the executor?

Do we need to call the notifyCheckpointCompleteAsync and notifyCheckpointAbortAsync` from the executor? I think those methods are called via RPC in the production code and are not executed from the executor thread...

[dawidwys]

Ok, I checked it myself. If we execute it from the test thread it gets stuck in the synchronous savepoint loop.

I think executing triggerCheckpointOnBarrier from the MailboxExecutor is fine here. I would not do it for the *Async methods.

[akalash]

ok, I have removed the mailbox Executor for async calls. Initially, I added it that only for keeping the one style but it indeed didn't have any real function.

[dawidwys]

I think your investigation is valid. Have you checked if it is a problem for earlier versions that we support? E.g. 1.12.x?

AFAICT it should not be a problem for master. It should've been fixed there. One change I'd do is to reset it for any newer checkpoint than that of the savepoint. Just to stay on the safe side. WDYT?

It should not happen, as there should be no checkpoints triggered from the checkpoint coordinator after a sync savepoint (it should either stop the job or trigger a failover).

[akalash]

It is also valid for at least release-1.12 - #17104

[dawidwys]

LGTM, let's wait for Azure