[FLINK-12203] Refactor ResultPartitionManager to break tie with Task #8210

azagrebin · 2019-04-18T13:51:42Z

What is the purpose of the change

The PR is based on #8133.

At the moment, we have ResultPartitionManager.releasePartitionsProducedBy which uses indexing by task in network environment. These methods are eventually used only by Task which already knows its partitions so Task can use ResultPartition.fail(cause) and TaskExecutor.failPartition could directly use NetworkEnviroment.releasePartitions(Collection). This also requires that JM Execution sends produced partition ids instead of just ExecutionAttemptID.

Later NetworkEnviroment.releasePartitions(Collection) could be refactored into ShuffleService.releasePartitions(Collection).

Brief change log

Change Execution to send partition ids instead of execution id to release task produced partitions
Change interface of TaskExecutorGateway.failPartition to releasePartitions
Index partitions by ResultPartitionID in ResultPartitionManager

Verifying this change

The change is simple refactoring and should be addressed by existing tests.

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-04-18T13:53: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.

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 @tillrohrmann [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.

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-04-18T13:53:53Z

@flinkbot attention @tillrohrmann @zhijiangW

tillrohrmann

Thanks for creating this refactoring @azagrebin. All in all, it looks good. I had some minor comments which we should address before merging. It would also be great if you could resolve the current merge conflicts and to see whether Travis passes.

tillrohrmann · 2019-04-24T15:09:26Z

...time/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionManager.java

@@ -41,19 +35,15 @@

 	private static final Logger LOG = LoggerFactory.getLogger(ResultPartitionManager.class);

-	public final Table<ExecutionAttemptID, IntermediateResultPartitionID, ResultPartition>
-			registeredPartitions = HashBasedTable.create();
+	private final Map<ResultPartitionID, ResultPartition> registeredPartitions = new HashMap<>();


Let's specify an initial capacity here.

tillrohrmann · 2019-04-24T15:12:07Z

...time/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionManager.java

-	}
-
-	public void releasePartitionsProducedBy(ExecutionAttemptID executionId, Throwable cause) {
+	public void releasePartitionsProducedBy(ResultPartitionID partitionId, Throwable cause) {


Let's rename this method into releasePartition

tillrohrmann · 2019-04-24T15:13:23Z

...time/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionManager.java

+				registeredPartitions.get(partitionId).release(cause);
+				registeredPartitions.remove(partitionId);
+				LOG.debug("Released partition {} produced by {}.",
+					partitionId.getPartitionId(), partitionId.getPartitionId());


I think it is better to do it the following way:

ResultPartition resultPartition = registeredPartitions.remove(partitionId); if (resultPartition != null) { resultPartition.release(cause); }

zhijiangW · 2019-04-25T08:14:12Z

flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java

@@ -667,11 +668,9 @@ private void stopTaskExecutorServices() throws Exception {
 	}

 	@Override
-	public void failPartition(ExecutionAttemptID executionAttemptID) {
-		log.info("Discarding the results produced by task execution {}.", executionAttemptID);


I am not sure whether to retain the previous log for tracing.

I was thinking about it, the problem here is that previously it happened per task. Now, it gets a list of partitions without assumption that it is per task. JM will still log it in sendReleaseIntermediateResultPartitionsRpcCall. If more verbose mode is needed, debug level will enable per partition logging in ResultPartitionManager.releasePartition in TM.

zhijiangW · 2019-04-25T08:20:50Z

...time/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionManager.java

+				registeredPartitions.get(partitionId).release(cause);
+				registeredPartitions.remove(partitionId);
+				LOG.debug("Released partition {} produced by {}.",
+					partitionId.getPartitionId(), partitionId.getPartitionId());


The last parameter in log should be partitionId.getProducerId()?

zhijiangW · 2019-04-25T08:24:31Z

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java

 		final LogicalSlot slot = assignedResource;
+		LOG.info("Discarding the results produced by task execution {}.", attemptId);


It might be better to put this log in the first line of this method.

zhijiangW · 2019-04-25T08:27:49Z

.../test/java/org/apache/flink/runtime/executiongraph/utils/SimpleAckingTaskManagerGateway.java

@@ -100,7 +102,9 @@ public String getAddress() {
 	}

 	@Override
-	public void failPartition(ExecutionAttemptID executionAttemptID) {}
+	public void releasePartitions(Collection<ResultPartitionID> partitionIds) {
+


remove empty line?

zhijiangW

Thanks for this nice refactor @azagrebin .

This work not only decouples the ExecutionAttemptID with ShuffleService, but also avoids calling NetworkEnvironment#getResultPartitionManager on TaskExecutor side. The introduced NetworkEnvironment#releasePartitions is really within our expectations.

Almost LGTM on my side, just left some minor format comments.

azagrebin · 2019-04-25T11:12:09Z

Thanks for the reviews @tillrohrmann @zhijiangW ! I've pushed a commit to address them

zhijiangW

Thanks for the updates and I have no other concerns. LGTM! 👍

tillrohrmann

Thanks for addressing my comments. Somehow all Travis builds failed. I'm not sure whether this was only a transient problem. Could you rebase this PR and trigger another build to verify this @azagrebin?

At the moment, we have ResultPartitionManager.releasePartitionsProducedBy which uses indexing by task in network environment. These methods are eventually used only by Task which already knows its partitions so Task can use ResultPartition.fail(cause) and TaskExecutor.failPartition could directly use NetworkEnviroment.releasePartitions(Collection<ResultPartitionID>). This also requires that JM Execution sends produced partition ids instead of just ExecutionAttemptID. This closes apache#8210.

tillrohrmann

Triggered another Travis build to see whether the build failures were transient. If Travis gives green light, I'll merge this PR.

At the moment, we have ResultPartitionManager.releasePartitionsProducedBy which uses indexing by task in network environment. These methods are eventually used only by Task which already knows its partitions so Task can use ResultPartition.fail(cause) and TaskExecutor.failPartition could directly use NetworkEnviroment.releasePartitions(Collection<ResultPartitionID>). This also requires that JM Execution sends produced partition ids instead of just ExecutionAttemptID. This closes apache#8210.

tillrohrmann

All builds still fail. @azagrebin could you please check why this is the case?

azagrebin · 2019-04-29T08:39:21Z

@tillrohrmann thanks for retrying
I have pushed fix, I will monitor travis build until it is mergable

azagrebin · 2019-04-30T11:56:29Z

@tillrohrmann PR should be good to go now.

tillrohrmann

Merging.

At the moment, we have ResultPartitionManager.releasePartitionsProducedBy which uses indexing by task in network environment. These methods are eventually used only by Task which already knows its partitions so Task can use ResultPartition.fail(cause) and TaskExecutor.failPartition could directly use NetworkEnviroment.releasePartitions(Collection<ResultPartitionID>). This also requires that JM Execution sends produced partition ids instead of just ExecutionAttemptID. This closes apache#8210.

azagrebin force-pushed the FLINK-12203 branch from 135a75f to c1a1c5e Compare April 18, 2019 13:52

rmetzger added the review=description? label Apr 18, 2019

rmetzger requested a review from tillrohrmann April 18, 2019 13:55

rmetzger added the component=<none> label Apr 18, 2019

azagrebin force-pushed the FLINK-12203 branch from c1a1c5e to 806b67c Compare April 18, 2019 16:46

tillrohrmann self-assigned this Apr 24, 2019

tillrohrmann requested changes Apr 24, 2019

View reviewed changes

rmetzger requested a review from tillrohrmann April 25, 2019 06:37

zhijiangW reviewed Apr 25, 2019

View reviewed changes

azagrebin force-pushed the FLINK-12203 branch from 806b67c to c5b90ba Compare April 25, 2019 11:11

zhijiangW approved these changes Apr 26, 2019

View reviewed changes

tillrohrmann requested changes Apr 26, 2019

View reviewed changes

rmetzger requested a review from tillrohrmann April 26, 2019 14:15

tillrohrmann force-pushed the FLINK-12203 branch from c5b90ba to 689f972 Compare April 27, 2019 09:49

tillrohrmann approved these changes Apr 27, 2019

View reviewed changes

rmetzger requested a review from tillrohrmann April 27, 2019 09:51

tillrohrmann force-pushed the FLINK-12203 branch from 689f972 to a250829 Compare April 28, 2019 15:32

tillrohrmann approved these changes Apr 28, 2019

View reviewed changes

rmetzger requested a review from tillrohrmann April 28, 2019 15:34

tillrohrmann requested changes Apr 28, 2019

View reviewed changes

rmetzger requested a review from tillrohrmann April 28, 2019 19:32

fix bug

2226957

rmetzger added component=Runtime/Network and removed component=<none> labels Apr 30, 2019

tillrohrmann approved these changes Apr 30, 2019

View reviewed changes

rmetzger requested a review from tillrohrmann April 30, 2019 12:15

tillrohrmann closed this in b62db93 Apr 30, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-12203] Refactor ResultPartitionManager to break tie with Task #8210

[FLINK-12203] Refactor ResultPartitionManager to break tie with Task #8210

azagrebin commented Apr 18, 2019

flinkbot commented Apr 18, 2019 •

edited

Loading

azagrebin commented Apr 18, 2019

tillrohrmann left a comment

tillrohrmann Apr 24, 2019

tillrohrmann Apr 24, 2019

tillrohrmann Apr 24, 2019

zhijiangW Apr 25, 2019

azagrebin Apr 25, 2019

zhijiangW Apr 25, 2019

zhijiangW Apr 25, 2019

zhijiangW Apr 25, 2019

zhijiangW left a comment

azagrebin commented Apr 25, 2019

zhijiangW left a comment

tillrohrmann left a comment

tillrohrmann left a comment

tillrohrmann left a comment

azagrebin commented Apr 29, 2019 •

edited

Loading

azagrebin commented Apr 30, 2019

tillrohrmann left a comment

		final LogicalSlot slot = assignedResource;
		LOG.info("Discarding the results produced by task execution {}.", attemptId);

[FLINK-12203] Refactor ResultPartitionManager to break tie with Task #8210

[FLINK-12203] Refactor ResultPartitionManager to break tie with Task #8210

Conversation

azagrebin commented Apr 18, 2019

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 Apr 18, 2019 • edited Loading

Review Progress

azagrebin commented Apr 18, 2019

tillrohrmann left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhijiangW left a comment

Choose a reason for hiding this comment

azagrebin commented Apr 25, 2019

zhijiangW left a comment

Choose a reason for hiding this comment

tillrohrmann left a comment

Choose a reason for hiding this comment

tillrohrmann left a comment

Choose a reason for hiding this comment

tillrohrmann left a comment

Choose a reason for hiding this comment

azagrebin commented Apr 29, 2019 • edited Loading

azagrebin commented Apr 30, 2019

tillrohrmann left a comment

Choose a reason for hiding this comment

flinkbot commented Apr 18, 2019 •

edited

Loading

azagrebin commented Apr 29, 2019 •

edited

Loading