[jira] [Created] (FLINK-40992) Allow spec.job.parallelism to take effect when spec.taskManager.replicas is set, to support standby TaskManagers in standalone mode

["Tomoyuki NAKAMURA (Jira)" <ji...@apache.org>]

Tomoyuki NAKAMURA created FLINK-40992:
-----------------------------------------

             Summary: Allow spec.job.parallelism to take effect when spec.taskManager.replicas is set, to support standby TaskManagers in standalone mode
                 Key: FLINK-40992
                 URL: https://issues.apache.org/jira/browse/FLINK-40992
             Project: Flink
          Issue Type: Improvement
          Components: Kubernetes Operator
            Reporter: Tomoyuki NAKAMURA


When {{spec.taskManager.replicas}} is set, {{FlinkConfigBuilder#getParallelism()}} ignores {{spec.job.parallelism}} and uses {{replicas * taskmanager.numberOfTaskSlots}} instead ("Job parallelism setting is ignored as TaskManager replicas are set"). {{DefaultValidator}} only requires {{spec.job.parallelism}} when replicas is not set, so the design is "replicas or parallelism; if both are set, replicas wins". In standalone mode, where the TaskManager count is fixed, this keeps a job from asking for more slots than exist.

It also makes it impossible to run spare (standby) TaskManagers in standalone mode: the job always occupies every slot, so when a TaskManager pod is lost (node failure, eviction, preemption), the job cannot restart until a replacement pod is scheduled and registered. With spare slots it could fail over onto them right away.

Native mode covers this with {{slotmanager.redundant-taskmanager-num}}, but that option has no effect in standalone mode (see the discussion in FLINK-32255). Reactive mode is not an alternative either, since it always uses every available slot.

Standalone mode is not only used for reactive mode: it runs TaskManagers as a Kubernetes Deployment, which tools that target Deployments (e.g. vertical autoscalers) can work with, and the JobManager does not need permissions to create pods. Moving such deployments to native mode just to get spare TaskManagers would give that up.

This was proposed before in FLINK-32883, which was closed as Won't Fix without an answer to the question raised there: why the operator aligns the job parallelism with the total number of slots.

*Proposal:* a boolean dynamic operator option, e.g. {{kubernetes.operator.job.parallelism.derive-from-replicas}} (name open), default {{true}} so the current behavior is unchanged.
* When it is {{false}} and {{spec.job.parallelism}} > 0, the operator uses {{spec.job.parallelism}} even if {{spec.taskManager.replicas}} is set, and keeps the TaskManager replicas as specified. The remaining slots stay free, e.g. as standby capacity. With parallelism 0 it keeps deriving {{replicas * taskmanager.numberOfTaskSlots}} as today.
* In standalone mode, where TaskManagers cannot be added, the validator rejects a parallelism (including the maximum of {{pipeline.jobvertex-parallelism-overrides}}) larger than {{replicas * taskmanager.numberOfTaskSlots}}.
* In native mode, {{spec.taskManager.replicas}} only feeds this derivation, so with the option set to {{false}} it has no effect; the option description would state that.
* The existing warning would point to the new option.

The default protects hand-written specs, where replicas and parallelism can easily disagree. Where FlinkDeployments are generated from a configuration language (e.g. CUE) that derives replicas from parallelism and the number of slots, the values are consistent by construction, and honoring the explicit parallelism is safe.

I can open a PR if this direction looks acceptable.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)