Repository navigation
[BP-2.3][FLINK-40904][FLINK-40780][runtime] Detect FORWARD parallelism mismatch correctly and make the downgrade configurable - #29548
Open
1996fanrui wants to merge 4 commits into
Open
1996fanrui wants to merge 4 commits into
1996fanrui wants to merge 4 commits into
Conversation
…artitionDeploymentDescriptor A task can only see the number of subpartitions of its outputs, which is not the consumer parallelism on a POINTWISE edge. Pass the parallelism of the consuming job vertices through ResultPartitionDeploymentDescriptor and expose it via Environment#getWriterConsumerParallelism. No behavior change. (cherry picked from commit 56dfbeb)
…ctual consumer parallelism The FORWARD-to-REBALANCE downgrade compared the producer parallelism with the number of subpartitions, which is not the consumer parallelism on a POINTWISE edge. Equal parallelism > 1 was taken as a mismatch, and N -> M with N > 1 could go undetected. Compare with the actual consumer parallelism instead and include both values in the log. (cherry picked from commit 96f5580)
…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. (cherry picked from commit 596c373)
…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. (cherry picked from commit 8a369e7)
Collaborator
Efrat19
approved these changes
Oct 22, 2026
github-actions
Bot
added
the
community-reviewed
MartijnVisser
left a comment
MartijnVisser
left a comment
Contributor
There was a problem hiding this comment.
Thanks, I have one comment.
| final int producerParallelism = environment.getTaskInfo().getNumberOfParallelSubtasks(); | ||
| // The number of subpartitions isn't the consumer parallelism on a POINTWISE edge, so the | ||
| // actual consumer parallelism is passed through the deployment descriptor. | ||
| final int consumerParallelism = environment.getWriterConsumerParallelism(outputIndex); |
Contributor
There was a problem hiding this comment.
I think this turns a forward edge overridden from 2 to 4 into a real rebalance by default; before, only consumers 0 and 2 got data. Is that intended for a patch release?
Member
Author
There was a problem hiding this comment.
Thanks for the reveiew.
Yes, it's intended. Before this fix:
- AdaptiveScheduler: always rebalances on a mismatch.
- DefaultScheduler: inconsistent. 1→M became a rebalance, but 2→4 didn't, because the check used the number of subpartitions, which isn't the consumer parallelism on POINTWISE edges.
Now both schedulers rebalance, as FLINK-30213 intended. Users who need the old behavior can set KEEP_FORWARD.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Backporting #29374 and #29272 to 2.3