[FLINK-18063][checkpointing] Fix the race condition for aborting current checkpoint in CheckpointBarrierUnaligner #12460

zhijiangW · 2020-06-03T09:39:08Z

What is the purpose of the change

There are three aborting scenarios which might encounter race condition:

CheckpointBarrierUnaligner#processCancellationBarrier
CheckpointBarrierUnaligner#processEndOfPartition
AlternatingCheckpointBarrierHandler#processBarrier

They only consider the pending checkpoint triggered by #processBarrier from task thread to abort it. Actually the checkpoint might also be triggered by #notifyBarrierReceived from netty thread in race condition, so we should also handle properly to abort it.

Brief change log

Fix the process of AlternatingCheckpointBarrierHandler#processBarrier
Fix the process of CheckpointBarrierUnaligner#processEndOfPartition to abort checkpoint properly
Fix the process of CheckpointBarrierUnaligner#processCancellationBarrier to abort checkpoint properly

Verifying this change

Added new unit test CheckpointBarrierUnalignerTest#testProcessCancellationBarrierAfterNotifyBarrierReceived
Added new unit test CheckpointBarrierUnalignerTest#testProcessCancellationBarrierAfterProcessBarrier
Added new unit test CheckpointBarrierUnalignerTest#testProcessCancellationBarrierBeforeProcessAndReceiveBarrier

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/Mesos, 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 · 2020-06-03T09:42:04Z

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 c3f3aca (Wed Jun 03 09:42:03 UTC 2020)

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 · 2020-06-03T10:18:01Z

CI report:

1d838c0 UNKNOWN
edfd5e4 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

...ing-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java

zhijiangW · 2020-06-04T05:38:56Z

Thanks for review @pnowojski ! I appended two hotfix commits and a fixup commit for solving the issue of AlternatingCheckpointBarrierHandler. We can further discuss the other issues then.

zhijiangW · 2020-06-04T12:09:44Z

FYI: Actually I also planned to get ride of ThreadSafeUnaligner#storeNewBuffers variable after refactoring the related processes a bit in this PR. Since the semantic of this state is hard to understand and we can also get ride of numberOfInputChannelsPerGate argument to simplify this class as a result.

It is almost done on my local but brought some unexpected other failures, i am not sure whether have time to further focus on it. Let's see then.

AHeise

Looks pretty good already. I guess I'd mostly like to see a guard in the mail that executes notifyCheckpoint.

...ing-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java

…d method in CheckpointBarrierHandler Simplify the implementations of CheckpointBarrierTracker and CheckpointBarrierUnaligner to reuse the parament default implementation. This closes apache#12460.

…atingCheckpointBarrierHandler#getAlignmentDurationNanos We should take the value from active handler instead of aligned handler, because aligned handler is only used for savepoint and in most cases the unaligned alignment duration should always be 0. This cloese apache#12460.

…d method in CheckpointBarrierHandler Simplify the implementations of CheckpointBarrierTracker and CheckpointBarrierUnaligner to reuse the parent default implementation. This closes apache#12460.

…atingCheckpointBarrierHandler#getAlignmentDurationNanos We should take the value from active handler instead of aligned handler, because aligned handler is only used for savepoint and in most cases the unaligned alignment duration should always be 0. This cloese apache#12460.

…point in CheckpointBarrierUnaligner There are three aborting scenarios which might encounter race condition: 1. CheckpointBarrierUnaligner#processCancellationBarrier 2. CheckpointBarrierUnaligner#processEndOfPartition 3. AlternatingCheckpointBarrierHandler#processBarrier They only consider the pending checkpoint triggered by #processBarrier from task thread to abort it. Actually the checkpoint might also be triggered by #notifyBarrierReceived from netty thread in race condition, so we should also handle properly to abort it. This closes apache#12460.

zhijiangW · 2020-06-05T11:14:47Z

Thanks for the review @AHeise .

I have arranged the commit messages for the final version. Could you double check whether there are any pending concerns now?

AHeise

LGTM with some smaller nits and suggestions.

...ing-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java

...java/src/test/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnalignerTest.java