[FLINK-17014][runtime] Implement PipelinedRegionSchedulingStrategy #11770

zhuzhurk · 2020-04-16T08:53:46Z

What is the purpose of the change

Implement PipelinedRegionSchedulingStrategy which schedules tasks in granularity of pipelined regions. More details see https://cwiki.apache.org/confluence/display/FLINK/FLIP-119+Pipelined+Region+Scheduling.

Brief change log

Implemented PipelinedRegionSchedulingStrategy

Verifying this change

Added unit tests PipelinedRegionSchedulingStrategyTest

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-04-16T08:55:57Z

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 2577ed1 (Thu Apr 16 08:55:57 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-04-16T09:15:47Z

CI report:

17d9f5e UNKNOWN
a47976d Travis: SUCCESS 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

GJL

The PipelinedRegionSchedulingStrategy implementation looks neat!

...main/java/org/apache/flink/runtime/scheduler/strategy/PipelinedRegionSchedulingStrategy.java

...ntime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyUtils.java

...main/java/org/apache/flink/runtime/scheduler/strategy/PipelinedRegionSchedulingStrategy.java

.../java/org/apache/flink/runtime/scheduler/strategy/PipelinedRegionSchedulingStrategyTest.java

GJL · 2020-04-20T07:18:15Z

.../java/org/apache/flink/runtime/scheduler/strategy/PipelinedRegionSchedulingStrategyTest.java

+
+	private TestingSchedulerOperations testingSchedulerOperation;
+
+	private int parallelism = 2;


Consider making this static final or convert to a local variable.

GJL · 2020-04-20T07:34:43Z

.../test/java/org/apache/flink/runtime/scheduler/strategy/TestingSchedulingPipelinedRegion.java

+
+	@Override
+	public Iterable<TestingSchedulingExecutionVertex> getVertices() {
+		return regionVertices.values();


Consider wrapping it in Collections.unmodifiableCollection() for immutability or returning a copy. Same in getConsumedResults().

Iterable is considered as unmodifiable so I think it's not very needed.

I don't think that's true, see Iterator#remove()

You are right. Will wrap it with Collections.unmodifiableCollection().

GJL · 2020-04-20T07:36:26Z

...ime/src/test/java/org/apache/flink/runtime/scheduler/strategy/TestingSchedulingTopology.java

@@ -78,6 +83,31 @@ public TestingSchedulingResultPartition getResultPartition(final IntermediateRes
 		return resultPartition;
 	}

+	@Override
+	public Iterable<SchedulingPipelinedRegion> getAllPipelinedRegions() {
+		return vertexRegions.values().stream().collect(Collectors.toSet());


I think that's the same as new HashSet<>(vertexRegions.values())

Ok. Let's simplify it.

GJL · 2020-04-20T07:37:24Z

...ime/src/test/java/org/apache/flink/runtime/scheduler/strategy/TestingSchedulingTopology.java

+		return vertexRegions.get(vertexId);
+	}
+
+	void generatePipelinedRegions() {


Consider invoking this method lazily in getAllPipelinedRegions() and getPipelinedRegionOfVertex() instead of relying the client (test) to invoke it.

Yes that would be better.
The only problem is that the regions cannot be refreshed if the topology is changed after invoking getAllPipelinedRegions(). So I will clear the cached regions on any topology changes.

… region getter interfaces

zhuzhurk · 2020-04-20T10:44:04Z

Will merge it once travis gives green.

zhuzhurk requested a review from GJL April 16, 2020 08:54

rmetzger added review=description? component=Runtime/Coordination labels Apr 16, 2020

GJL self-assigned this Apr 16, 2020

GJL reviewed Apr 16, 2020

View reviewed changes

zhuzhurk force-pushed the FLINK_17014_pipelined_region_scheduling branch from 2577ed1 to 395043e Compare April 17, 2020 06:14

GJL mentioned this pull request Apr 17, 2020

[FLINK-17181][runtime] Drop generic Types in SchedulingTopology Interface #11783

Closed

zhuzhurk force-pushed the FLINK_17014_pipelined_region_scheduling branch from 395043e to d8568c4 Compare April 19, 2020 00:59

GJL approved these changes Apr 20, 2020

View reviewed changes

zhuzhurk force-pushed the FLINK_17014_pipelined_region_scheduling branch from d8568c4 to 17d9f5e Compare April 20, 2020 08:26

zhuzhurk added 2 commits April 20, 2020 16:47

[FLINK-17014][runtime] TestingSchedulingTopology implements pipelined…

ebbb158

… region getter interfaces

[FLINK-17014][runtime] Implement PipelinedRegionSchedulingStrategy

a47976d

zhuzhurk force-pushed the FLINK_17014_pipelined_region_scheduling branch from 17d9f5e to a47976d Compare April 20, 2020 08:49

zhuzhurk merged commit 99cbaa9 into apache:master Apr 21, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-17014][runtime] Implement PipelinedRegionSchedulingStrategy #11770

[FLINK-17014][runtime] Implement PipelinedRegionSchedulingStrategy #11770

zhuzhurk commented Apr 16, 2020

flinkbot commented Apr 16, 2020

flinkbot commented Apr 16, 2020 •

edited

GJL left a comment

GJL Apr 20, 2020

zhuzhurk Apr 20, 2020

GJL Apr 20, 2020

zhuzhurk Apr 20, 2020

GJL Apr 20, 2020

zhuzhurk Apr 20, 2020

GJL Apr 20, 2020

zhuzhurk Apr 20, 2020

GJL Apr 20, 2020

zhuzhurk Apr 20, 2020

zhuzhurk commented Apr 20, 2020


		private TestingSchedulerOperations testingSchedulerOperation;

		private int parallelism = 2;

[FLINK-17014][runtime] Implement PipelinedRegionSchedulingStrategy #11770

[FLINK-17014][runtime] Implement PipelinedRegionSchedulingStrategy #11770

Conversation

zhuzhurk commented Apr 16, 2020

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 16, 2020

Automated Checks

Review Progress

flinkbot commented Apr 16, 2020 • edited

CI report:

GJL 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhuzhurk commented Apr 20, 2020

flinkbot commented Apr 16, 2020 •

edited