[FLINK-17861] Channel state handles, when inlined, duplicate underlying data

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 1.11.0
• Fix Version/s: 1.11.0
• Component/s: Runtime / Checkpointing, Runtime / Task
• Labels: pull-request-available
• Assignee: Roman Khachatryan
• Reporter: Roman Khachatryan
• Created: 21/May/20 08:17
• Updated: 28/May/20 06:15
• Resolved: 28/May/20 06:15

Description

If Unaligned checkpoints are enabled, channel state is written as state handles. Each channel has a handle and each such handle references the same underlying streamStateHandle (this is done to have a single file per subtask).
But, if the state is less then state.backend.fs.memory-threshold, the data is sent directly to JM as a byteStreamHandle. This causes each channel state handle to hold the whole subtask state.

This PR solves this by extracting relevant potions of the underlying handles if they are {{byteStreamHandle}}s.

Issue Links

• links to: GitHub Pull Request #12292