[FLINK-29459] Sink v2 has bugs in supporting legacy v1 implementations with global committer

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 1.16.0, 1.17.0, 1.15.3
• Fix Version/s: 2.0.0
• Component/s: API / DataStream
• Assignee: Yun Gao
• Reporter: Yun Gao
• Created: 28/Sep/22 15:15
• Updated: 10/Jul/24 06:57

Description

Currently when supporting Sink implementation using version 1 interface, there are issues after restoring from a checkpoint after failover:

1. In global committer operator, when restoring SubtaskCommittableManager, the subtask id is replaced with the one in the current operator. This means that the id originally is the id of the sender task (0 ~ N - 1), but after restoring it has to be 0. This would cause Duplication Key exception during restoring.
2. For Committer operator, the subtaskId of CheckpointCommittableManagerImpl is always restored to 0 after failover for all the subtasks. This makes the summary sent to the Global Committer is attached with wrong subtask id.
3. For Committer operator, the checkpoint id of SubtaskCommittableManager is always restored to 1 after failover, this make the following committable sent to the global committer is attached with wrong checkpoint id. 

Issue Links

• is related to: FLINK-29583 Ensure correct subtaskId and checkpointId is set during committer state migration
• is related to: FLINK-29509 Set correct subtaskId during recovery of committables
• is related to: FLINK-29512 Align SubtaskCommittableManager checkpointId with CheckpointCommittableManagerImpl checkpointId during recovery
• is related to: FLINK-29627 Sink - Duplicate key exception during recover more than 1 committable.