[FLINK-21080] Handle UnionListState with finished operators

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Implemented
• Component/s: API / DataStream, Runtime / Checkpointing
• Labels: auto-unassigned pull-request-available
• Assignee: Yun Gao
• Reporter: Yun Gao
• Created: 22/Jan/21 01:24
• Updated: 28/Aug/21 13:08
• Resolved: 03/Aug/21 11:23

Description

Most legacy source operators would record the offset for each partitions, and after recovery it would read from the recorded offset. If before a checkpoint some subtasks are finished, the corresponding partition offsets would be deserted in the checkpoint. Then if the job recover with this checkpoint, the legacy source would re-discovery all the partitions and for those finished tasks, the legacy source would re-read them since their offsets are not recorded. 

Therefore, we would like to fail the checkpoint if some legacy source operators have part of subtasks finished. 

Issue Links

• causes: FLINK-23759 notifyCheckpointComplete without corresponding snapshotState
• links to: GitHub Pull Request #16633