[FLINK-38943][runtime] Support Adaptive Partition Selection for RescalePartitioner and RebalancePartitioner #27446

RocMarshal · 2026-01-20T00:15:36Z

What is the purpose of the change

[FLINK-38943][runtime] Support Adaptive Partition Selection for RescalePartitioner and RebalancePartitioner

Brief change log

Introduce the following:

config options
- taskmanager.network.adaptive-partitioner.enabled
- taskmanager.network.adaptive-partitioner.max-traverse-size
AdaptiveLoadBasedRecordWriter.java for adaptive partition

Verifying this change

This change added tests and can be verified as follows:

AdaptiveLoadBasedRecordWriterTest.java

The benchmark about it is here

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)

flinkbot · 2026-01-20T00:42:46Z

CI report:

17c175a Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

RocMarshal · 2026-01-20T04:05:06Z

Hi, @davidradl @X-czh Could you help take a look ? thx a lot.

RocMarshal · 2026-01-20T08:49:31Z

@flinkbot run azure

X-czh · 2026-01-20T12:26:55Z

@RocMarshal Thanks for the quick contribution. I'll take a look later this week.

davidradl · 2026-01-21T12:30:06Z

docs/layouts/shortcodes/generated/all_taskmanager_network_section.html

+            <td><h5>taskmanager.network.adaptive-partitioner.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether to enabled adaptive partitioner feature for rescale and rebalance partitioners based on the loading of the downstream tasks.</td>


nit: enabled => enable

Thanks, @davidradl
The naming is following the conclusion here

@RocMarshal understood- but the English is not correct.

Hi, @davidradl ,
Apologizes for the misunderstanding, and thank you for the correction.
Updated.

davidradl · 2026-01-21T12:31:26Z

docs/layouts/shortcodes/generated/all_taskmanager_network_section.html

+            <td><h5>taskmanager.network.adaptive-partitioner.max-traverse-size</h5></td>
+            <td style="word-wrap: break-word;">4</td>
+            <td>Integer</td>
+            <td>How many channels to traverse at most when looking for the idlest channel for rescale and rebalance partitioners when enabled <code class="highlighter-rouge">taskmanager.network.adaptive-partitioner.enabled</code>.</td>


Should document the default.
nit: How many channels to traverse at most -> Maximum number of channels to traverse,

the line 19 is the default value description.

davidradl · 2026-01-21T12:32:54Z

flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java

