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 d407c026 [FLINK-40583] Filter operator keys on the remaining runtime
config paths
d407c026 is described below
commit d407c02661ef245c523e2b4fa9e47bd843562e5f
Author: Dennis-Mircea Ciupitu <[email protected]>
AuthorDate: Mon Sep 14 17:57:07 2026 +0200
[FLINK-40583] Filter operator keys on the remaining runtime config paths
---
.../operator/service/AbstractFlinkService.java | 16 +++++++----
.../utils/FlinkRuntimeConfigurationUtils.java | 2 +-
.../operator/service/AbstractFlinkServiceTest.java | 32 ++++++++++++++++++++--
3 files changed, 41 insertions(+), 9 deletions(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
index ad650423..2e5e89bf 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
@@ -21,7 +21,6 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.autoscaler.config.AutoScalerOptions;
import org.apache.flink.autoscaler.utils.JobStatusUtils;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.CheckpointingOptions;
@@ -167,7 +166,6 @@ import static
org.apache.flink.kubernetes.operator.api.status.CommonStatus.MSG_H
import static
org.apache.flink.kubernetes.operator.api.status.CommonStatus.MSG_JOB_FINISHED_OR_CONFIGMAPS_DELETED;
import static
org.apache.flink.kubernetes.operator.api.status.CommonStatus.MSG_MANUAL_RESTORE_REQUIRED;
import static
org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder.FLINK_VERSION;
-import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.K8S_OP_CONF_PREFIX;
import static org.apache.flink.util.ExceptionUtils.findThrowable;
/**
@@ -983,7 +981,13 @@ public abstract class AbstractFlinkService implements
FlinkService {
TimeUnit.SECONDS);
Map<String, String> jmConfig = new HashMap<>();
- configurationInfo.forEach(entry -> jmConfig.put(entry.getKey(),
entry.getValue()));
+ configurationInfo.forEach(
+ entry -> {
+ if
(!FlinkRuntimeConfigurationUtils.isOperatorControlledKey(
+ entry.getKey())) {
+ jmConfig.put(entry.getKey(), entry.getValue());
+ }
+ });
LOG.debug("Fetched {} JobManager configuration entries",
jmConfig.size());
return jmConfig;
}
@@ -1064,7 +1068,8 @@ public abstract class AbstractFlinkService implements
FlinkService {
? RestoreMode.DEFAULT
: null,
conf.get(FLINK_VERSION).isEqualOrNewer(FlinkVersion.v1_17)
- ? configToMapWithVersionDialect(conf,
flinkVersion)
+ ? configToMapWithVersionDialect(
+ removeOperatorConfigs(conf),
flinkVersion)
: null);
LOG.info("Submitting job: {} to session cluster.", jobID);
clusterClient
@@ -1176,8 +1181,7 @@ public abstract class AbstractFlinkService implements
FlinkService {
protected static Configuration removeOperatorConfigs(Configuration config)
{
Configuration newConfig = new Configuration(config);
for (String key : config.keySet()) {
- if (key.startsWith(K8S_OP_CONF_PREFIX)
- ||
key.startsWith(AutoScalerOptions.AUTOSCALER_CONF_PREFIX)) {
+ if (FlinkRuntimeConfigurationUtils.isOperatorControlledKey(key)) {
newConfig.removeKey(key);
}
}
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 e557e109..5e5b7f58 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
@@ -162,7 +162,7 @@ public class FlinkRuntimeConfigurationUtils {
* autoscaler options, including the autoscaler's legacy {@code
kubernetes.operator.} prefixed
* form.
*/
- private static boolean isOperatorControlledKey(String key) {
+ public static boolean isOperatorControlledKey(String key) {
return
key.startsWith(KubernetesOperatorConfigOptions.K8S_OP_CONF_PREFIX)
|| key.startsWith(AutoScalerOptions.AUTOSCALER_CONF_PREFIX);
}
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 a212f265..ac113012 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
@@ -239,9 +239,12 @@ public class AbstractFlinkServiceTest {
flinkService.submitJobToSessionCluster(
job.getMetadata(), job.getSpec(), JobID.generate(),
deployConf, null);
- // Make sure that deploy conf was passed to jar run
+ // Make sure that deploy conf was passed to jar run, without the
operator keys
+
assertTrue(deployConf.contains(KubernetesOperatorConfigOptions.JAR_ARTIFACT_HTTP_HEADER));
if (flinkVersion.isEqualOrNewer(FlinkVersion.v1_17)) {
- assertEquals(deployConf.toMap(),
jarRuns.get(0).getFlinkConfiguration().toMap());
+ assertEquals(
+
AbstractFlinkService.removeOperatorConfigs(deployConf).toMap(),
+ jarRuns.get(0).getFlinkConfiguration().toMap());
} else {
assertTrue(jarRuns.get(0).getFlinkConfiguration().toMap().isEmpty());
}
@@ -1620,6 +1623,31 @@ public class AbstractFlinkServiceTest {
assertEquals("ZOOKEEPER", result.get("high-availability"));
}
+ @Test
+ void testGetRuntimeConfigurationDropsOperatorControlledJmConfig() throws
Exception {
+ var jmConfig = new ConfigurationInfo();
+ jmConfig.add(new ConfigurationInfoEntry("jobmanager.scheduler",
"Adaptive"));
+ jmConfig.add(
+ new ConfigurationInfoEntry(
+
"kubernetes.operator.job.upgrade.last-state-fallback.enabled", "false"));
+ jmConfig.add(new ConfigurationInfoEntry("job.autoscaler.enabled",
"false"));
+
+ var service =
+ getTestingService(
+ (headers, params, body) -> {
+ if (headers instanceof
JobManagerJobConfigurationHeaders) {
+ return
CompletableFuture.completedFuture((ResponseBody) jmConfig);
+ }
+ return CompletableFuture.failedFuture(
+ new UnsupportedOperationException());
+ });
+
+ var result = service.getRuntimeConfiguration(configuration,
JobID.generate());
+
+ assertThat(result)
+
.containsExactlyInAnyOrderEntriesOf(Map.of("jobmanager.scheduler", "Adaptive"));
+ }
+
@Test
void testGetRuntimeConfigurationIncludesVersionIndependentJobConfig()
throws Exception {
var executionConfig =
