[FLINK-40780][runtime] Make FORWARD-to-REBALANCE downgrade on parallelism mismatch configurable by 1996fanrui · Pull Request #29272 · apache/flink

1996fanrui · 2026-09-23T16:53:53Z

This PR is stacked on #29374 (FLINK-40904), which fixes the mismatch detection. Only the last two commits belong to this PR.

What is the purpose of the change

When the producer and consumer of a FORWARD edge end up with different parallelism (e.g. after rescaling via the AdaptiveScheduler or pipeline.jobvertex-parallelism-overrides), StreamTask silently replaces the ForwardPartitioner with a RebalancePartitioner (FLINK-30213). That's fine for append-only streams, but it reorders records in changelog streams and produces wrong results. This PR makes the behavior configurable.

Brief change log

Add pipeline.forward-edge.parallelism-mismatch-mode (enum):
- REBALANCE (default): current behavior, which may reorder records.
- FAIL: fail the job with an exception, for setups where a mismatch is never expected.
- KEEP_FORWARD: keep the forward partitioner (pre-FLINK-30213 behavior; order is preserved, and records go to a single consumer subtask).
StreamTask#replaceForwardPartitionerIfConsumerParallelismDoesNotMatch applies the configured mode.

Verifying this change

StreamTaskTest covers the three modes.
ForwardEdgeParallelismMismatchOverridesITCase and ForwardEdgeParallelismMismatchRestRescaleITCase cover every combination of scheduler (Default, Adaptive, AdaptiveBatch), parallelism change and mode, changing the parallelism at submission via pipeline.jobvertex-parallelism-overrides or at runtime via the resource requirements API.

Does this pull request potentially affect one of the following parts

Dependencies: no
The public API: yes (new @PublicEvolving config option)
The serializers: no
The runtime per-record code paths: no (the check runs once when record writers are created)
Anything that affects deployment or recovery: no
The S3 file system connector: no

Documentation

New feature: yes
Docs: the config docs were regenerated (pipeline_configuration.html)

flinkbot · 2026-09-23T17:06:38Z

CI report:

2e53679 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

rkhachatryan · 2026-09-23T23:00:11Z

+        final int producerParallelism = environment.getTaskInfo().getNumberOfParallelSubtasks();
+        final int consumerParallelism =
+                environment.getWriter(outputIndex).getNumberOfSubpartitions();


