New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[FLINK-28663][runtime] Allow multiple downstream consumer job vertices sharing the same intermediate dataset at scheduler side #20350

Merged

wsry merged 2 commits into apache:master from wsry:FLINK-28663

Aug 8, 2022

Contributor

wsry commented Jul 25, 2022

What is the purpose of the change

Currently, one intermediate dataset can only be consumed by one downstream consumer vertex. If there are multiple consumer vertices consuming the same output of the same upstream vertex, multiple intermediate datasets will be produced. We can optimize this behavior to produce only one intermediate dataset which can be shared by multiple consumer vertices. As the first step, we should allow multiple downstream consumer job vertices sharing the same intermediate dataset at scheduler side. (Note that this optimization only works for blocking shuffle because pipelined shuffle result partition can not be consumed multiple times)

Brief change log

Allow multiple downstream consumer job vertices sharing the same intermediate dataset at scheduler side

Verifying this change

This change added 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, 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)

wsry requested a review from zhuzhurk

July 25, 2022 08:10

Collaborator

flinkbot commented Jul 25, 2022 •

edited

Loading

CI report:

77e307b Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

flinkbot added the component=Runtime/Coordination label

wsry added a commit to wsry/flink that referenced this pull request


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

08ea37e

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

wsry force-pushed the FLINK-28663 branch from 120097c to 9f8ed98 Compare

July 27, 2022 12:02

wsry added a commit to wsry/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

9f8ed98

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, JobIntermediateDatasetReuseTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

b55e1c9

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

afb0cd0

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, JobIntermediateDatasetReuseTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

fa83c8c

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

eed4e46

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, JobIntermediateDatasetReuseTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

f22418d

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

d1c4ca3

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, JobIntermediateDatasetReuseTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

4afc212

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, JobIntermediateDatasetReuseTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

wsry force-pushed the FLINK-28663 branch from 9f8ed98 to 4afc212 Compare

July 29, 2022 06:59

zhuzhurk reviewed

View reviewed changes

Contributor

zhuzhurk left a comment

Thanks for opening this PR! @wsry
I have just reviewed the test migration part and here are some comments for it.
Will continue reviewing the core changes commit.

...test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphConstructionTest.java Outdated Show resolved Hide resolved

...test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphConstructionTest.java Outdated Show resolved Hide resolved

flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertexTest.java Outdated Show resolved Hide resolved

flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertexTest.java Outdated Show resolved Hide resolved

...e/src/test/java/org/apache/flink/runtime/executiongraph/IntermediateResultPartitionTest.java Outdated Show resolved Hide resolved

...e/src/test/java/org/apache/flink/runtime/scheduler/adapter/DefaultExecutionTopologyTest.java Outdated Show resolved Hide resolved

...ime/src/test/java/org/apache/flink/runtime/scheduler/adapter/DefaultExecutionVertexTest.java Outdated Show resolved Hide resolved

...ime/src/test/java/org/apache/flink/runtime/scheduler/adapter/DefaultResultPartitionTest.java Outdated Show resolved Hide resolved

...ime/src/test/java/org/apache/flink/runtime/scheduler/adapter/DefaultResultPartitionTest.java Outdated Show resolved Hide resolved

...g/apache/flink/runtime/scheduler/adaptivebatch/forwardgroup/ForwardGroupComputeUtilTest.java Show resolved Hide resolved

TanYuxin-tyx reviewed

View reviewed changes

flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java Outdated Show resolved Hide resolved

zhuzhurk reviewed

View reviewed changes

Contributor

zhuzhurk left a comment

Nice work!
Here are a few comments.

flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobEdge.java Show resolved Hide resolved

...untime/src/main/java/org/apache/flink/runtime/scheduler/strategy/ConsumedPartitionGroup.java Show resolved Hide resolved

...ntime/src/main/java/org/apache/flink/runtime/scheduler/SsgNetworkMemoryCalculationUtils.java Outdated Show resolved Hide resolved

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

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/IntermediateResult.java Show resolved Hide resolved

...k-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java Outdated Show resolved Hide resolved

...k-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java Outdated Show resolved Hide resolved

...k-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java Outdated Show resolved Hide resolved

...e/src/test/java/org/apache/flink/runtime/executiongraph/IntermediateResultPartitionTest.java Show resolved Hide resolved

