From 0d180bd50a1b5dfd29b1ce53df7b32ed78741ad9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pierre-Olivier=20B=C3=A9dard?= Date: Wed, 29 Jul 2026 15:30:12 -0700 Subject: [PATCH] [FLINK-39958] Honor per-deployment Flink REST client timeout on rescale MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Autoscaler in-place rescales intermittently fail with a bare TimeoutException out of NativeFlinkService.updateVertexResources. When that happens the operator abandons in-place scaling with no retry and falls back to a full savepoint-and-redeploy: suspend with savepoint, delete the JobManager deployment, delete HA metadata, redeploy. On the webhooks-production-java-change-event-producer FlinkDeployment that costs ~150s of no checkpoint progress plus a slow-recovery tail, and it fired 7 times in 24h on that pipeline alone (17 times fleet-wide across three pipelines) in place of what is normally a ~3s rescale. The budget for that REST call is kubernetes.operator.flink.client.timeout (default 10s), which is operator-global: it is read from FlinkOperatorConfiguration, built by FlinkConfigManager.getDefaultConfig from the operator's own configuration, so a pipeline owner cannot raise it from their FlinkDeployment spec. The per-deployment-looking job.autoscaler.flink.rest-client.timeout is silently clobbered, which is FLINK-39958. Two changes: - Cherry-pick FLINK-39958 (apache/flink-kubernetes-operator#1141) so FlinkResourceContext only seeds AutoScalerOptions.FLINK_CLIENT_TIMEOUT from the operator timeout when the deployment has not set it. - Apply the same honor-the-per-deployment-value rule to the in-place rescale REST calls in NativeFlinkService. Those read operatorConfig directly and so are untouched by FLINK-39958, which is why the cherry-pick alone would not stop the teardowns. No defaults change. With job.autoscaler.flink.rest-client.timeout unset, both paths resolve to the operator-global timeout exactly as before. Co-Authored-By: Claude Opus 5 (1M context) Co-authored-by: Pierre-Olivier Bédard --- .../controller/FlinkResourceContext.java | 8 ++-- .../operator/service/NativeFlinkService.java | 32 +++++++++++--- .../controller/FlinkResourceContextTest.java | 40 +++++++++++++++++ .../deployment/ApplicationReconcilerTest.java | 5 ++- .../service/NativeFlinkServiceTest.java | 43 +++++++++++++++++-- 5 files changed, 116 insertions(+), 12 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 f62df74f79..c495381d1e 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 @@ -68,9 +68,11 @@ public KubernetesJobAutoScalerContext getJobAutoScalerContext() { private KubernetesJobAutoScalerContext createJobAutoScalerContext() { Configuration conf = new Configuration(getDeployConfig(resource.getSpec())); - 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/main/java/org/apache/flink/kubernetes/operator/service/NativeFlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/NativeFlinkService.java index a5a63f8b4f..6fcbed78c6 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/NativeFlinkService.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/NativeFlinkService.java @@ -20,6 +20,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.autoscaler.config.AutoScalerOptions; import org.apache.flink.client.cli.ApplicationDeployer; import org.apache.flink.client.deployment.ClusterClientFactory; import org.apache.flink.client.deployment.ClusterClientServiceLoader; @@ -187,8 +188,11 @@ public boolean scale(FlinkResourceContext ctx, Configuration deployConfig) th return false; } + var restClientTimeout = getRescaleRestClientTimeout(observeConfig); + try (var client = getClusterClient(observeConfig)) { - var requirements = new HashMap<>(getVertexResources(client, resource)); + var requirements = + new HashMap<>(getVertexResources(client, resource, restClientTimeout)); var alreadyScaled = true; for (Map.Entry entry : @@ -224,7 +228,7 @@ public boolean scale(FlinkResourceContext ctx, Configuration deployConfig) th if (alreadyScaled) { LOG.info("Vertex resources requirements already match target, nothing to do..."); } else { - updateVertexResources(client, resource, requirements); + updateVertexResources(client, resource, restClientTimeout, requirements); eventRecorder.triggerEvent( resource, EventRecorder.Type.Normal, @@ -269,10 +273,26 @@ private static boolean supportsInPlaceScaling( return true; } + /** + * Timeout for the resource requirement REST calls made against the running JobManager during + * in-place scaling. Deployments may raise it through {@link + * AutoScalerOptions#FLINK_CLIENT_TIMEOUT}, otherwise the operator-global {@code + * kubernetes.operator.flink.client.timeout} is used as before. + * + * @param conf Config of the deployment being rescaled. + * @return Timeout to apply to the in-place scaling REST calls. + */ + @VisibleForTesting + protected Duration getRescaleRestClientTimeout(Configuration conf) { + return conf.getOptional(AutoScalerOptions.FLINK_CLIENT_TIMEOUT) + .orElseGet(operatorConfig::getFlinkClientTimeout); + } + @VisibleForTesting protected void updateVertexResources( RestClusterClient client, AbstractFlinkResource resource, + Duration restClientTimeout, Map newReqs) throws Exception { var jobParameters = new JobMessageParameters(); @@ -282,12 +302,14 @@ protected void updateVertexResources( var requestBody = new JobResourceRequirementsBody(new JobResourceRequirements(newReqs)); client.sendRequest(new JobResourcesRequirementsUpdateHeaders(), jobParameters, requestBody) - .get(operatorConfig.getFlinkClientTimeout().toSeconds(), TimeUnit.SECONDS); + .get(restClientTimeout.toSeconds(), TimeUnit.SECONDS); } @VisibleForTesting protected Map getVertexResources( - RestClusterClient client, AbstractFlinkResource resource) + RestClusterClient client, + AbstractFlinkResource resource, + Duration restClientTimeout) throws Exception { var jobParameters = new JobMessageParameters(); jobParameters.jobPathParameter.resolve( @@ -298,7 +320,7 @@ protected Map getVertexResources( new JobResourceRequirementsHeaders(), jobParameters, EmptyRequestBody.getInstance()) - .get(operatorConfig.getFlinkClientTimeout().toSeconds(), TimeUnit.SECONDS); + .get(restClientTimeout.toSeconds(), TimeUnit.SECONDS); return currentRequirements.asJobResourceRequirements().get().getJobVertexParallelisms(); } 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 1f734274a3..3e302ecce3 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 Configuration getObserveConfig(FlinkDeployment deployment) { 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( diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java index a5d87bc946..268dd948b2 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java @@ -892,7 +892,9 @@ public void testScaleWithRescaleApi() throws Exception { @Override protected Map getVertexResources( - RestClusterClient c, AbstractFlinkResource r) { + RestClusterClient c, + AbstractFlinkResource r, + Duration restClientTimeout) { return submitted; } @@ -900,6 +902,7 @@ protected Map getVertexResources( protected void updateVertexResources( RestClusterClient c, AbstractFlinkResource r, + Duration restClientTimeout, Map req) { submitted = req; rescaleCounter.incrementAndGet(); diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java index 666f24da81..d96f4f6456 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java @@ -19,6 +19,7 @@ import org.apache.flink.api.common.JobID; import org.apache.flink.api.common.JobStatus; +import org.apache.flink.autoscaler.config.AutoScalerOptions; import org.apache.flink.client.program.rest.RestClusterClient; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.JobManagerOptions; @@ -260,7 +261,8 @@ public void testScaling() throws Exception { @Override protected Map getVertexResources( RestClusterClient client, - AbstractFlinkResource resource) { + AbstractFlinkResource resource, + Duration restClientTimeout) { return current.get(); } @@ -268,6 +270,7 @@ protected Map getVertexResources( protected void updateVertexResources( RestClusterClient client, AbstractFlinkResource resource, + Duration restClientTimeout, Map newReqs) { updated.set(newReqs); } @@ -552,11 +555,45 @@ public void resourceRestApiTest() throws Exception { var deployment = TestUtils.buildApplicationCluster(); deployment.getStatus().getJobStatus().setJobId(jobId.toString()); + var restClientTimeout = service.getRescaleRestClientTimeout(configuration); assertEquals( reqs.getJobVertexParallelisms(), - service.getVertexResources(testingClusterClient, deployment)); + service.getVertexResources(testingClusterClient, deployment, restClientTimeout)); service.updateVertexResources( - testingClusterClient, deployment, reqs.getJobVertexParallelisms()); + testingClusterClient, + deployment, + restClientTimeout, + reqs.getJobVertexParallelisms()); + } + + @Test + public void testRescaleRestClientTimeoutFallsBackToOperatorTimeout() { + var service = createServiceWithOperatorClientTimeout(Duration.ofSeconds(60)); + + // Key unset -> the operator-global timeout keeps governing the rescale REST calls. + assertEquals( + Duration.ofSeconds(60), service.getRescaleRestClientTimeout(new Configuration())); + } + + @Test + public void testRescaleRestClientTimeoutHonoursDeploymentOverride() { + var service = createServiceWithOperatorClientTimeout(Duration.ofSeconds(60)); + + var deployConf = new Configuration(); + deployConf.set(AutoScalerOptions.FLINK_CLIENT_TIMEOUT, Duration.ofSeconds(30)); + + // Key explicitly set -> the deployment value wins over the operator-global timeout. + assertEquals(Duration.ofSeconds(30), service.getRescaleRestClientTimeout(deployConf)); + } + + private NativeFlinkService createServiceWithOperatorClientTimeout(Duration timeout) { + configuration.set(KubernetesOperatorConfigOptions.OPERATOR_FLINK_CLIENT_TIMEOUT, timeout); + return new NativeFlinkService( + client, + null, + executorService, + FlinkOperatorConfiguration.fromConfiguration(configuration), + eventRecorder); } class TestingNativeFlinkService extends NativeFlinkService {