Thank you, Yuepeng Pan, for looking into this long-standing issue! The original proposal looks good to me. I agree with Zhanghao Chen to keep things simple and avoid complexity.
I would not add a configuration option, but if we must add one, then let's only have RESCALE / REBALANCE as option, and default to REBALANCE. Let's not add a "mixed" strategy, it should be clear which strategy gets chosen. In practice, I don't believe this configuration option will ever be set, so I would prefer not to add it at all, but I leave this up to the community. Cheers, Max On Tue, Jan 20, 2026 at 11:59 AM Yuepeng Pan <[email protected]> wrote: > > Bumping this thread. Thanks! > > Best regards, > Yuepeng Pan > > Yuepeng Pan <[email protected]> 于2026年1月17日周六 11:06写道： > > > Thanks to Zhanghao Chen for the feedback. > > > > Please let me sort out the candidate solutions from the discussion history > > to facilitate gathering clearer preferences or feedback: > > > > For JobVertices with Forward edges in streaming jobs with the > > AdaptiveScheduler enabled: > > > > Design 1: > > - When upstream and downstream parallelism are the same, restore the > > partitioning strategy to ForwardPartitioner. > > - When upstream and downstream parallelism differ but have a multiple > > relationship, replace the partitioner with RescalePartitioner. > > - When upstream and downstream parallelism differ and do not have a > > multiple relationship, replace the partitioner with RebalancePartitioner. > > > > Design 2: > > Introduce a new parameter: > > > > - name: > > jobmanager.adaptive-scheduler.jobgraph.mutated-forward-edge.replacement-policy > > - type: enum > > - value options: > > - MIXED: Use the strategy from Design 1 > > - RESCALE: Replace the partitioner with RescalePartitioner when upstream > > and downstream JobVertices have different parallelism > > - REBALANCE: Replace the partitioner with RebalancePartitioner when > > upstream and downstream JobVertices have different parallelism > > - default value: MIXED > > > > Looking forward to feedback about it! > > > > Best regards, > > Yuepeng Pan > > > > > > > > Zhanghao Chen <[email protected]> 于2026年1月15日周四 23:35写道： > > > >> Thanks Yuepeng for the detailed elaboration. The idea makes sense, but > >> I'd prefer adding an explicit option to control the behavior for two > >> reasons: > >> > >> 1. > >> A complex strategy in black box may be confusing for others. > >> 2. > >> The real-world cases can be much more complex, e.g. the source > >> parallelism can be limited by MQ partitions, and maintaining a > >> multiplicative relationship between the parallelism of upstream and > >> downstream vertices can be really costly in some cases, but even under a > >> non-multiplicative relationship, rescale can still easily outperform > >> rebalance in some cases (21-to-25 for example). If we can't make it right > >> under all cases, maybe just keep it simple. > >> > >> Best, > >> Zhanghao Chen > >> ________________________________ > >> From: Yuepeng Pan <[email protected]> > >> Sent: Thursday, January 15, 2026 23:03 > >> To: [email protected] <[email protected]> > >> Subject: Re: [DISCUSS] A design proposal to fix the wrong dynamic > >> replacement of partitioner from FORWARD to REBLANCE for AutoScaler and > >> AdaptiveScheduler > >> > >> Thanks Zhanghao Chen for the comments. > >> > >> As mentioned in the previous emails, we have to take one thing into > >> consideration: > >> the final parallelism configuration depends not only on external > >> adjustments, but also on the actual amount of resources that become > >> available. > >> > >> - In an ideal situation with sufficient resources, the external adjustment > >> strategy determines the final parallelism and partitioning. > >> - When resources are insufficient, the actually available resources may > >> also affect the final parallelism and partitioning. > >> > >> Therefore, based on your proposal, we do not introduce any new parameters. > >> Instead, we only apply the following adjustments to pairs of vertices > >> whose > >> initial partitioning type is ForwardPartitioner: > >> > >> - When the upstream and downstream vertex parallelisms have a multiple > >> relationship (and are not equal), we change the partitioning type to > >> RescalePartitioner. > >> - When the upstream and downstream vertex parallelisms do not have a > >> multiple relationship (and are not equal), we change the partitioning type > >> to RebalancePartitioner. > >> - When the upstream and downstream vertex parallelisms are equal, we > >> change > >> the partitioning type back to ForwardPartitioner. > >> > >> In this way, we can also achieve a decoupling from concrete model-specific > >> strategies. > >> > >> WDYTA ? > >> > >> Best regards, > >> Yuepeng Pan > >> > >> Zhanghao Chen <[email protected]> 于2026年1月15日周四 22:44写道： > >> > >> > I think it should definitely be controlled in the model rather than in > >> the > >> > engine. Maybe we can add an option to control its behavior? > >> > > >> > Best, > >> > Zhanghao Chen > >> > ________________________________ > >> > From: Yuepeng Pan <[email protected]> > >> > Sent: Thursday, January 15, 2026 21:39 > >> > To: [email protected] <[email protected]> > >> > Subject: Re: [DISCUSS] A design proposal to fix the wrong dynamic > >> > replacement of partitioner from FORWARD to REBLANCE for AutoScaler and > >> > AdaptiveScheduler > >> > > >> > Thanks Zhanghao Chen for the response. > >> > > >> > Please let me add some historical context[1]. > >> > > >> > In the previous discussions, there were two alternative replacement > >> > strategies, with the following main characteristics: > >> > - RescalePartitioner: Compared to RebalancePartitioner, it introduces > >> fewer > >> > network connections and less shuffle overhead. > >> > However, it is more prone to load skew and therefore lacks generality. > >> > > >> > - RebalancePartitioner: In theory, it can evenly distribute the load > >> across > >> > downstream tasks and is more general, > >> > but at the cost of increased network connections and shuffle overhead. > >> > > >> > To balance generality and correctness, the community eventually chose > >> the > >> > latter. > >> > > >> > I'd like to apologize for not providing a detailed response earlier to > >> this > >> > suggestion[2](switching to RescalePartitioner and enforcing a > >> > multiplicative relationship between upstream and downstream > >> parallelism). > >> > > >> > If this strategy is implemented on the AutoScaler side, we may consider > >> > whether it can be migrated into the engine. > >> > The reason is that inconsistent parallelism between upstream and > >> downstream > >> > vertices connected by a forward edge is not only caused by AutoScaler > >> > requests, > >> > but can also result from rescaling triggered via the REST API or > >> internal > >> > events such as failover. > >> > Therefore, placing the implementation on the engine side would help > >> ensure > >> > the safety and consistency of this strategy. > >> > > >> > If the cost of moving this strategy into the engine is too high, we > >> could > >> > alternatively propose > >> > a new FLIP to discuss and advance it as a new feature on the AutoScaler > >> > side. > >> > > >> > If the strategy you mentioned is indeed intended to be implemented in > >> the > >> > engine, > >> > I have one question. Consider a job consisting of two JobVertices, A > >> and B: > >> > > >> > A (p = 100) --forward--> B (p = 100) > >> > > >> > After one AutoScaler adjustment, the resulting parallelism proposal is: > >> > > >> > A (p = 60) --rescale--> B (p = 100) > >> > > >> > I assume that, in order to maintain a multiplicative relationship > >> between > >> > the parallelism > >> > of upstream and downstream vertices, there are roughly two possible > >> > directions: > >> > > >> > a) Adjust A from p = 60 to p = 50. In this case, some tasks of vertex A > >> may > >> > become bottlenecks. > >> > b) Adjust B from p = 100 to p = 120. In this case, we may end up > >> reserving > >> > some idle resources, > >> > and the scale-down effect may be less significant. > >> > > >> > Any input is appreciated! > >> > > >> > > >> > [1]https://github.com/apache/flink/pull/21443#discussion_r1042919428 > >> > [2] > >> > > >> > > >> https://issues.apache.org/jira/browse/FLINK-33123?focusedCommentId=17767397&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-17767397 > >> > > >> > Best regards, > >> > Yuepeng Pan > >> > > >> > > >> > > >> > Zhanghao Chen <[email protected]> 于2026年1月15日周四 19:37写道： > >> > > >> > > Thanks Yuepeng for the proposal. Overall LGTM. However, I'm a bit > >> > > concerned about the potential performance impact of changing a forward > >> > edge > >> > > to rebalance. The autoscaler currently assumes a linear performance > >> model > >> > > between the throughput and the parallelism. The edge change can easily > >> > > break this assumption as Rebalance introduces more shuffle and > >> results in > >> > > higher CPU usage and network memory consumption. I suggest > >> considering it > >> > > on the algorithm side as well. > >> > > > >> > > Best, > >> > > Zhanghao Chen > >> > > ________________________________ > >> > > From: Yuepeng Pan <[email protected]> > >> > > Sent: Tuesday, January 13, 2026 23:46 > >> > > To: [email protected] <[email protected]> > >> > > Subject: [DISCUSS] A design proposal to fix the wrong dynamic > >> replacement > >> > > of partitioner from FORWARD to REBLANCE for AutoScaler and > >> > AdaptiveScheduler > >> > > > >> > > Hi community, > >> > > > >> > > I would like to start a discussion around the issue described in > >> > > **FLINK-33123[1]**. > >> > > > >> > > This issue can mainly be broken down into two parts: > >> > > a). > >> > > Assuming that initially two upstream and downstream JobVertices > >> connected > >> > > by a FORWARD edge have the same parallelism, > >> > > due to a rescale operation their parallelism becomes different. > >> > > In this case, the current strategy may produce incorrect results when > >> > > rebuilding the upstream–downstream network partition connections. > >> > > b). > >> > > Assuming that the parallelism of two upstream and downstream > >> JobVertices > >> > is > >> > > different, > >> > > but due to a rescale operation their parallelism needs to be adjusted > >> to > >> > be > >> > > the same. > >> > > In this scenario, it is not possible to determine the partition type > >> > after > >> > > the rescale. > >> > > > >> > > So, I'd like to share a design proposal[2] that attempts to address > >> the > >> > > problem described in the ticket[1]. > >> > > > >> > > Thanks in advance for your time and feedback. > >> > > Looking forward to the discussion! > >> > > > >> > > > >> > > [1]https://issues.apache.org/jira/browse/FLINK-33123 > >> > > [2] > >> > > > >> > > > >> > > >> https://docs.google.com/document/d/1e_6o4bdXcKtFL3xYxKeyKnRjR8ffsw6Z8frp3tp7u-M/edit?usp=sharing > >> > > > >> > > Best regards, > >> > > Yuepeng Pan > >> > > > >> > > >> > >
