[FLINK-33743] Support consuming multiple subpartitions on a single channel

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Done
• Component/s: Runtime / Network
• Labels: pull-request-available
• Assignee: Yunfeng Zhou
• Reporter: Yuxin Tan
• Created: 05/Dec/23 05:12
• Updated: 29/Jan/24 14:39
• Resolved: 17/Jan/24 03:35

Description

In Flink jobs that use the AdaptiveBatchScheduler and enable adaptive parallelism, a downstream operator might consume multiple subpartitions from an upstream operator. While downstream operators would create an InputChannel for each upstream subpartition in Flink's current implementation, The many InputChannels created in this situation may consume more memory resources than needed, affecting the usability of Hybrid Shuffle and AdaptiveBatchScheduler. In order to solve this problem, we plan to allow one InputChannel to consume multiple subpartitions.

Issue Links

• causes: FLINK-34225 HybridShuffleITCase.testHybridSelectiveExchangesRestart failed due to NullPointerException
• causes: FLINK-34233 HybridShuffleITCase.testHybridSelectiveExchangesRestart failed due to a IllegalStateException
• links to: GitHub Pull Request #23927