[FLINK-17861][task][checkpointing] Split channel state handles sent to JM #12292

rkhachatryan · 2020-05-22T09:06:17Z

What is the purpose of the change

If Unaligned checkpoints are enabled, channel state is written as state handles. Each channel has a handle and each such handle references the same underlying streamStateHandle (this is done to have a single file per subtask).
But, if the state is less then state.backend.fs.memory-threshold, the data is sent directly to JM as a byteStreamHandle. This causes each channel state handle to hold the whole subtask state.

This PR solves this by extracting relevant potions of the underlying handles if they are byteStreamHandles.

Another approach would be to move the underlying state handle one level up (OperatorSubtaskState) and not store references to it in channelStateHandles. It would be more effective (less data duplication) but also more error-prone (implicit structure), less flexible (re-scaling), and require more changes.

Verifying this change

This change is already covered by existing tests, such as:

ChannelStateCheckpointWriterTest.testRecordingOffsets (the test was corrected)
ChannelPersistenceITCase

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

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

Documentation

Does this pull request introduce a new feature? no
If yes, how is the feature documented? not applicable

flinkbot · 2020-05-22T09:08:07Z

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 96bfb4f (Fri May 22 09:08:06 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-05-22T09:15:59Z

CI report:

96bfb4f UNKNOWN
7cbe552 Azure: CANCELED

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

pnowojski

Another approach would be to move the underlying state handle one level up (OperatorSubtaskState) and not store references to it in channelStateHandles. It would be more effective (less data duplication) but also more error-prone (implicit structure), less flexible (re-scaling), and require more changes.

Could you elaborate a bit more? What's the alternative? How would it avoid more data duplication? Are we still duplicating data with this PR?

.../test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateCheckpointWriterTest.java

.../src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateCheckpointWriter.java

pnowojski · 2020-05-25T13:08:25Z

.../src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateCheckpointWriter.java

+		if (underlying instanceof ByteStreamStateHandle) {
+			ByteStreamStateHandle byteHandle = (ByteStreamStateHandle) underlying;
+			return buildHandle.apply(
+				e.getKey(),
+				new ByteStreamStateHandle(randomUUID().toString(), serializer.extractAndMerge(byteHandle.getData(), e.getValue())),
+				singletonList(serializer.getHeaderLength()));
+		} else {


It doesn't look like a generic/error prone solution. It looks like something should be extracted to the StreamStateHandle. boolean StreamStateHandle#isDirectlyHoldingData()? Or we are missing some kind of different abstraction of shared state handles.

Why is this issue new for spilled channel state? What's different for operators state?

Added StreamStateHandle.asBytesIfInMemory that returns Optional<byte[]> (better names are welcomed ;)

Why is this issue new for spilled channel state? What's different for operators state?

Operators don't share state handles with each other so they don't have this problem.

I think it's better now, but I still can not put a finger on a feeling that something might not be entirely correct.

Why do we have multiple handles per channel, instead of a single handle that has multiple channels?

There is only one InputChannelStateHandle per channel.
Putting multiple channels could make recovery (rescaling) more difficult.

After an offline discussion, we decided to keep it for now as it is. Could you create a JIRA ticket to revisit this issue later (linked to how to handle rescaling) and linked it as TODO here?

Sure. Created a JIRA ticket: https://issues.apache.org/jira/browse/FLINK-17972

rkhachatryan · 2020-05-25T22:00:14Z

Thanks for the feedback @pnowojski ,

I've addressed the issues (except this one).

Answering your question:

Could you elaborate a bit more? What's the alternative? How would it avoid more data duplication? Are we still duplicating data with this PR?

Current structure is the following (this PR doesn't change it):

Each subtask reports to JM TaskStateSnapshot, 
    each with zero ore more OperatorSubtaskState,
        each with zero or more InputChannelStateHandle and ResultSubpartitionStateHandle
            each referencing an underlying StreamStateHandle

The underlying StreamStateHandle duplicates filename (ByteStreamStateHandle has it too at least because of equals/hashcode I guess).

An alternative would be something like

Each subtask reports to JM TaskStateSnapshot, 
    each with zero ore more OperatorSubtaskState,
        each with zero or one StreamStateHandle (for channel state)
        each with zero or more InputChannelStateHandle and ResultSubpartitionStateHandle

pnowojski · 2020-05-27T11:53:22Z

My questions probing in the direction of single channel state handle seems to be going in a very similar direction as the alternate solution that you proposed: moving the shared StreamStateHandle one level up, but also additionally encapsulating them into something, for example:

Each subtask reports to JM TaskStateSnapshot, 
    each with zero ore more OperatorSubtaskState,
        each with zero or one InFlightDataStateHandle
            each with one (or more for rescaling) StreamStateHandle
            each with zero or more InputChannelStateHandle and ResultSubpartitionStateHandle

…tRecordingOffsets 1. Set writer position in NetworkBuffer passed to ChannelStateCheckpointWriter.write 2. Reduce state size to fit in the configured MemoryCheckpointOutputStream

… deserializer Motivation: 1. add a method that deserializes and then serializes data 2. simplify

…o JM Before this commit, if Unaligned checkpoints are enabled, channel state is written as state handles. Each channel has a handle and each such handle references the same underlying streamStateHandle (this is done to have a single file per subtask). But, if the state is less then state.backend.fs.memory-threshold, the data is sent directly to JM as a byteStreamHandle. This causes each channel state handle to hold the whole subtask state. This change solves this by extracting relevant potions of the underlying handles if they are byteStreamHandles.

rkhachatryan · 2020-05-27T22:25:37Z

@flinkbot run azure

pnowojski · 2020-05-28T06:13:47Z

Merging from #12362 . Build was successful there, here e2e timed out because of getting artefacts from mvn cache took very long time

rmetzger added the review=description? label May 22, 2020

rkhachatryan force-pushed the uc-split-byte-state-handles branch from 96bfb4f to 14a727c Compare May 22, 2020 09:08

rkhachatryan marked this pull request as ready for review May 22, 2020 09:08

rmetzger added component=Runtime/Checkpointing component=Runtime/Task labels May 22, 2020

pnowojski reviewed May 25, 2020

View reviewed changes

rkhachatryan force-pushed the uc-split-byte-state-handles branch from 14a727c to 1d9ab48 Compare May 25, 2020 21:08

rkhachatryan force-pushed the uc-split-byte-state-handles branch 2 times, most recently from c9a38fe to 3814750 Compare May 26, 2020 21:24

rkhachatryan added 3 commits May 27, 2020 15:09

[FLINK-17861][tests][hotfix] Fix ChannelStateCheckpointWriterTest.tes…

7ec5221

…tRecordingOffsets 1. Set writer position in NetworkBuffer passed to ChannelStateCheckpointWriter.write 2. Reduce state size to fit in the configured MemoryCheckpointOutputStream

[FLINK-17861][task][checkpointing] Merge channel state serializer and…

c7741af

… deserializer Motivation: 1. add a method that deserializes and then serializes data 2. simplify

rkhachatryan force-pushed the uc-split-byte-state-handles branch from 3814750 to 7cbe552 Compare May 27, 2020 13:11

rkhachatryan mentioned this pull request May 27, 2020

[FLINK-17928][checkpointing] Fix ChannelStateHandle size #12362

Merged

pnowojski approved these changes May 27, 2020

View reviewed changes

pnowojski closed this May 28, 2020

rkhachatryan deleted the uc-split-byte-state-handles branch August 17, 2020 17:52

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[FLINK-17861][task][checkpointing] Split channel state handles sent to JM #12292

[FLINK-17861][task][checkpointing] Split channel state handles sent to JM #12292

Uh oh!

rkhachatryan commented May 22, 2020

Uh oh!

flinkbot commented May 22, 2020

Uh oh!

flinkbot commented May 22, 2020 •

edited

Loading

Uh oh!

pnowojski left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

pnowojski May 25, 2020

Uh oh!

rkhachatryan May 25, 2020

Uh oh!

pnowojski May 26, 2020

Uh oh!

rkhachatryan May 26, 2020

Uh oh!

pnowojski May 27, 2020

Uh oh!

rkhachatryan May 27, 2020

Uh oh!

rkhachatryan commented May 25, 2020 •

edited

Loading

Uh oh!

pnowojski commented May 27, 2020 •

edited

Loading

Uh oh!

rkhachatryan commented May 27, 2020

Uh oh!

pnowojski commented May 28, 2020

Uh oh!

Uh oh!

[FLINK-17861][task][checkpointing] Split channel state handles sent to JM #12292

[FLINK-17861][task][checkpointing] Split channel state handles sent to JM #12292

Uh oh!

Conversation

rkhachatryan commented May 22, 2020

What is the purpose of the change

Verifying this change

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

Documentation

Uh oh!

flinkbot commented May 22, 2020

Automated Checks

Review Progress

Uh oh!

flinkbot commented May 22, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

pnowojski left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

pnowojski May 25, 2020

Choose a reason for hiding this comment

Uh oh!

rkhachatryan May 25, 2020

Choose a reason for hiding this comment

Uh oh!

pnowojski May 26, 2020

Choose a reason for hiding this comment

Uh oh!

rkhachatryan May 26, 2020

Choose a reason for hiding this comment

Uh oh!

pnowojski May 27, 2020

Choose a reason for hiding this comment

Uh oh!

rkhachatryan May 27, 2020

Choose a reason for hiding this comment

Uh oh!

rkhachatryan commented May 25, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

pnowojski commented May 27, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

rkhachatryan commented May 27, 2020

Uh oh!

pnowojski commented May 28, 2020

Uh oh!

Uh oh!

flinkbot commented May 22, 2020 •

edited

Loading

rkhachatryan commented May 25, 2020 •

edited

Loading

pnowojski commented May 27, 2020 •

edited

Loading