[FLINK-22638] Keep channels blocked on alignment timeout #15897

dawidwys · 2021-05-11T16:03:27Z

What is the purpose of the change

Improve the end to end checkpointing time in case alignment timeout occurs.

Brief change log

See the commits

Verifying this change

Updated tests.

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 · 2021-05-11T16:07:05Z

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 bab4983 (Sat Aug 28 12:11:27 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 · 2021-05-11T16:14:30Z

CI report:

bab4983 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

This commit keeps channels blocked in case an alignment timeout occurs. That way we prioritize the channels that we have not received the barrier yet. This solution is based on the assumption that all upstream operators are working with aligned checkpoints and we do not mind delaying the subsequent checkpoints on the blocked channels.

dawidwys · 2021-05-17T06:53:36Z

@pnowojski What do you think about those changes?

dawidwys · 2021-05-17T13:29:19Z

@curcur Do you mind taking a look?

.../flink/streaming/runtime/io/checkpointing/AbstractAlternatingAlignedBarrierHandlerState.java

.../apache/flink/streaming/runtime/io/checkpointing/AlternatingCollectingBarriersUnaligned.java

...che/flink/streaming/runtime/io/checkpointing/AlternatingWaitingForFirstBarrierUnaligned.java

curcur · 2021-05-18T16:08:37Z

...che/flink/streaming/runtime/io/checkpointing/AlternatingWaitingForFirstBarrierUnaligned.java

-            inputs[channelInfo.getGateIdx()].resumeConsumption(channelInfo);
+            channelState.blockChannel(channelInfo);


I am a bit struggling thinking of cases where out-of-order barriers come? Since transition happens when AlternatingWaitingForFirstBarrier#alignmentTimeout, and converting to priority events happen before transition. On the other hand, all barriers seen has already have the announcement removed.

...che/flink/streaming/runtime/io/checkpointing/AlternatingWaitingForFirstBarrierUnaligned.java

curcur · 2021-05-19T04:12:24Z

...est/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingCheckpointsTest.java

@@ -507,13 +477,20 @@ public void testAllChannelsUnblockedAfterAlignmentTimeout() throws Exception {

        // we set timer on announcement and test channels do not produce announcements by themselves
        send(EventSerializer.toBuffer(new EventAnnouncement(checkpointBarrier, 0), true), 0, gate);
-        send(checkpointBarrierBuffer, 0, gate);
        // emulate blocking channels on aligned barriers
        ((TestInputChannel) gate.getChannel(0)).setBlocked(true);


Is this necessary? The channel is blocked after receiving the first aliged barrier in

AbstractAlternatingAlignedBarrierHandlerState#barrierReceived

Yes, it is. I agree it is a bit confusing. The network stack actually blocks the regular channels. Therefore the calls in BarrierHandlerState are necessary only for bookkeeping to unblock them in the end. This emulates the network stack behaviour.

See

public abstract class IndexedInputGate extends InputGate implements CheckpointableInput { .... @Override public void blockConsumption(InputChannelInfo channelInfo) { // Unused. Network stack is blocking consumption automatically by revoking credits. } ... }

curcur · 2021-05-19T05:36:40Z

@dawidwys Hey Dawid, thanks very much for the PR. I did one pass and I do not see any obvious problem with the change of "blocking after receiving barrier". There are some inline comments above + a few more comments in general (which may not directly related the change made here)

I think it worth some explanation (one-sentense doc) for these four states, especially for the flag alternating, it would be even better to describe the transition relations between these four (But I do not think this has to be included in this PR though).

AlternatingWaitingForFirstBarrier
AlternatingCollectingBarriers
AlternatingWaitingForFirstBarrierUnaligned
AlternatingCollectingBarriersUnaligned

Do we need a test specifically to test the channels are blocked (for this change)?

dawidwys · 2021-05-19T13:42:59Z

Ad. 2 I think correctness-wise it does not make a difference when the channels are unblocked. What we care about is that the channels are unblocked once a checkpoint finishes (either completed or aborted). I would not enforce that contract in unit tests. It is kind of covered by the performance tests. If we say at some point that actually it's better to unblock the channels as soon as possible we can change it easily. WDYT?

curcur · 2021-05-21T08:26:06Z

@dawidwys , thanks for addressing the comments, I've approved it.

As to the question about out-of-order cases. My concern is if we can not reason about it, it may potentially hide bugs behind it. But I understand this was there before this PR, so maybe we can discuss it after this PR.

This commit keeps channels blocked in case an alignment timeout occurs. That way we prioritize the channels that we have not received the barrier yet. This solution is based on the assumption that all upstream operators are working with aligned checkpoints and we do not mind delaying the subsequent checkpoints on the blocked channels. This closes #15897

dawidwys requested a review from AHeise May 11, 2021 16:03

dawidwys changed the title ~~Flink 22638~~ [FLINK-22638] Keep channels blocked on alignment timeout May 11, 2021

rmetzger added the review=description? label May 11, 2021

rmetzger added the component=Runtime/Checkpointing label May 11, 2021

dawidwys requested a review from pnowojski May 12, 2021 08:26

dawidwys force-pushed the FLINK-22638 branch from 659023b to 4d404aa Compare May 17, 2021 06:49

curcur reviewed May 19, 2021

View reviewed changes

Address comments

bab4983

curcur approved these changes May 21, 2021

View reviewed changes

dawidwys closed this in f041516 May 21, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-22638] Keep channels blocked on alignment timeout #15897

[FLINK-22638] Keep channels blocked on alignment timeout #15897

dawidwys commented May 11, 2021

flinkbot commented May 11, 2021 •

edited

flinkbot commented May 11, 2021 •

edited

dawidwys commented May 17, 2021

dawidwys commented May 17, 2021

curcur May 18, 2021 •

edited

curcur May 19, 2021

dawidwys May 19, 2021

curcur commented May 19, 2021

dawidwys commented May 19, 2021

curcur commented May 21, 2021

		inputs[channelInfo.getGateIdx()].resumeConsumption(channelInfo);
		channelState.blockChannel(channelInfo);

[FLINK-22638] Keep channels blocked on alignment timeout #15897

[FLINK-22638] Keep channels blocked on alignment timeout #15897

Conversation

dawidwys commented May 11, 2021

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 May 11, 2021 • edited

Automated Checks

Review Progress

flinkbot commented May 11, 2021 • edited

CI report:

dawidwys commented May 17, 2021

dawidwys commented May 17, 2021

curcur May 18, 2021 • edited

Choose a reason for hiding this comment

curcur May 19, 2021

Choose a reason for hiding this comment

dawidwys May 19, 2021

Choose a reason for hiding this comment

curcur commented May 19, 2021

dawidwys commented May 19, 2021

curcur commented May 21, 2021

flinkbot commented May 11, 2021 •

edited

flinkbot commented May 11, 2021 •

edited

curcur May 18, 2021 •

edited