[FLINK-12960] Introduce ShuffleDescriptor#ReleaseType and ShuffleDescriptor#getSupportedReleaseTypes #8857

azagrebin · 2019-06-24T12:21:51Z

What is the purpose of the change

ResultPartitionDeploymentDescriptor#releasedOnConsumption shows the intention how the partition is going to be used by the shuffle user and released. The ShuffleDescriptor should provide a way to query which release type is supported by shuffle service for this partition. If the requested release type is not supported by the shuffle service for a certain type of partition, the job should fail fast.

Brief change log

Introduce ShuffleDescriptor#ReleaseType.AUTO and MANUAL
Introduce ShuffleDescriptor#getSupportedReleaseTypes
Add assertion whether the auto-release on consumption or manual are supported for the requested ResultPartitionDeploymentDescriptor#releaseType
adjust tests add test for the assertion

Verifying this change

Simple refactoring covered by existing tests, plus ResultPartitionDeploymentDescriptorTest#testIncompatibleReleaseTypeManual

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, 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 · 2019-06-24T12:24:15Z

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.

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❗ 3. Needs [attention] from.
- Needs attention by @zentol [PMC], @zhijiangW
❓ 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.

Details

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

azagrebin · 2019-06-24T12:25:05Z

@flinkbot attention @zentol @zhijiangW

zentol

minor comments.

zentol · 2019-06-24T13:40:06Z

flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleUtils.java

