This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 5f1780e4 [FLINK-40380] Document the runtime configuration and stop
jobs from overriding operator config through it (#1178)
5f1780e4 is described below
commit 5f1780e40dd361b9ce1dfa8a6b6f4f2f009f7200
Author: Dennis-Mircea Ciupitu <[email protected]>
AuthorDate: Thu Aug 13 13:49:32 2026 +0300
[FLINK-40380] Document the runtime configuration and stop jobs from
overriding operator config through it (#1178)
---
docs/content.zh/docs/internals/startup.md | 13 ++++++--
docs/content/docs/internals/startup.md | 13 ++++++--
.../utils/FlinkRuntimeConfigurationUtils.java | 24 ++++++++++++++-
.../operator/service/AbstractFlinkServiceTest.java | 36 ++++++++++++++++++++++
4 files changed, 79 insertions(+), 7 deletions(-)
diff --git a/docs/content.zh/docs/internals/startup.md
b/docs/content.zh/docs/internals/startup.md
index 2f5f1b1b..1b663dd8 100644
--- a/docs/content.zh/docs/internals/startup.md
+++ b/docs/content.zh/docs/internals/startup.md
@@ -170,16 +170,23 @@ The process installs a shutdown hook honoring
`kubernetes.operator.termination.t
### Configuration
-The configuration manager loads the operator's settings at startup and derives
every Flink-facing configuration from them. Three configurations matter
throughout the operator:
+The configuration manager loads the operator's settings at startup and derives
every Flink-facing configuration from them. Four configurations matter
throughout the operator:
| Config | Derived From
| Used For
|
|---------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------|
| Default | The `flink-operator-config` ConfigMap, refined by per-namespace
overrides
(`kubernetes.operator.default-configuration.namespace.<namespace>.<key>`) and
per-Flink-version overrides
(`kubernetes.operator.default-configuration.flink-version.<version>.<key>`) |
The operator's own behavior, and the base of everything below
|
| Deploy | The namespace and Flink version defaults, always merged with the
current spec's `flinkConfiguration`, plus operator-managed additions such as
the resource generation annotation
| Submitting and upgrading the Flink cluster
|
-| Observe | The same recipe applied to the last reconciled spec and its
`flinkConfiguration`, with a session job's own `flinkConfiguration` layered on
top of its cluster's
| Talking to the running cluster, which
must be addressed with the configuration it was actually deployed with |
+| Observe | The same recipe applied to the last reconciled spec and its
`flinkConfiguration`, with a session job's own `flinkConfiguration` layered on
top of its cluster's, and the runtime configuration layered on top of that
| Talking to the running cluster, which
must be addressed with the configuration it was actually deployed with |
+| Runtime | The running job itself, read through the JobManager configuration,
job execution and checkpoint config REST endpoints
| Correcting the observed view wherever the job's
effective settings differ from the spec |
Deploy and observe are the same derivation applied to different specs: deploy
reads the spec being rolled out, observe reads the spec that last reached the
cluster. The two differ exactly while an upgrade is in flight, and converge
again once the new spec lands.
+The runtime configuration exists because a spec is a request, not a record of
what the job ended up running with. A job's main method can change settings
programmatically, and those take precedence over anything the operator
submitted. The job status observer therefore reads them back from the cluster
and layers them over the observed configuration, so that decisions such as
whether checkpointing is enabled are made against the job's real settings. Its
lifecycle differs from the derived [...]
+
+- It is fetched once per job, and skipped for jobs in a globally terminal
state, whose REST endpoints are already gone.
+- It is cached per resource and job id, under the same cache size and timeout
limits as the derived configurations. A new job id, which every redeploy
produces, naturally invalidates it.
+- A failed fetch is logged and retried on the next cycle, and the observed
configuration stays purely spec-derived until one succeeds.
+
Three runtime behaviors round out the configuration machinery:
- Derived configurations are cached per resource and spec, at most
`kubernetes.operator.config.cache.size` entries (default `1000`), expiring
after `kubernetes.operator.config.cache.timeout` (default 10 minutes).
@@ -187,7 +194,7 @@ Three runtime behaviors round out the configuration
machinery:
- The watched namespaces come from the same configuration, and with
`kubernetes.operator.dynamic.namespaces.enabled`, namespace changes adjust the
controllers' informers at runtime.
{{< hint warning >}}
-None of these is the running configuration. The operator derives its view from
the spec and its own defaults, while the cluster can pick up settings the
operator never sees: a `config.yaml` baked into the image, environment
overrides, or properties the job sets programmatically. The configuration a
pipeline actually runs with can therefore differ from everything the operator
tracks.
+The runtime configuration covers a mapped subset of the job's settings, mainly
the default parallelism, object reuse, the job's global parameters, and the
checkpointing, state backend and changelog options, and only for as long as the
job is running. Whatever falls outside that subset stays invisible to the
operator, including a `config.yaml` baked into the image and environment
overrides. The configuration a pipeline actually runs with can therefore still
differ from what the operator tracks.
{{< /hint >}}
### Services
diff --git a/docs/content/docs/internals/startup.md
b/docs/content/docs/internals/startup.md
index 2f5f1b1b..1b663dd8 100644
--- a/docs/content/docs/internals/startup.md
+++ b/docs/content/docs/internals/startup.md
@@ -170,16 +170,23 @@ The process installs a shutdown hook honoring
`kubernetes.operator.termination.t
### Configuration
-The configuration manager loads the operator's settings at startup and derives
every Flink-facing configuration from them. Three configurations matter
throughout the operator:
+The configuration manager loads the operator's settings at startup and derives
every Flink-facing configuration from them. Four configurations matter
throughout the operator:
| Config | Derived From
| Used For
|
|---------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------|
| Default | The `flink-operator-config` ConfigMap, refined by per-namespace
overrides
(`kubernetes.operator.default-configuration.namespace.<namespace>.<key>`) and
per-Flink-version overrides
(`kubernetes.operator.default-configuration.flink-version.<version>.<key>`) |
The operator's own behavior, and the base of everything below
|
| Deploy | The namespace and Flink version defaults, always merged with the
current spec's `flinkConfiguration`, plus operator-managed additions such as
the resource generation annotation
| Submitting and upgrading the Flink cluster
|
-| Observe | The same recipe applied to the last reconciled spec and its
`flinkConfiguration`, with a session job's own `flinkConfiguration` layered on
top of its cluster's
| Talking to the running cluster, which
must be addressed with the configuration it was actually deployed with |
+| Observe | The same recipe applied to the last reconciled spec and its
`flinkConfiguration`, with a session job's own `flinkConfiguration` layered on
top of its cluster's, and the runtime configuration layered on top of that
| Talking to the running cluster, which
must be addressed with the configuration it was actually deployed with |
+| Runtime | The running job itself, read through the JobManager configuration,
job execution and checkpoint config REST endpoints
| Correcting the observed view wherever the job's
effective settings differ from the spec |
Deploy and observe are the same derivation applied to different specs: deploy
reads the spec being rolled out, observe reads the spec that last reached the
cluster. The two differ exactly while an upgrade is in flight, and converge
again once the new spec lands.
+The runtime configuration exists because a spec is a request, not a record of
what the job ended up running with. A job's main method can change settings
programmatically, and those take precedence over anything the operator
submitted. The job status observer therefore reads them back from the cluster
and layers them over the observed configuration, so that decisions such as
whether checkpointing is enabled are made against the job's real settings. Its
lifecycle differs from the derived [...]
+
+- It is fetched once per job, and skipped for jobs in a globally terminal
state, whose REST endpoints are already gone.
+- It is cached per resource and job id, under the same cache size and timeout
limits as the derived configurations. A new job id, which every redeploy
produces, naturally invalidates it.
+- A failed fetch is logged and retried on the next cycle, and the observed
configuration stays purely spec-derived until one succeeds.
+
Three runtime behaviors round out the configuration machinery:
- Derived configurations are cached per resource and spec, at most
`kubernetes.operator.config.cache.size` entries (default `1000`), expiring
after `kubernetes.operator.config.cache.timeout` (default 10 minutes).
@@ -187,7 +194,7 @@ Three runtime behaviors round out the configuration
machinery:
- The watched namespaces come from the same configuration, and with
`kubernetes.operator.dynamic.namespaces.enabled`, namespace changes adjust the
controllers' informers at runtime.
{{< hint warning >}}
-None of these is the running configuration. The operator derives its view from
the spec and its own defaults, while the cluster can pick up settings the
operator never sees: a `config.yaml` baked into the image, environment
overrides, or properties the job sets programmatically. The configuration a
pipeline actually runs with can therefore differ from everything the operator
tracks.
+The runtime configuration covers a mapped subset of the job's settings, mainly
the default parallelism, object reuse, the job's global parameters, and the
checkpointing, state backend and changelog options, and only for as long as the
job is running. Whatever falls outside that subset stays invisible to the
operator, including a `config.yaml` baked into the image and environment
overrides. The configuration a pipeline actually runs with can therefore still
differ from what the operator tracks.
{{< /hint >}}
### Services
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkRuntimeConfigurationUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkRuntimeConfigurationUtils.java
index eb464224..e557e109 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkRuntimeConfigurationUtils.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkRuntimeConfigurationUtils.java
@@ -17,11 +17,13 @@
package org.apache.flink.kubernetes.operator.utils;
+import org.apache.flink.autoscaler.config.AutoScalerOptions;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.configuration.StateBackendOptions;
import org.apache.flink.configuration.StateChangelogOptions;
+import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import org.apache.flink.runtime.rest.messages.JobConfigInfo;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointConfigInfo;
@@ -131,6 +133,10 @@ public class FlinkRuntimeConfigurationUtils {
/**
* Extract execution configuration (parallelism, object-reuse, global job
parameters) from a
* {@link JobConfigInfo} REST response.
+ *
+ * <p>Global job parameters are set by the job itself and are merged over
the observed
+ * configuration, so keys belonging to the operator's own namespaces are
dropped. Otherwise a
+ * job could change how the operator manages it simply by declaring a
matching global parameter.
*/
public static Map<String, String> mapJobConfiguration(JobConfigInfo
configurationInfo) {
Map<String, String> jobConfig = new HashMap<>();
@@ -141,10 +147,26 @@ public class FlinkRuntimeConfigurationUtils {
jobConfig.put(
CoreOptions.DEFAULT_PARALLELISM.key(),
String.valueOf(execInfo.getParallelism()));
jobConfig.put(PipelineOptions.OBJECT_REUSE.key(),
String.valueOf(execInfo.isObjectReuse()));
- jobConfig.putAll(execInfo.getGlobalJobParameters());
+ execInfo.getGlobalJobParameters()
+ .forEach(
+ (key, value) -> {
+ if (!isOperatorControlledKey(key)) {
+ jobConfig.put(key, value);
+ }
+ });
return jobConfig;
}
+ /**
+ * Whether the key belongs to a namespace owned by the operator, covering
both operator and
+ * autoscaler options, including the autoscaler's legacy {@code
kubernetes.operator.} prefixed
+ * form.
+ */
+ private static boolean isOperatorControlledKey(String key) {
+ return
key.startsWith(KubernetesOperatorConfigOptions.K8S_OP_CONF_PREFIX)
+ || key.startsWith(AutoScalerOptions.AUTOSCALER_CONF_PREFIX);
+ }
+
/**
* Convert a {@link CheckpointConfigInfo} REST response into Flink
configuration key-value
* pairs.
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
index bb00337c..8735ca7f 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
@@ -1639,6 +1639,42 @@ public class AbstractFlinkServiceTest {
assertEquals(3, result.size());
}
+ @Test
+ void testMapJobConfigurationDropsOperatorControlledGlobalParameters() {
+ JobConfigInfo configInfo =
+ new JobConfigInfo(
+ new JobID(),
+ "test-job",
+ new JobConfigInfo.ExecutionConfigInfo(
+ "PIPELINED",
+ "fixedDelay",
+ 4,
+ true,
+ Map.of(
+ "user.param",
+ "value1",
+
"kubernetes.operator.job.upgrade.last-state-fallback.enabled",
+ "false",
+ "job.autoscaler.enabled",
+ "false")));
+
+ Map<String, String> result =
FlinkRuntimeConfigurationUtils.mapJobConfiguration(configInfo);
+
+ // A job must not be able to change how the operator manages it.
+ assertNull(
+
result.get("kubernetes.operator.job.upgrade.last-state-fallback.enabled"),
+ "operator keys must not be taken from global job parameters");
+ assertNull(
+ result.get("job.autoscaler.enabled"),
+ "autoscaler keys must not be taken from global job
parameters");
+
+ // Unrelated parameters and the mapped execution fields are still
present.
+ assertEquals("value1", result.get("user.param"));
+ assertEquals("4", result.get("parallelism.default"));
+ assertEquals("true", result.get("pipeline.object-reuse"));
+ assertEquals(3, result.size());
+ }
+
@Test
void testMapJobConfigurationHandlesNullGracefully() {
assertTrue(FlinkRuntimeConfigurationUtils.mapJobConfiguration(null).isEmpty());
