[FLINK-14859] Avoid leaking unassigned Slot in DefaultScheduler when Deployment is outdated

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.10.0
• Fix Version/s: 1.10.0
• Component/s: Runtime / Coordination
• Labels: pull-request-available
• Assignee: Gary Yao
• Reporter: Gary Yao
• Created: 19/Nov/19 14:08
• Updated: 28/Nov/19 20:46
• Resolved: 28/Nov/19 20:46

Description

In DefaultScheduler#assignResourceOrHandleError(), if the deployment is outdated, we should release the possibly acquired LogicalSlot so that we do not leak resources.

Below is an example to illustrate how slot leak is currently possible:

1. Vertices A1, A2, A3 are scheduled in a batch.
2. A2 acquires a slot. A1, A3 do not.
3. A1 fails due to slot allocation timeout and triggers failover (DefaultScheduler#cancelTasksAsync)
4. A2 is canceled first and its returned slot is assigned to A3, which triggers DefaultScheduler#assignResourceOrHandleError of A3.
   However, A3 is not canceled yet but it is outdated because executionVertexVersioner#recordVertexModifications was already invoked

Issue Links

• links to: GitHub Pull Request #10255
• links to: GitHub Pull Request #10351