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 2025/03/27 16:58:16 UTC
(flink-kubernetes-operator) branch main updated: [hotfix] Do not leak autoscaler configs to jobs
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 b18bdb47 [hotfix] Do not leak autoscaler configs to jobs
b18bdb47 is described below
commit b18bdb47fd18e6f51b79bc1321f7917d95021c04
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Fri Mar 7 14:40:05 2025 +0100
[hotfix] Do not leak autoscaler configs to jobs
---
.../kubernetes/operator/service/AbstractFlinkService.java | 4 +++-
.../kubernetes/operator/service/AbstractFlinkServiceTest.java | 11 ++++++++---
2 files changed, 11 insertions(+), 4 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 48a70b3a..733ebd3a 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;
@@ -984,7 +985,8 @@ public abstract class AbstractFlinkService implements FlinkService {
config.toMap()
.forEach(
(k, v) -> {
- if (!k.startsWith(K8S_OP_CONF_PREFIX)) {
+ if (!k.startsWith(K8S_OP_CONF_PREFIX)
+ && !k.startsWith(AutoScalerOptions.AUTOSCALER_CONF_PREFIX)) {
newConfig.setString(k, v);
}
});
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 9082ff35..a57d60a8 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
@@ -987,10 +987,15 @@ public class AbstractFlinkServiceTest {
@Test
public void removeOperatorConfigTest() {
- var key = "kubernetes.operator.meyKey";
- var deployConfig = Configuration.fromMap(Map.of("kubernetes.operator.meyKey", "v"));
+ var opKey1 = "kubernetes.operator.meyKey";
+ var opKey2 = "job.autoscaler.";
+ var regularKey = "k";
+ var deployConfig =
+ Configuration.fromMap(Map.of(opKey1, "v", opKey2, "v", regularKey, "v1"));
var newConf = AbstractFlinkService.removeOperatorConfigs(deployConfig);
- assertFalse(newConf.containsKey(key));
+ assertFalse(newConf.containsKey(opKey1));
+ assertFalse(newConf.containsKey(opKey2));
+ assertTrue(newConf.containsKey(regularKey));
}
@Test