[FLINK-36733] Don't transition task to RUNNING until the inputs are recovered (UC)

• Type: Improvement
• Status: In Progress
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 1.20.0, 1.19.1
• Fix Version/s: 1.19.2, 1.20.1
• Component/s: Runtime / Task
• Assignee: Roman Khachatryan
• Reporter: Roman Khachatryan
• Created: 1 week ago 13:40
• Updated: 1 week ago 13:40

Description

When recovering from an Unaligned Checkpoint, a task transitions to RUNNING after restoring:

1. Output channel state
2. Operator state
3. Input channel state 

However, the upstream task(s) might not yet send all the recovered buffers; therefore, in case of rescaling, downstream task must keep the virtual channel infrastructure up (RescalingStreamTaskNetworkInput).

{{}}

That means in particular that checkpoints might be triggered by the `CheckpointCoordinator` but declined by the downstream task (because RescalingStreamTaskNetworkInput doesn't support checkpointing).

 

In case of long recovery, many declined checkpoints might exhaust some resources, e.g. transaction ID pools in our case.

It's confusing (for humans and observability tools) to see tasks switched to RUNNING but still not able to checkpoint due to recovery.

 

The proposal is to transition task to RUNNING only after all the inputs are recovered.