[FLINK-30773] Add API for rescaling of jobs based on per-vertex parallelism overrides

• Type: New Feature
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.18.0
• Component/s: Autoscaler, Runtime / Coordination, Runtime / REST
• Assignee: Maximilian Michels
• Reporter: Maximilian Michels
• Created: 23/Jan/23 16:02
• Updated: 28/Apr/23 10:04
• Resolved: 28/Apr/23 10:04

Description

FLINK-29501 introduced a way to rescale jobs via a user-provided parallelism overrides map. This feature is already used today by the Autoscaler of the Flink Kubernetes operator. However, it requires a full restart of the Flink job and only supports the application deployment mode.

In a K8s environment, this is inefficient because all pods for a deployment will be surrendered. Upon restart, they have to be re-acquired. In addition to being slow, this can also lead to situations where resource constraints prevent a restart from executing properly.

Ideally, we would would want the following to happen on receiving a rescale request:

1. Rescale API receives a request with a parallelism overrides map (vertexId => parallelism) for a jobId
2. Compute the number of required task slots using the overrides and the current JobGraph
   1. If the total number of task slots for the cluster is less than the required number of task slots of the rescale, acquire the missing task slots. Otherwise, do nothing
   2. Wait for new task slots to become available
   3. Abort rescale request on timeout
3. Redeploy the JobGraph / Tasks with the updated parallelisms
4. Surrender any unused task slots in case of scaling down

 

There is an existing "Rescaling" API which is currently disabled. We have to evaluate whether reusing it makes sense.

Issue Links

• is fixed by: FLINK-31316 FLIP-291: Externalized Declarative Resource Management
• is related to: FLINK-29501 Allow overriding JobVertex parallelisms during job submission
• is related to: FLINK-30260 FLIP-271: Add initial Autoscaling implementation