[FLINK-21378] Rescale pointwise connection during unaligned checkpoint recovery

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Incomplete
• Affects Version/s: 1.13.0
• Component/s: Runtime / Checkpointing
• Assignee: Arvid Heise
• Reporter: Arvid Heise
• Created: 16/Feb/21 07:21
• Updated: 22/Jun/21 14:06
• Resolved: 07/Apr/21 07:30

Description

FLINK-19801 added support for rescaling of unaligned checkpoints through virtual channels: A mapping of old to new channel infos helped to create a virtual channel that demultiplexes buffers from different original channel over the same physical channel.

The calculation of FLINK-19801, however, assumes that subpartition = channel index, which holds for all fully connected exchanges, but not for point-wise connection. For point-wise connections, there are few channels per subtask and they correspond to one particular subpartition.

A possible approach is to actually use the subpartition information while constructing InflightDataRescalingDescriptor in TaskStateAssignment. Thus, instead of taking subtask index as the channel index, we should take the subpartition as the channel index. The easiest way to implement it is, by translating subtask index to subpartition index and then calculate the channel index from it.

For that, the following changes are needed:

• StateAssignmentOperation attaches the (upstream/downstream) -> subpartition mapping to all assignments of pointwise exchanges. The information can be derived through ExecutionEdge -> IntermediateResultPartition.partitionNumber (note that on execution graph level subpartition is named partition).
• For non-pointwise exchanges, this mapping is the identity function.
• TaskStateAssignment uses this additional lookup to translate subtask mapping to subpartition mappings, which can then be used to calculate the channel indexes both on input and output side.

Issue Links

• Is contained by: FLINK-21936 Rescale pointwise connections for unaligned checkpoints