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
commit c1ca49edde4a877afc1f2ec059fa58d028c30d25 Author: Gyula Fora <[email protected]> AuthorDate: Thu Sep 17 17:24:54 2026 +0200 Revert "[FLINK-40380] Document the runtime configuration and stop jobs from overriding operator config through it (#1178)" This reverts commit 5f1780e40dd361b9ce1dfa8a6b6f4f2f009f7200. --- 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, 7 insertions(+), 79 deletions(-) diff --git a/docs/content.zh/docs/internals/startup.md b/docs/content.zh/docs/internals/startup.md index 1b663dd82..2f5f1b1b4 100644 --- a/docs/content.zh/docs/internals/startup.md +++ b/docs/content.zh/docs/internals/startup.md @@ -170,23 +170,16 @@ 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. Four configurations matter throughout the operator: +The configuration manager loads the operator's settings at startup and derives every Flink-facing configuration from them. Three 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, 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 | +| 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 | 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). @@ -194,7 +187,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 >}} -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. +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. {{< /hint >}} ### Services diff --git a/docs/content/docs/internals/startup.md b/docs/content/docs/internals/startup.md index 1b663dd82..2f5f1b1b4 100644 --- a/docs/content/docs/internals/startup.md +++ b/docs/content/docs/internals/startup.md @@ -170,23 +170,16 @@ 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. Four configurations matter throughout the operator: +The configuration manager loads the operator's settings at startup and derives every Flink-facing configuration from them. Three 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, 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 | +| 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 | 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). @@ -194,7 +187,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 >}} -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. +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. {{< /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 e557e109c..eb4642247 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,13 +17,11 @@ 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; @@ -133,10 +131,6 @@ 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<>(); @@ -147,26 +141,10 @@ public class FlinkRuntimeConfigurationUtils { jobConfig.put( CoreOptions.DEFAULT_PARALLELISM.key(), String.valueOf(execInfo.getParallelism())); jobConfig.put(PipelineOptions.OBJECT_REUSE.key(), String.valueOf(execInfo.isObjectReuse())); - execInfo.getGlobalJobParameters() - .forEach( - (key, value) -> { - if (!isOperatorControlledKey(key)) { - jobConfig.put(key, value); - } - }); + jobConfig.putAll(execInfo.getGlobalJobParameters()); 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 8735ca7f6..bb00337c6 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,42 +1639,6 @@ 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());