+			boolean isReleasedOnConsumption,
+			boolean isBlockingPartition) {
+		Preconditions.checkArgument(
+			!isReleasedOnConsumption || isBlockingPartition,


Long-term we probably want to change this a bit; this kind of setup forces users of the shuffle service (i.e. the ones creating a PartitionDescriptor) to be fully aware of it's capabilities.

Basically, both the partition and shuffle descriptor describe an is-state; this partition must be released on consumption, and we expect the shuffle service to fail if it can't fulfill that.

I'd like to see this changed at some point so that the partition descriptor (or some other form of input) describes a wish instead, to facilitate something like:

Scheduler: "hey shuffle service, if you could persist this partition, that would be great."
ShuffleService: "Sorry kiddo, this one must be released on consumption."
Scheduler: "oh alright then, I'll just adjust my scheduling."

The usefulness of this can be easily shown with a simple though experiment: assume there exists a shuffle service implementation that supports pipelined partitions that are consumable multiple times.
Currently, it would be impossible for the runtime to ever make use of this flag, so long as we don't resort to type checks.

flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleUtils.java

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironment.java

flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleUtils.java

zentol · 2019-06-25T10:00:02Z

...e/src/main/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptor.java

+			releasedOnConsumption ?
+				shuffleDescriptor.getSupportedReleaseTypes().contains(ReleaseType.AUTO) :
+				shuffleDescriptor.getSupportedReleaseTypes().contains(ReleaseType.MANUAL),
+			"Release on consumption <%s> is not supported by the shuffle service for this partition, " +


let's include the result partition ID as well

zentol · 2019-06-25T10:02:58Z

flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleDescriptor.java

+		 * Manually release the partition, the partition has to support consumption multiple times.
+		 *
+		 * <p>The partition requires manual actions to release it once all consumption is done:
+		 * {@link ShuffleMaster#releasePartitionExternally(ShuffleDescriptor)} and {@link ShuffleEnvironment#releasePartitionsLocally(Collection)}


this is slight ambiguous as it isn't whether the local resource condition applies to just the ShuffleEnvironment call or both of them.
I'd suggest something along the lines of ShuffleMaster[...] and, if [...], ShuffleEnvironment [...]

zhijiangW · 2019-06-25T10:19:54Z

...e/src/main/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptor.java

+			sendScheduleOrUpdateConsumersMessage,
+			// Later we might have to make the scheduling adjust automatically
+			// if certain release type is not supported by shuffle service implementation at hand
+			partitionDescriptor.getPartitionType() != ResultPartitionType.BLOCKING);


I think in future the releasedOnConsumption could be refactored into one ReleaseStrategy as proposed in #8804 , then it could be configured which strategy is used in practice (not only limited to ResultPartitionType) and check whether this config is consistent with shuffle implementation.

True, the PR just leaves it as it was before at the moment

zhijiangW · 2019-06-25T10:25:16Z

...e/src/main/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptor.java

+			int maxParallelism,
+			boolean sendScheduleOrUpdateConsumersMessage,
+			boolean releasedOnConsumption) {
+		checkReleaseOnConsumptionIsSupportedForPartition(shuffleDescriptor, releasedOnConsumption);


checkNotNull(partitionDescriptor) should be put here.

zhijiangW · 2019-06-25T10:29:36Z

flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleDescriptor.java

 	Optional<ResourceID> storesLocalResourcesOn();
+
+	/**
+	 * Return release types supported by Shuffle Service for this partition.


remove for this partition.
by this shuffle service.

why? if we leave by this shuffle service then it might sound like for all partitions but this method returns supported release types for certain partition.

There might exist two options:

One ShuffleService implementation might provide different ShuffleDescriptor implementations for different partitions with different release types. Then the previous comment makes sense.

One ShuffleService implementation would only have one kind of ShuffleDescriptor and provide one enum of release types for all the partitions. Just like atm we only have NettyShuffleDescriptor implementation which provides the same release types for all partitions. So my above comment suggestion was based on this. The release type is ShuffleService global level suitable for all partitions.

The NettyShuffleDescriptor does not have the same release types for all partitions. For non-blocking partitions we do not support ReleaseType.MANUAL.

Yes, I have no other concerns here now.

zhijiangW · 2019-06-25T10:39:04Z

flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleMaster.java

 	 * <p>This call triggers release of any resources which are occupied by the given partition in the external systems
 	 * outside of the producer executor. This is mostly relevant for the batch jobs and blocking result partitions.
 	 * This method is not called if {@link ResultPartitionDeploymentDescriptor#isReleasedOnConsumption()} is {@code true}.
+	 * The partition has to support the {@link ReleaseType#MANUAL} in {@link ShuffleDescriptor#getSupportedReleaseTypes()}.


In theory different ShuffleManager implementations could choose which release strategy is supported, we should not limit all the implementations must support one type, otherwise we could make this limitation reflected in interface methods.

The goal of PR is to let shuffle services decide what to support in ShuffleDescriptor#getSupportedReleaseTypes. This doc comment line is related to when the manual release is possible (releasePartitionLocally/releasePartitionExternally).

Or do you mean, that we implicitly assume at the moment that ReleaseType#MANUAL has to be always supported for blocking partitions? For this, we have assertion ResultPartitionDeploymentDescriptor#checkReleaseOnConsumptionIsSupportedForPartition which will fail the job fast. It is true, we have to make scheduling/release strategies automatically adjust later if something is not supported as written in the comment to the hardcoded releasedOnConsumption = partitionDescriptor.getPartitionType() != ResultPartitionType.BLOCKING.

Sorry I have not noticed that this javadoc change was for the method of releasePartitionsExternally, if so I agree it makes sense that this method implementation is for MANUL release.
Before I thought the new info was added on the class javadoc, that means all the ShuffleMaster implementation must support MANUL release type, so I pointed out that concern.

zhijiangW

Thanks for opening this PR @azagrebin !

I only reviewed the overall codes and left some minor comments, might further think it if possible.

azagrebin · 2019-06-25T12:04:47Z

Thanks for the review @zentol @zhijiangW ! I updated the PR.

zhijiangW · 2019-06-26T07:48:08Z

...c/test/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptorTest.java

-			createCopyAndVerifyResultPartitionDeploymentDescriptor(shuffleDescriptor);
-
-		ShuffleDescriptor shuffleDescriptorCopy = copy.getShuffleDescriptor();
+		ShuffleDescriptor shuffleDescriptorCopy = CommonTestUtils.createCopySerializable(shuffleDescriptor);


this change should be a separate hotfix

zhijiangW · 2019-06-26T07:57:30Z

.../src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java

 				1,
 				0),
-			NettyShuffleDescriptorBuilder.newBuilder().buildLocal(),
+			NettyShuffleDescriptorBuilder.newBuilder().setBlocking(true).buildLocal(),


might be proper for using partitionType.isBlocking() instead of direct true

zhijiangW · 2019-06-26T07:59:32Z

flink-runtime/src/test/java/org/apache/flink/runtime/util/NettyShuffleDescriptorBuilder.java

 		return this;
 	}

+	public NettyShuffleDescriptorBuilder setBlocking(boolean blocking) {


setIsBlocking(boolean isBlocking) ?

…est serialization of UnknownShuffleDescriptor without ResultPartitionDeploymentDescriptor

zhijiangW · 2019-06-26T08:04:39Z

...e/src/main/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptor.java

 	 * by {@link ShuffleMaster#releasePartitionExternally(ShuffleDescriptor)}
 	 * and {@link ShuffleEnvironment#releasePartitionsLocally(Collection)}.
 	 *
+	 * <p>The partition has to support the corresponding {@link ReleaseType} in


I think it might be better to indicate which value true/false should support which releaseType, otherwise it seems not clear the mapping between this value and release type.

zhijiangW · 2019-06-26T08:12:35Z

@azagrebin I have no other concerns except some nit comments.

…aseType and ShuffleDescriptor#getSupportedReleaseTypes `ResultPartitionDeploymentDescriptor#releasedOnConsumption` shows the intention how the partition is going to be used by the shuffle user and released. The `ShuffleDescriptor` should provide a way to query which release type is supported by shuffle service for this partition. If the requested release type is not supported by the shuffle service for a certain type of partition, the job should fail fast.

zentol self-assigned this Jun 24, 2019

azagrebin mentioned this pull request Jun 24, 2019

[FLINK-12890] Add partition lifecycle related Shuffle API #8789

Merged

rmetzger added the review=description? label Jun 24, 2019

rmetzger added the component=Runtime/Coordination label Jun 24, 2019

azagrebin force-pushed the FLINK-12960 branch from 3d281e5 to 30cdf3c Compare June 24, 2019 13:18

zentol reviewed Jun 24, 2019

View reviewed changes

azagrebin force-pushed the FLINK-12960 branch from 30cdf3c to 9806c5b Compare June 25, 2019 07:57

azagrebin closed this Jun 25, 2019

azagrebin force-pushed the FLINK-12960 branch from 74942c5 to e9642f5 Compare June 25, 2019 09:29

azagrebin reopened this Jun 25, 2019

azagrebin force-pushed the FLINK-12960 branch 2 times, most recently from 3281582 to 99fe0ab Compare June 25, 2019 09:56

zentol reviewed Jun 25, 2019

View reviewed changes

zhijiangW reviewed Jun 25, 2019

View reviewed changes

azagrebin force-pushed the FLINK-12960 branch from 99fe0ab to 258554d Compare June 25, 2019 12:04

zhijiangW reviewed Jun 26, 2019

View reviewed changes

[hotfix][tests] ResultPartitionDeploymentDescriptorTest cleanup and t…

2201b8a

…est serialization of UnknownShuffleDescriptor without ResultPartitionDeploymentDescriptor

zhijiangW reviewed Jun 26, 2019

View reviewed changes

azagrebin force-pushed the FLINK-12960 branch from 258554d to 6565a1d Compare June 26, 2019 08:06

azagrebin force-pushed the FLINK-12960 branch from 6565a1d to 31a9f54 Compare June 26, 2019 08:34

azagrebin changed the title ~~[FLINK-12960][coordination][shuffle] Move ResultPartitionDeploymentDescriptor#releasedOnConsumption to PartitionDescriptor#releasedOnConsumption~~ [FLINK-12960] Introduce ShuffleDescriptor#ReleaseType and ShuffleDescriptor#getSupportedReleaseTypes Jun 26, 2019

azagrebin force-pushed the FLINK-12960 branch from 31a9f54 to 0cb32fd Compare June 26, 2019 09:58

zentol merged commit d1ce804 into apache:master Jun 26, 2019

[FLINK-12960] Introduce ShuffleDescriptor#ReleaseType and ShuffleDescriptor#getSupportedReleaseTypes #8857

[FLINK-12960] Introduce ShuffleDescriptor#ReleaseType and ShuffleDescriptor#getSupportedReleaseTypes #8857

Uh oh!

Conversation

azagrebin commented Jun 24, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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

Uh oh!

flinkbot commented Jun 24, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Review Progress

Uh oh!

azagrebin commented Jun 24, 2019

Uh oh!

zentol left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zhijiangW Jun 25, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zentol Jun 25, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zhijiangW left a comment

Choose a reason for hiding this comment

Uh oh!

azagrebin commented Jun 25, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zhijiangW commented Jun 26, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

azagrebin commented Jun 24, 2019 •

edited

Loading

flinkbot commented Jun 24, 2019 •

edited

Loading

zhijiangW Jun 25, 2019 •

edited

Loading

zentol Jun 25, 2019 •

edited

Loading