[FLINK-23833] Cache of ShuffleDescriptors should be individually cleaned up

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 1.14.0
• Fix Version/s: 1.14.0, 1.15.0
• Component/s: Runtime / Coordination
• Labels: pull-request-available
• Assignee: Zhilong Hong
• Reporter: Zhilong Hong
• Created: 17/Aug/21 06:59
• Updated: 30/Aug/21 07:50
• Resolved: 30/Aug/21 07:50

Description

In FLINK-23005, we introduce the cache of compressed serialized value for ShuffleDescriptors to improve the performance of deployment. To make sure the cache wouldn't stay too long and become a burden for GC, the cache would be cleaned up when the partition is released or reset for new execution. In the implementation, the cache of the entire IntermediateResult is cleaned up because a partition is released only when the entire IntermediateResult is released.

However, after FLINK-22017, the BLOCKING result partition is allowed to be consumable individually. It also means that the result partition doesn't need to wait for other result partitions and can be released individually. After this change, there may be a scene: when a result partition is finished, the cache of IntermediateResult on the blob is deleted, while other result partitions corresponding to this IntermediateResult is just deployed to the TaskExecutor. Then when TaskExecutors are trying to download TDD from the blob, they will find the blob is deleted and get stuck.

This bug only happens for jobs with POINTWISE BLOCKING edge. Also, the blob.offload.minsize is set to be a extremely small value, since the size of  ShuffleDescriptors of POINTWISE BLOCKING edges is usually small. To solve this issue, we just need to clean up the cache of ShuffleDescriptors individually.

Issue Links

• relates to: FLINK-23005 Cache the compressed serialized value of ShuffleDescriptors
• relates to: FLINK-22017 Regions may never be scheduled when there are cross-region blocking edges
• links to: GitHub Pull Request #16856