Skip to content

[FLINK-39958] Honor per-deployment Flink REST client timeout on rescale #33

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
pobed2 wants to merge 1 commit into from
Closed

[FLINK-39958] Honor per-deployment Flink REST client timeout on rescale #33

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions ...r/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
32 changes: 27 additions & 5 deletions ...erator/src/main/java/org/apache/flink/kubernetes/operator/service/NativeFlinkService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<JobVertexID, JobVertexResourceRequirements> entry :
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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<String> client,
AbstractFlinkResource<?, ?> resource,
Duration restClientTimeout,
Map<JobVertexID, JobVertexResourceRequirements> newReqs)
throws Exception {
var jobParameters = new JobMessageParameters();
Expand All @@ -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<JobVertexID, JobVertexResourceRequirements> getVertexResources(
RestClusterClient<String> client, AbstractFlinkResource<?, ?> resource)
RestClusterClient<String> client,
AbstractFlinkResource<?, ?> resource,
Duration restClientTimeout)
throws Exception {
var jobParameters = new JobMessageParameters();
jobParameters.jobPathParameter.resolve(
Expand All @@ -298,7 +320,7 @@ protected Map<JobVertexID, JobVertexResourceRequirements> getVertexResources(
new JobResourceRequirementsHeaders(),
jobParameters,
EmptyRequestBody.getInstance())
.get(operatorConfig.getFlinkClientTimeout().toSeconds(), TimeUnit.SECONDS);
.get(restClientTimeout.toSeconds(), TimeUnit.SECONDS);

return currentRequirements.asJobResourceRequirements().get().getJobVertexParallelisms();
}
Expand Down
40 changes: 40 additions & 0 deletions ...c/test/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContextTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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(
Expand Down
5 changes: 4 additions & 1 deletion ...org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -892,14 +892,17 @@ public void testScaleWithRescaleApi() throws Exception {

@Override
protected Map<JobVertexID, JobVertexResourceRequirements> getVertexResources(
RestClusterClient<String> c, AbstractFlinkResource<?, ?> r) {
RestClusterClient<String> c,
AbstractFlinkResource<?, ?> r,
Duration restClientTimeout) {
return submitted;
}

@Override
protected void updateVertexResources(
RestClusterClient<String> c,
AbstractFlinkResource<?, ?> r,
Duration restClientTimeout,
Map<JobVertexID, JobVertexResourceRequirements> req) {
submitted = req;
rescaleCounter.incrementAndGet();
Expand Down
43 changes: 40 additions & 3 deletions ...or/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -260,14 +261,16 @@ public void testScaling() throws Exception {
@Override
protected Map<JobVertexID, JobVertexResourceRequirements> getVertexResources(
RestClusterClient<String> client,
AbstractFlinkResource<?, ?> resource) {
AbstractFlinkResource<?, ?> resource,
Duration restClientTimeout) {
return current.get();
}

@Override
protected void updateVertexResources(
RestClusterClient<String> client,
AbstractFlinkResource<?, ?> resource,
Duration restClientTimeout,
Map<JobVertexID, JobVertexResourceRequirements> newReqs) {
updated.set(newReqs);
}
Expand Down Expand Up @@ -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 {
Expand Down
Loading