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)
