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 d971aa3ef [FLINK-39958] Autoscaler Flink REST client timeout is
silently overridden by the operator client timeout (#1141)
d971aa3ef is described below
commit d971aa3ef0dec19cd98361162a252e1ede575ab1
Author: Dennis-Mircea Ciupitu <[email protected]>
AuthorDate: Mon Jun 22 14:48:25 2026 +0300
[FLINK-39958] Autoscaler Flink REST client timeout is silently overridden
by the operator client timeout (#1141)
---
.../operator/controller/FlinkResourceContext.java | 8 +++--
.../controller/FlinkResourceContextTest.java | 40 ++++++++++++++++++++++
2 files changed, 45 insertions(+), 3 deletions(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContext.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContext.java
index 959d6cb1c..15c0242d2 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContext.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContext.java
@@ -72,9 +72,11 @@ public abstract class FlinkResourceContext<CR extends
AbstractFlinkResource<?, ?
if (deployConf != null) {
conf.addAll(deployConf);
}
- conf.set(
- AutoScalerOptions.FLINK_CLIENT_TIMEOUT,
- getOperatorConfig().getFlinkClientTimeout());
+ if (!conf.contains(AutoScalerOptions.FLINK_CLIENT_TIMEOUT)) {
+ conf.set(
+ AutoScalerOptions.FLINK_CLIENT_TIMEOUT,
+ getOperatorConfig().getFlinkClientTimeout());
+ }
CommonStatus<?> status = getResource().getStatus();
String jobId = status.getJobStatus().getJobId();
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContextTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContextTest.java
index 1f734274a..3e302ecce 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContextTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContextTest.java
@@ -17,6 +17,7 @@
package org.apache.flink.kubernetes.operator.controller;
+import org.apache.flink.autoscaler.config.AutoScalerOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.TestingFlinkService;
@@ -150,6 +151,45 @@ public class FlinkResourceContextTest {
assertEquals(2, counter.get());
}
+ @Test
+ void testAutoscalerFlinkClientTimeoutDefaultsToOperatorTimeout() {
+ var conf = new Configuration();
+ conf.set(
+ KubernetesOperatorConfigOptions.OPERATOR_FLINK_CLIENT_TIMEOUT,
+ Duration.ofSeconds(60));
+ configManager = new FlinkConfigManager(conf);
+
+ // No explicit autoscaler client timeout -> falls back to the operator
timeout.
+ var ctx = getContext(TestUtils.buildApplicationCluster());
+ assertEquals(
+ Duration.ofSeconds(60),
+ ctx.getJobAutoScalerContext()
+ .getConfiguration()
+ .get(AutoScalerOptions.FLINK_CLIENT_TIMEOUT));
+ }
+
+ @Test
+ void testExplicitAutoscalerFlinkClientTimeoutIsRespected() {
+ var conf = new Configuration();
+ conf.set(
+ KubernetesOperatorConfigOptions.OPERATOR_FLINK_CLIENT_TIMEOUT,
+ Duration.ofSeconds(60));
+ configManager = new FlinkConfigManager(conf);
+
+ var dep = TestUtils.buildApplicationCluster();
+ dep.getSpec()
+ .getFlinkConfiguration()
+ .put(AutoScalerOptions.FLINK_CLIENT_TIMEOUT.key(), "30 s");
+
+ // An explicit autoscaler timeout must not be silently overridden by
the operator timeout.
+ var ctx = getContext(dep);
+ assertEquals(
+ Duration.ofSeconds(30),
+ ctx.getJobAutoScalerContext()
+ .getConfiguration()
+ .get(AutoScalerOptions.FLINK_CLIENT_TIMEOUT));
+ }
+
FlinkResourceContext getContext(AbstractFlinkResource<?, ?> cr) {
if (cr instanceof FlinkDeployment) {
return new FlinkDeploymentContext(