...src/test/java/org/apache/flink/runtime/executiongraph/RemoveCachedShuffleDescriptorTest.java Show resolved Hide resolved

wsry added a commit to wsry/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

8812cc6

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, JobIntermediateDatasetReuseTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

wsry force-pushed the FLINK-28663 branch from 4afc212 to 8812cc6 Compare

August 4, 2022 15:31

wsry added a commit to wsry/flink that referenced this pull request


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

b56c1cb

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

3e19b4a

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, JobIntermediateDatasetReuseTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

wsry force-pushed the FLINK-28663 branch from 8812cc6 to 3e19b4a Compare

August 4, 2022 15:35

wsry added a commit to wsry/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

00e174e

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, JobIntermediateDatasetReuseTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

d6c1839

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

wsry force-pushed the FLINK-28663 branch from 3e19b4a to 8e4a08e Compare

August 5, 2022 04:45

Contributor Author

wsry commented Aug 5, 2022

@zhuzhurk @TanYuxin-tyx Thanks for the review and feedbacks. I have updated the PR.

wsry added a commit to wsry/flink that referenced this pull request


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

3dd2bcf

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

a218e9a

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, JobIntermediateDatasetReuseTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

6550ee4

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, JobIntermediateDatasetReuseTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

wsry force-pushed the FLINK-28663 branch from 8e4a08e to a60cdfa Compare

August 5, 2022 08:09

zhuzhurk reviewed

View reviewed changes

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java Outdated Show resolved Hide resolved

...ntime/src/main/java/org/apache/flink/runtime/executiongraph/IntermediateResultPartition.java Outdated Show resolved Hide resolved

...k-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java Outdated Show resolved Hide resolved

Contributor Author

wsry commented Aug 6, 2022

@zhuzhurk I have updated the PR and add two fixup commits. Please take another look.

wsry added a commit to wsry/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

46cda30

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, JobIntermediateDatasetReuseTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

20cedd9

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

wsry force-pushed the FLINK-28663 branch from a60cdfa to 356fd06 Compare

August 6, 2022 04:29

Contributor Author

wsry commented Aug 6, 2022

@flinkbot run azure

zhuzhurk approved these changes

View reviewed changes

Contributor

zhuzhurk left a comment

Thanks for addressing all the comments! @wsry
The change looks good to me.
Would you squash it? I will merge it after CI gives green.

wsry added a commit to wsry/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

c70dc72

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

f6f309e

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

wsry force-pushed the FLINK-28663 branch from 356fd06 to f6f309e Compare

August 7, 2022 12:22

Contributor Author

wsry commented Aug 7, 2022

Thanks for addressing all the comments! @wsry The change looks good to me. Would you squash it? I will merge it after CI gives green.

@zhuzhurk Thanks a lot. Rebased and squashed.

wsry added a commit to wsry/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

936913e

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

wsry force-pushed the FLINK-28663 branch from f6f309e to 936913e Compare

August 8, 2022 06:13

wsry added a commit to wsry/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

a02e26e

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

68c1b46

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

wsry added 2 commits

August 8, 2022 16:24


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

6c36374

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

77e307b

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

wsry force-pushed the FLINK-28663 branch from 936913e to 77e307b Compare

August 8, 2022 08:25

wsry added a commit to wsry/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

dadd72e

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

wsry added a commit to wsry/flink that referenced this pull request


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

672202e

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

wsry merged commit 7240536 into apache:master

wsry added a commit that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

b3be6bb

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes #20350.

huangxiaofeng10047 pushed a commit to huangxiaofeng10047/flink that referenced this pull request


[hotfix][tests] Migrate tests relevant to FLINK-28663 to Junit5/AssertJ

7b11404

Migrated tests include DefaultExecutionGraphConstructionTest, EdgeManagerBuildUtilTest, EdgeManagerTest, ExecutionJobVertexTest, IntermediateResultPartitionTest, RemoveCachedShuffleDescriptorTest, JobTaskVertexTest, DefaultExecutionTopologyTest, DefaultExecutionVertexTest, DefaultResultPartitionTest, AdaptiveBatchSchedulerTest and ForwardGroupComputeUtilTest.

This closes apache#20350.

huangxiaofeng10047 pushed a commit to huangxiaofeng10047/flink that referenced this pull request


[FLINK-28663][runtime] Allow multiple downstream consumer job vertice…

0885de8

…s sharing the same intermediate dataset at scheduler side

This closes apache#20350.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment