[FLINK-12070] Make blocking result partitions consumable multiple times

• Type: Improvement
• Status: Closed
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 1.9.0
• Fix Version/s: 1.9.0
• Component/s: Runtime / Network
• Labels: pull-request-available
• Assignee: Stephan Ewen
• Reporter: Till Rohrmann
• Created: 29/Mar/19 16:18
• Updated: 26/Nov/19 13:43
• Resolved: 29/Jun/19 19:35

Description

In order to avoid writing produced results multiple times for multiple consumers and in order to speed up batch recoveries, we should make the blocking result partitions to be consumable multiple times. At the moment a blocking result partition will be released once the consumers has processed all data. Instead the result partition should be released once the next blocking result has been produced and all consumers of a blocking result partition have terminated. Moreover, blocking results should not hold on slot resources like network buffers or memory as it is currently the case with SpillableSubpartitions.

Issue Links

• causes: FLINK-12674 BoundedBlockingSubpartition#unsynchronizedGetNumberOfQueuedBuffers is not implemented
• causes: FLINK-14952 Yarn containers can exceed physical memory limits when using BoundedBlockingSubpartition.
• causes: FLINK-13039 Clean up thrown exception lists of `ResultSubpartitionView#getNextBuffer`
• is related to: FLINK-11309 Make SpillableSubpartition repeatably read to enable
• relates to: FLINK-12069 Add proper lifecycle management for intermediate result partitions
• supercedes: FLINK-12329 Netty thread deadlock bug of the SpilledSubpartitionView
• links to: GitHub Pull Request #8105
• links to: GitHub Pull Request #8290

Sub-Tasks

• Add alternative implementations for BoundedBlockingSubpartition to address stability issues: Closed
• Fix the bug of throwing IOException while FileBufferReader#nextBuffer: Closed