You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2026/09/18 07:08:38 UTC
(flink-kubernetes-operator) 01/04: Revert "[FLINK-40583] Filter operator keys on the remaining runtime config paths"
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 <g_...@apple.com>
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 =