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 c6098b0e4 [FLINK-40535] Standalone session clusters are redeployed
instead of scaled in-place
c6098b0e4 is described below
commit c6098b0e48f7bd982c7110c9a8749e35e97be2b8
Author: Dennis-Mircea Ciupitu <[email protected]>
AuthorDate: Tue Sep 15 15:30:51 2026 +0200
[FLINK-40535] Standalone session clusters are redeployed instead of scaled
in-place
---
.../operator/service/StandaloneFlinkService.java | 11 +-
.../service/StandaloneFlinkServiceTest.java | 146 +++++++--------------
2 files changed, 54 insertions(+), 103 deletions(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkService.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkService.java
index ac3dab338..c1888d604 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkService.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkService.java
@@ -45,6 +45,7 @@ import io.fabric8.kubernetes.client.KubernetesClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -166,7 +167,7 @@ public class StandaloneFlinkService extends
AbstractFlinkService {
@Override
public boolean scale(FlinkResourceContext<?> ctx, Configuration
deployConfig) {
var observeConfig = ctx.getObserveConfig();
- var jobSpec = ctx.getResource().getSpec();
+ var jobSpec = ctx.getResource().getSpec().getJob();
var meta = ctx.getResource().getMetadata();
if (observeConfig.get(JobManagerOptions.SCHEDULER_MODE) !=
SchedulerExecutionMode.REACTIVE
&& jobSpec != null) {
@@ -179,17 +180,17 @@ public class StandaloneFlinkService extends
AbstractFlinkService {
var name =
StandaloneKubernetesUtils.getTaskManagerDeploymentName(clusterId);
var deployment =
kubernetesClient.apps().deployments().inNamespace(namespace).withName(name);
-
- if (deployment == null || deployment.get() == null) {
+ var tmDeployment = deployment.get();
+ if (tmDeployment == null) {
LOG.warn("TM Deployment ({}) not found", name);
return false;
}
- var actualReplicas = deployment.get().getSpec().getReplicas();
+ var actualReplicas = tmDeployment.getSpec().getReplicas();
var desiredReplicas =
deployConfig.get(
StandaloneKubernetesConfigOptionsInternal.KUBERNETES_TASKMANAGER_REPLICAS);
- if (actualReplicas != desiredReplicas) {
+ if (!Objects.equals(actualReplicas, desiredReplicas)) {
LOG.info(
"Scaling TM replicas: actual({}) -> desired({})",
actualReplicas,
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java
index e4e245fa9..a20ee83a7 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java
@@ -31,6 +31,7 @@ import
org.apache.flink.kubernetes.operator.artifact.ArtifactManager;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
import org.apache.flink.kubernetes.operator.utils.StandaloneKubernetesUtils;
import org.apache.flink.util.concurrent.Executors;
@@ -49,6 +50,8 @@ import
io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import java.time.Duration;
import java.util.List;
@@ -136,119 +139,49 @@ public class StandaloneFlinkServiceTest {
assertEquals(0, deployments.size());
}
- @Test
- public void testTMReplicaScaleApplication() {
- flinkDeployment.getSpec().setJob(new JobSpec());
+ @ParameterizedTest
+ @CsvSource({"true,true,true", "true,false,false", "false,false,true"})
+ public void testTMReplicaScale(
+ boolean applicationCluster, boolean reactiveMode, boolean
expectScaled) {
var clusterId = flinkDeployment.getMetadata().getName();
var namespace = flinkDeployment.getMetadata().getNamespace();
flinkDeployment.getSpec().setMode(KubernetesDeploymentMode.STANDALONE);
-
- // Add parallelism change, verify it is honoured in reactive mode
- flinkDeployment
- .getSpec()
- .getFlinkConfiguration()
- .put(
- JobManagerOptions.SCHEDULER_MODE.key(),
- SchedulerExecutionMode.REACTIVE.name());
+ if (applicationCluster) {
+ flinkDeployment.getSpec().setJob(new JobSpec());
+ }
+ if (reactiveMode) {
+ flinkDeployment
+ .getSpec()
+ .getFlinkConfiguration()
+ .put(
+ JobManagerOptions.SCHEDULER_MODE.key(),
+ SchedulerExecutionMode.REACTIVE.name());
+ }
flinkDeployment
.getStatus()
.getReconciliationStatus()
.serializeAndSetLastReconciledSpec(flinkDeployment.getSpec(),
flinkDeployment);
- flinkDeployment.getSpec().getTaskManager().setReplicas(5);
- var ctx =
- new FlinkResourceContextFactory(
- configManager,
- TestUtils.createTestMetricGroup(new
Configuration()),
- null)
- .getResourceContext(flinkDeployment,
TestUtils.createEmptyContext());
- assertTrue(
- flinkStandaloneService.scale(ctx,
ctx.getDeployConfig(flinkDeployment.getSpec())));
- assertEquals(
- 5,
- kubernetesClient
- .apps()
- .deployments()
- .inNamespace(namespace)
-
.withName(StandaloneKubernetesUtils.getTaskManagerDeploymentName(clusterId))
- .get()
- .getSpec()
- .getReplicas());
+ var ctx = getResourceContext();
- flinkDeployment
- .getSpec()
- .getFlinkConfiguration()
- .remove(JobManagerOptions.SCHEDULER_MODE.key());
- flinkDeployment
- .getStatus()
- .getReconciliationStatus()
- .serializeAndSetLastReconciledSpec(flinkDeployment.getSpec(),
flinkDeployment);
- ctx =
- new FlinkResourceContextFactory(
- configManager,
- TestUtils.createTestMetricGroup(new
Configuration()),
- null)
- .getResourceContext(flinkDeployment,
TestUtils.createEmptyContext());
-
- // Add replicas and verify that the scaling is not honoured as
reactive mode not enabled
- flinkDeployment.getSpec().getTaskManager().setReplicas(10);
- assertFalse(
+ flinkDeployment.getSpec().getTaskManager().setReplicas(200);
+ assertEquals(
+ expectScaled,
flinkStandaloneService.scale(ctx,
ctx.getDeployConfig(flinkDeployment.getSpec())));
- }
-
- @Test
- public void testTMReplicaScaleSession() {
- var clusterId = flinkDeployment.getMetadata().getName();
- var namespace = flinkDeployment.getMetadata().getNamespace();
- flinkDeployment.getSpec().setMode(KubernetesDeploymentMode.STANDALONE);
- // Add replicas
- flinkDeployment.getSpec().getTaskManager().setReplicas(3);
- flinkDeployment
- .getSpec()
- .getFlinkConfiguration()
- .put(
- JobManagerOptions.SCHEDULER_MODE.key(),
- SchedulerExecutionMode.REACTIVE.name());
- flinkDeployment
- .getStatus()
- .getReconciliationStatus()
- .serializeAndSetLastReconciledSpec(flinkDeployment.getSpec(),
flinkDeployment);
+ if (!expectScaled) {
+ return;
+ }
+ assertEquals(200, getTMReplicas(clusterId, namespace));
- var ctx =
- new FlinkResourceContextFactory(
- configManager,
- TestUtils.createTestMetricGroup(new
Configuration()),
- null)
- .getResourceContext(flinkDeployment,
TestUtils.createEmptyContext());
+ flinkDeployment.getSpec().getTaskManager().setReplicas(300);
assertTrue(
flinkStandaloneService.scale(ctx,
ctx.getDeployConfig(flinkDeployment.getSpec())));
+ assertEquals(300, getTMReplicas(clusterId, namespace));
- assertEquals(
- 3,
- kubernetesClient
- .apps()
- .deployments()
- .inNamespace(namespace)
-
.withName(StandaloneKubernetesUtils.getTaskManagerDeploymentName(clusterId))
- .get()
- .getSpec()
- .getReplicas());
-
- // Scale the replica count of the task managers
- flinkDeployment.getSpec().getTaskManager().setReplicas(10);
- createDeployments(flinkDeployment);
+ var requestsBefore = mockServer.getRequestCount();
assertTrue(
flinkStandaloneService.scale(ctx,
ctx.getDeployConfig(flinkDeployment.getSpec())));
-
- assertEquals(
- 10,
- kubernetesClient
- .apps()
- .deployments()
- .inNamespace(namespace)
-
.withName(StandaloneKubernetesUtils.getTaskManagerDeploymentName(clusterId))
- .get()
- .getSpec()
- .getReplicas());
+ assertEquals(1, mockServer.getRequestCount() - requestsBefore);
+ assertEquals(300, getTMReplicas(clusterId, namespace));
}
@Test
@@ -431,4 +364,21 @@ public class StandaloneFlinkServiceTest {
.inNamespace(cr.getMetadata().getNamespace())
.createOrReplace();
}
+
+ private FlinkResourceContext<FlinkDeployment> getResourceContext() {
+ return new FlinkResourceContextFactory(
+ configManager, TestUtils.createTestMetricGroup(new
Configuration()), null)
+ .getResourceContext(flinkDeployment,
TestUtils.createEmptyContext());
+ }
+
+ private Integer getTMReplicas(String clusterId, String namespace) {
+ return kubernetesClient
+ .apps()
+ .deployments()
+ .inNamespace(namespace)
+
.withName(StandaloneKubernetesUtils.getTaskManagerDeploymentName(clusterId))
+ .get()
+ .getSpec()
+ .getReplicas();
+ }
}
