[FLINK-31330] Batch shuffle may deadlock for operator with priority input

• Type: Technical Debt
• Status: Closed
• Priority: Major
• Resolution: Not A Problem
• Affects Version/s: 1.16.1
• Component/s: Runtime / Network
• Assignee: Weijie Guo
• Reporter: Weijie Guo
• Created: 06/Mar/23 08:08
• Updated: 12/Mar/23 15:09
• Resolved: 12/Mar/23 15:09

Description

For batch job, some operator's input have priority. For example, hash join operator has two inputs called build and probe respectively. Only after the build input is finished can the probe input start consuming. Unfortunately, the priority of input will not affect multiple inputs to request upstream data(i.e. request partition). In current implementation, when all states are restored, inputGate will start to request partition. This will enable the upstream IO scheduler to register readers for all downstream channels, so there is the possibility of deadlock.
Assume that the build and probe input's upstream tasks of hash join are deployed in the same TM. Then the corresponding readers will be registered to an single IO scheduler, and they share the same BatchShuffleReadBufferPool. If the IO thread happens to load too many buffers for the probe reader, but the downstream will not consume the data, which will cause the build reader to be unable to request enough buffers. Therefore, deadlock occurs.
In fact, we realized this problem at the beginning of the design of SortMergeShuffle, so we introduced a timeout mechanism when requesting read buffers. If this happens, the downstream task will trigger failover to avoid permanent blocking. However, under the default configuration, TPC-DS test with 10T data can easily cause the job to fail because of this reason. It seems that this problem needs to be solved more better.

Issue Links

• is related to: FLINK-31346 Batch shuffle IO scheduler does not throw TimeoutException if numRequestedBuffers is greater than 0