+                    .withDescription(
+                            Description.builder()
+                                    .text(
+                                            "How many channels to traverse at most when looking for the idlest channel for rescale and rebalance partitioners when enabled %s.",


nit: when enabled %s.", -> when %s is enabled .",

davidradl · 2026-01-21T12:34:31Z

.../main/java/org/apache/flink/runtime/io/network/api/writer/AdaptiveLoadBasedRecordWriter.java

+        super(writer, timeout, taskName);
+        this.numberOfSubpartitions = writer.getNumberOfSubpartitions();
+        this.maxTraverseSize = Math.min(maxTraverseSize, numberOfSubpartitions);
+        this.currentChannel = -1;


nit: does this need to be in the constructor - why not initialise the variable with -1.

davidradl · 2026-01-21T12:37:50Z

...untime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterBuilder.java

+
+    public RecordWriterBuilder<T> setMaxTraverseSize(int maxTraverseSize) {
+        Preconditions.checkState(
+                maxTraverseSize > 1, "The maxTraverseSize must be greater than 1.");


I am curious why it cannot be one

IIUC, Because if it is set to 1, this is equivalent to not enabling the adaptive partition selection feature.

Actively enabling an option that does not even provide theoretical benefits is generally not acceptable.

Unless the number of downstream partitions is exactly 1.

Pls correct me if I am wrong.

davidradl · 2026-01-21T12:40:29Z

...untime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterBuilder.java

+
+    public RecordWriterBuilder<T> setMaxTraverseSize(int maxTraverseSize) {
+        Preconditions.checkState(
+                maxTraverseSize > 1, "The maxTraverseSize must be greater than 1.");


is there a better place to check this, in configuration validation - rather than the setter?

Moved it before the RecordWriter construction

…lePartitioner & RebalancePartitioner

RocMarshal

Thanks @davidradl for the review.
I updated the related lines based on your comments.
PTAL ~

RocMarshal · 2026-01-21T14:59:25Z

docs/layouts/shortcodes/generated/all_taskmanager_network_section.html

+        </tr>
+        <tr>
+            <td><h5>taskmanager.network.adaptive-partitioner.max-traverse-size</h5></td>
+            <td style="word-wrap: break-word;">4</td>


the line is the default value description.

RocMarshal · 2026-01-21T15:01:09Z

docs/layouts/shortcodes/generated/all_taskmanager_network_section.html

+            <td><h5>taskmanager.network.adaptive-partitioner.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether to enabled adaptive partitioner feature for rescale and rebalance partitioners based on the loading of the downstream tasks.</td>


Thanks, @davidradl
The naming is following the conclusion here

RocMarshal · 2026-01-21T15:01:42Z

docs/layouts/shortcodes/generated/all_taskmanager_network_section.html

+            <td><h5>taskmanager.network.adaptive-partitioner.max-traverse-size</h5></td>
+            <td style="word-wrap: break-word;">4</td>
+            <td>Integer</td>
+            <td>How many channels to traverse at most when looking for the idlest channel for rescale and rebalance partitioners when enabled <code class="highlighter-rouge">taskmanager.network.adaptive-partitioner.enabled</code>.</td>


the line 19 is the default value description.

RocMarshal · 2026-01-21T15:21:09Z

...untime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterBuilder.java

+
+    public RecordWriterBuilder<T> setMaxTraverseSize(int maxTraverseSize) {
+        Preconditions.checkState(
+                maxTraverseSize > 1, "The maxTraverseSize must be greater than 1.");


IIUC, Because if it is set to 1, this is equivalent to not enabling the adaptive partition selection feature.

Actively enabling an option that does not even provide theoretical benefits is generally not acceptable.

Unless the number of downstream partitions is exactly 1.

Pls correct me if I am wrong.

RocMarshal · 2026-01-21T15:25:41Z

...untime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterBuilder.java

+
+    public RecordWriterBuilder<T> setMaxTraverseSize(int maxTraverseSize) {
+        Preconditions.checkState(
+                maxTraverseSize > 1, "The maxTraverseSize must be greater than 1.");


Moved it before the RecordWriter construction

RocMarshal force-pushed the FLINK-38943 branch from 046c85b to e5ea873 Compare January 20, 2026 02:36

RocMarshal mentioned this pull request Jan 20, 2026

[FLINK-38944][benchmark] Introduce a benchmark on adaptive partition selection for RescalePartitioner and RebalancePartitioner apache/flink-benchmarks#112

Open

RocMarshal force-pushed the FLINK-38943 branch from e5ea873 to b95259b Compare January 20, 2026 06:52

davidradl reviewed Jan 21, 2026

View reviewed changes

RocMarshal added 2 commits January 21, 2026 23:03

[FLINK-38943][runtime] Support Adaptive Partition Selection for Resca…

3a9963b

…lePartitioner & RebalancePartitioner

addressed review comments.

750f38c

RocMarshal force-pushed the FLINK-38943 branch from b95259b to 750f38c Compare January 21, 2026 16:07

RocMarshal commented Jan 21, 2026

View reviewed changes

github-actions bot added the community-reviewed PR has been reviewed by the community. label Jan 21, 2026

fix typos

17c175a

RocMarshal requested a review from davidradl January 22, 2026 12:36

[FLINK-38943][runtime] Support Adaptive Partition Selection for RescalePartitioner and RebalancePartitioner #27446

Are you sure you want to change the base?

[FLINK-38943][runtime] Support Adaptive Partition Selection for RescalePartitioner and RebalancePartitioner #27446

Conversation

RocMarshal commented Jan 20, 2026 • 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 Jan 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

RocMarshal commented Jan 20, 2026

Uh oh!

RocMarshal commented Jan 20, 2026

Uh oh!

X-czh commented Jan 20, 2026

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!

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!

RocMarshal Jan 21, 2026 • 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!

RocMarshal left a comment

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!

RocMarshal Jan 21, 2026 • 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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

RocMarshal commented Jan 20, 2026 •

edited

Loading

flinkbot commented Jan 20, 2026 •

edited

Loading

RocMarshal Jan 21, 2026 •

edited

Loading

RocMarshal Jan 21, 2026 •

edited

Loading