getNumberOfSubpartitions() isn't the consumer parallelism on a POINTWISE (forward) edge. In a static graph it's the size of the consumer vertex group for this producer subtask (IntermediateResultPartition#getNumberOfSubpartitions), so it's 1 when producer and consumer parallelism are equal. In a dynamic graph (adaptive batch) a forward result always has exactly 1 subpartition (computeNumberOfSubpartitionsForDynamicGraph). So the != producerParallelism check is wrong both ways.

False positive: any non-chained forward edge with parallelism > 1 and no rescale counts as a mismatch. Before this PR that was harmless, since the result was a RebalancePartitioner over a single channel. Now FAIL rejects such jobs and KEEP_FORWARD logs a WARN for every subtask.

False negative: producer 2 → consumer 4 gives 2 subpartitions per producer, which equals the producer parallelism of 2, so nothing is detected. FAIL doesn't fail and REBALANCE doesn't rebalance, and half the consumers get no data.

I checked this with a MiniCluster ITCase on this branch, with mode FAIL, source → forward() → map (startNewChain):

Case Result

streaming 4 → 4, no rescale fails: producer parallelism 4 != consumer parallelism 1

batch 4 → 4, no rescale fails, same error

streaming 2 → 4 (via pipeline.jobvertex-parallelism-overrides) succeeds, records per consumer subtask {0=500, 1=0, 2=500, 3=0}

streaming 1 → 2 (the case covered by StreamTaskTest) fails, as expected

The existing tests only cover 1 → 2, where the two quantities happen to coincide.

The error and WARN messages have the same mix-up: they report the subpartition count as "consumer parallelism", e.g. "consumer parallelism 1" for a consumer running at 4.

I think the check needs the actual consumer vertex parallelism, which I don't believe is available in the task today, so it may have to be passed through the deployment descriptor or NonChainedOutput. Alternatively, the mismatch could be decided at scheduling time. Could you also add tests for equal parallelism > 1 (streaming and batch) and for N → M with N > 1?

Good catch, I created a new pr to fix it since it is an existing bug.

Regarding tests, added two ITCases, each covering 11 parallelism changes × 3 modes:

ForwardEdgeParallelismMismatchOverridesITCase: changes the parallelism via pipeline.jobvertex-parallelism-overrides under the Default, Adaptive and AdaptiveBatch schedulers (99 cases).

ForwardEdgeParallelismMismatchRestRescaleITCase: changes it at runtime via the resource requirements API under the AdaptiveScheduler (30 cases).

All 129 pass.

Reading the consumer parallelism in StreamTask from getWriter(outputIndex).getNumberOfSubpartitions() again (a one-line revert) makes 117 fail; only the 1 -> 2 cases pass, since there the two values happen to match.

Efrat19

Thanks
I wonder if ForwardEdgeParallelismMismatchMode could be auto-inferred instead of a config option most users won't understand?
For example Give ForwardPartitioner a field for the mismatch mode, and have the planner set it to REBALANCE on insert-only edges, and KEEP_FORWARD on changelog edges, from a table option.

wdyt?

Efrat19 · 2026-10-15T07:22:25Z

+            FORWARD_EDGE_PARALLELISM_MISMATCH_MODE =
+                    key("pipeline.forward-edge.parallelism-mismatch-mode")
+                            .enumType(ForwardEdgeParallelismMismatchMode.class)
+                            .defaultValue(ForwardEdgeParallelismMismatchMode.REBALANCE)


IIUC REBALANCE breaks changelog mode - should it be the default?

Good point. REBALANCE stays the default to keep the existing behavior, so jobs relying on it today (e.g. append-only jobs rescaled by the AdaptiveScheduler) aren't affected. Changelog jobs can opt into FAIL or KEEP_FORWARD.

Efrat19 · 2026-10-15T07:23:56Z

+        final String taskNameWithSubtasks = environment.getTaskInfo().getTaskNameWithSubtasks();
+        switch (mode) {
+            case REBALANCE:
+                LOG.debug(


Nit? as it can be harmful for changelog

Suggested change

LOG.debug(

LOG.warn(

Agreed, changed it to WARN and the message now states that records may be reordered and how to avoid it.

Efrat19 · 2026-10-15T07:37:24Z

+                // An undecided consumer parallelism is not a mismatch.
+                || consumerParallelism
+                        == ResultPartitionDeploymentDescriptor.UNKNOWN_CONSUMER_PARALLELISM
+                || consumerParallelism == producerParallelism) {


wdyt about

Suggested change

|| consumerParallelism == producerParallelism) {

|| consumerParallelism <= producerParallelism) {

maybe we can keep the forward partitioner in this case as well, since each producer will still forward to 1 consumer?

I'd prefer each mode to mean exactly one behavior:

REBALANCE always rebalances

KEEP_FORWARD always keeps forward

Special-casing downscale would make REBALANCE silently behave like KEEP_FORWARD and may bring other risks we haven't thought of. Users who want that can set KEEP_FORWARD.

Efrat19 · 2026-10-15T07:45:47Z

+                        producerParallelism,
+                        consumerParallelism);
+                break;
+            case FAIL:


I don't understand why do we need the fail mode?
Wouldn't it always be preferred to downgrade performance with warn (KEEP_FORWARD) other than crash?

FAIL is for setups where a parallelism mismatch is never expected, so it's surfaced immediately instead of being handled silently.

1996fanrui

Thanks I wonder if ForwardEdgeParallelismMismatchMode could be auto-inferred instead of a config option most users won't understand? For example Give ForwardPartitioner a field for the mismatch mode, and have the planner set it to REBALANCE on insert-only edges, and KEEP_FORWARD on changelog edges, from a table option.

Thanks, nice idea. It would only cover SQL though, order-sensitive DataStream jobs (e.g. CDC or per-key state machines) still need the option. I'd keep this PR as the runtime safeguard and leave planner inference as a follow-up.

WDYT?

1996fanrui · 2026-10-15T19:33:17Z

+            FORWARD_EDGE_PARALLELISM_MISMATCH_MODE =
+                    key("pipeline.forward-edge.parallelism-mismatch-mode")
+                            .enumType(ForwardEdgeParallelismMismatchMode.class)
+                            .defaultValue(ForwardEdgeParallelismMismatchMode.REBALANCE)


Good point. REBALANCE stays the default to keep the existing behavior, so jobs relying on it today (e.g. append-only jobs rescaled by the AdaptiveScheduler) aren't affected. Changelog jobs can opt into FAIL or KEEP_FORWARD.

1996fanrui · 2026-10-15T19:33:19Z

+        final String taskNameWithSubtasks = environment.getTaskInfo().getTaskNameWithSubtasks();
+        switch (mode) {
+            case REBALANCE:
+                LOG.debug(


Agreed, changed it to WARN and the message now states that records may be reordered and how to avoid it.

1996fanrui · 2026-10-15T19:33:21Z

+                // An undecided consumer parallelism is not a mismatch.
+                || consumerParallelism
+                        == ResultPartitionDeploymentDescriptor.UNKNOWN_CONSUMER_PARALLELISM
+                || consumerParallelism == producerParallelism) {


I'd prefer each mode to mean exactly one behavior:

REBALANCE always rebalances

KEEP_FORWARD always keeps forward

Special-casing downscale would make REBALANCE silently behave like KEEP_FORWARD and may bring other risks we haven't thought of. Users who want that can set KEEP_FORWARD.

1996fanrui · 2026-10-15T19:33:22Z

+                        producerParallelism,
+                        consumerParallelism);
+                break;
+            case FAIL:


FAIL is for setups where a parallelism mismatch is never expected, so it's surfaced immediately instead of being handled silently.

Efrat19

LGTM, thanks for addressing the notes

…lism mismatch configurable When a FORWARD edge connects a producer and consumer whose parallelism no longer match (e.g. after a rescale via the AdaptiveScheduler or pipeline.jobvertex-parallelism-overrides), the runtime silently replaces the ForwardPartitioner with a RebalancePartitioner (FLINK-30213). This is safe for append-only streams but silently reorders order-sensitive (changelog) streams, producing incorrect results. Add pipeline.forward-edge.parallelism-mismatch-mode to control this behavior: - REBALANCE (default): current behavior, warn log. - FAIL: reject the job with an exception. - KEEP_FORWARD: keep the forward partitioner (order-preserving), warn log.

…h modes Cover every combination of scheduler (Default, Adaptive, AdaptiveBatch), parallelism change and mismatch mode, with the new parallelism applied either at submission via pipeline.jobvertex-parallelism-overrides or at runtime via the resource requirements API of the AdaptiveScheduler.

1996fanrui · 2026-10-21T11:10:08Z

The CI failure in MiniClusterJobEntryPointITCase is unrelated, it's the REST port race tracked in FLINK-41048.

1996fanrui · 2026-10-21T11:12:27Z

Thank you @Efrat19 for the review, merging

rkhachatryan reviewed Sep 23, 2026

View reviewed changes

1996fanrui mentioned this pull request Oct 2, 2026

[FLINK-40904][runtime] Detect FORWARD parallelism mismatch with the actual consumer parallelism #29374

Merged

1996fanrui force-pushed the FLINK-40780 branch 2 times, most recently from 1b897e7 to 6c77218 Compare October 2, 2026 22:21

1996fanrui marked this pull request as draft October 2, 2026 22:26

1996fanrui force-pushed the FLINK-40780 branch 2 times, most recently from e6ec054 to 3686690 Compare October 3, 2026 14:49

1996fanrui marked this pull request as ready for review October 3, 2026 14:53

Efrat19 reviewed Oct 15, 2026

View reviewed changes

github-actions Bot added the community-reviewed PR has been reviewed by the community. label Oct 15, 2026

1996fanrui force-pushed the FLINK-40780 branch from 3686690 to ddc3016 Compare October 15, 2026 19:07

1996fanrui commented Oct 15, 2026

View reviewed changes

Efrat19 approved these changes Oct 18, 2026

View reviewed changes

1996fanrui added 2 commits October 21, 2026 10:33

1996fanrui force-pushed the FLINK-40780 branch from 7d780b8 to 2e53679 Compare October 21, 2026 08:39

1996fanrui merged commit 8a369e7 into apache:master Oct 21, 2026
1 check passed

Case	Result
streaming 4 → 4, no rescale	fails: producer parallelism 4 != consumer parallelism 1
batch 4 → 4, no rescale	fails, same error
streaming 2 → 4 (via pipeline.jobvertex-parallelism-overrides)	succeeds, records per consumer subtask {0=500, 1=0, 2=500, 3=0}
streaming 1 → 2 (the case covered by StreamTaskTest)	fails, as expected

	\|\| consumerParallelism == producerParallelism) {
	\|\| consumerParallelism <= producerParallelism) {

Conversation

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

Uh oh!

CI report:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Efrat19 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!

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!

1996fanrui 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!

Efrat19 left a comment

Choose a reason for hiding this comment

Uh oh!

1996fanrui commented Oct 21, 2026

Uh oh!

1996fanrui commented Oct 21, 2026

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

1996fanrui commented Sep 23, 2026 •

edited

Loading

flinkbot commented Sep 23, 2026 •

edited

Loading