[FLINK-30189] HsSubpartitionFileReader may load data that has been consumed from memory

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.16.0, 1.17.0
• Fix Version/s: 1.17.0, 1.16.1
• Component/s: Runtime / Network
• Labels: pull-request-available
• Assignee: Weijie Guo
• Reporter: Weijie Guo
• Created: 24/Nov/22 09:07
• Updated: 07/Dec/22 04:39
• Resolved: 07/Dec/22 04:39

Description

In order to solve the problem that data cannot be read from the disk correctly after failover, we changed the calculation logical of the buffer's readable state in FLINK-29238.  Buffers that are greater than consumingOffset and have been released can be pre-load from file. However, the update of consumingOffset is asynchronous, If it lags behind the actual consumption progress, the buffer will have a chance to be load from the disk again. 

IMO, we can record the consumed status of buffer by each consumer in the InternalRegion. Only the buffers that have not been consumed and have been released will be considered as readable. In the case of failover, a new consumerId will be generated, so all buffers will be considered as unconsumed and can be correctly read from the disk too.

Issue Links

• causes: FLINK-29419 HybridShuffleITCase.testHybridFullExchangesRestart hangs
• links to: GitHub Pull Request #21415