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 776fdd74634f10b5301c468f8dc623a96ebc732c Author: Gyula Fora <[email protected]> AuthorDate: Thu Sep 17 17:24:54 2026 +0200 Revert "[FLINK-40583] Filter operator keys on the remaining runtime config paths" This reverts commit d407c02661ef245c523e2b4fa9e47bd843562e5f. --- .../operator/service/AbstractFlinkService.java | 16 ++++------- .../utils/FlinkRuntimeConfigurationUtils.java | 2 +- .../operator/service/AbstractFlinkServiceTest.java | 32 ++-------------------- 3 files changed, 9 insertions(+), 41 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 2e5e89bf2..ad6504236 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,6 +21,7 @@ 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; @@ -166,6 +167,7 @@ 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; /** @@ -981,13 +983,7 @@ public abstract class AbstractFlinkService implements FlinkService { TimeUnit.SECONDS); Map<String, String> jmConfig = new HashMap<>(); - configurationInfo.forEach( - entry -> { - if (!FlinkRuntimeConfigurationUtils.isOperatorControlledKey( - entry.getKey())) { - jmConfig.put(entry.getKey(), entry.getValue()); - } - }); + configurationInfo.forEach(entry -> jmConfig.put(entry.getKey(), entry.getValue())); LOG.debug("Fetched {} JobManager configuration entries", jmConfig.size()); return jmConfig; } @@ -1068,8 +1064,7 @@ public abstract class AbstractFlinkService implements FlinkService { ? RestoreMode.DEFAULT : null, conf.get(FLINK_VERSION).isEqualOrNewer(FlinkVersion.v1_17) - ? configToMapWithVersionDialect( - removeOperatorConfigs(conf), flinkVersion) + ? configToMapWithVersionDialect(conf, flinkVersion) : null); LOG.info("Submitting job: {} to session cluster.", jobID); clusterClient @@ -1181,7 +1176,8 @@ public abstract class AbstractFlinkService implements FlinkService { protected static Configuration removeOperatorConfigs(Configuration config) { Configuration newConfig = new Configuration(config); for (String key : config.keySet()) { - if (FlinkRuntimeConfigurationUtils.isOperatorControlledKey(key)) { + if (key.startsWith(K8S_OP_CONF_PREFIX) + || key.startsWith(AutoScalerOptions.AUTOSCALER_CONF_PREFIX)) { 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 5e5b7f588..e557e109c 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. */ - public static boolean isOperatorControlledKey(String key) { + private 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 ac1130121..a212f2658 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,12 +239,9 @@ public class AbstractFlinkServiceTest { flinkService.submitJobToSessionCluster( job.getMetadata(), job.getSpec(), JobID.generate(), deployConf, null); - // Make sure that deploy conf was passed to jar run, without the operator keys - assertTrue(deployConf.contains(KubernetesOperatorConfigOptions.JAR_ARTIFACT_HTTP_HEADER)); + // Make sure that deploy conf was passed to jar run if (flinkVersion.isEqualOrNewer(FlinkVersion.v1_17)) { - assertEquals( - AbstractFlinkService.removeOperatorConfigs(deployConf).toMap(), - jarRuns.get(0).getFlinkConfiguration().toMap()); + assertEquals(deployConf.toMap(), jarRuns.get(0).getFlinkConfiguration().toMap()); } else { assertTrue(jarRuns.get(0).getFlinkConfiguration().toMap().isEmpty()); } @@ -1623,31 +1620,6 @@ 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 =
