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 {