[FLINK-19801] Add support for virtual channels

• Type: Sub-task
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.12.0
• Fix Version/s: 1.13.0
• Component/s: Runtime / Checkpointing
• Labels: pull-request-available
• Assignee: Arvid Heise
• Reporter: Arvid Heise
• Created: 25/Oct/20 18:46
• Updated: 23/Jul/21 15:00
• Resolved: 11/Mar/21 18:47

Description

During rescaling of unaligned checkpoints, if state from multiple former channels are read on input or output side to recover a specific channel, then these buffers are multiplexed on output side and demultiplexed on input side to guarantee a consistent recovery of spanning records:

Assume two channels C1, C2 connect operator A and B and both have one buffer in the output and in the input part of the channel respectively, where a record spans. Assume that the buffers are named O1 for output buffer of C1 and I2 for input buffer of C2 etc. Then after rescaling both channels become one channel C. Then, the buffers may be restored as I1, I2, O1, O2.

Channels use the mapping of FLINK-19533 to infer the need for virtual channels and distribute the needed resources. Virtual channels are removed on the EndOfChannelRecovery epoch marker.

Issue Links

• causes: FLINK-23474 InputStatus should not contain END_OF_RECOVERY
• causes: FLINK-21797 Performance regression on 03/11/21
• links to: GitHub Pull Request #13845