This is an automated email from the ASF dual-hosted git repository.
rmetzger 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 6eb5d9c09 [FLINK-37448] Report actual TaskManager count in
`status.taskManager.replicas` (#1146)
6eb5d9c09 is described below
commit 6eb5d9c09d57806114d3e1d66438b5b5c2675c48
Author: Santwana Verma <[email protected]>
AuthorDate: Thu Oct 29 00:26:24 2026 +0530
[FLINK-37448] Report actual TaskManager count in
`status.taskManager.replicas` (#1146)
* [FLINK-37448] Report actual TaskManager count in
status.taskManager.replicas
status.taskManager.replicas was computed from the spec
(ceil(parallelism / slots)), which is wrong when the TaskManager count
changes at runtime, e.g. in standalone reactive mode. The observer now
sets it from the TaskManagers registered with the cluster, for
application and session clusters.
Generated-by: Claude Code (Claude Opus 4.8)
* [FLINK-37448] Reuse the getClusterInfo REST call and keep the count on
in-place changes
- getClusterInfo returns the TaskManager count with the info map, so the
observer makes no extra /taskmanagers call.
- A RUNNING reconcile keeps the last observed count instead of resetting
it to 0. This covers in-place changes such as scaling; a redeploy still
reports 0 until the new cluster is observed.
- Drop comments that restate the code.
Generated-by: Claude Code (Claude Opus 4.8, Claude Opus 5.5)
* [FLINK-37448][docs] Update taskManager row in the status and lifecycle
page
The taskManager status is now populated for session clusters too and
reports the number of TaskManagers registered with the running cluster.
Generated-by: Claude Code (Claude Opus 5.5)
---
docs/content.zh/docs/custom-resource/reference.md | 2 +-
.../docs/custom-resource/status-and-lifecycle.md | 2 +-
docs/content/docs/custom-resource/reference.md | 2 +-
.../docs/custom-resource/status-and-lifecycle.md | 2 +-
.../operator/api/status/TaskManagerInfo.java | 5 +++-
.../AbstractFlinkDeploymentObserver.java | 12 ++++++--
.../operator/reconciler/ReconciliationUtils.java | 20 +++++++------
.../operator/service/AbstractFlinkService.java | 5 ++--
.../kubernetes/operator/service/ClusterInfo.java | 29 +++++++++++++++++++
.../kubernetes/operator/service/FlinkService.java | 2 +-
.../kubernetes/operator/utils/FlinkUtils.java | 4 +++
.../kubernetes/operator/TestingFlinkService.java | 17 +++++++++--
.../controller/FlinkDeploymentControllerTest.java | 8 ++++--
.../deployment/ApplicationObserverTest.java | 33 ++++++++++++++++++++++
.../observer/deployment/SessionObserverTest.java | 24 ++++++++++++++++
.../operator/service/AbstractFlinkServiceTest.java | 6 ++--
.../operator/utils/ReconciliationUtilsTest.java | 15 ++++++++++
17 files changed, 161 insertions(+), 27 deletions(-)
diff --git a/docs/content.zh/docs/custom-resource/reference.md
b/docs/content.zh/docs/custom-resource/reference.md
index 4348451ec..51bf7b3e1 100644
--- a/docs/content.zh/docs/custom-resource/reference.md
+++ b/docs/content.zh/docs/custom-resource/reference.md
@@ -554,4 +554,4 @@ This serves as a full reference for the FlinkDeployment,
FlinkSessionJob, FlinkS
| Parameter | Type | Docs |
| ----------| ---- | ---- |
| labelSelector | java.lang.String | TaskManager label selector. |
-| replicas | int | Number of TaskManager replicas if defined in the spec. |
+| replicas | int | Number of TaskManagers registered with the running Flink
cluster, as of the operator's last successful observation. |
diff --git a/docs/content.zh/docs/custom-resource/status-and-lifecycle.md
b/docs/content.zh/docs/custom-resource/status-and-lifecycle.md
index 8c3167c03..ba288d9c0 100644
--- a/docs/content.zh/docs/custom-resource/status-and-lifecycle.md
+++ b/docs/content.zh/docs/custom-resource/status-and-lifecycle.md
@@ -61,7 +61,7 @@ The status is the operator's write-back half of the contract:
users declare the
| `observedGeneration` | Both | The generation of the
resource the operator last observed
|
| `jobManagerDeploymentStatus` | `FlinkDeployment` | The observed state of the
JobManager's Kubernetes Deployment, see [JobManager Deployment
Status](#jobmanager-deployment-status)
|
| `clusterInfo` | `FlinkDeployment` | The Flink version and
revision of the running cluster, its total CPU and memory, the last observed
state size, and the operator's cluster health bookkeeping |
-| `taskManager` | `FlinkDeployment` | The TaskManager label
selector and replica count backing the Kubernetes scale subresource, populated
in application mode only |
+| `taskManager` | `FlinkDeployment` | The TaskManager label
selector and the number of TaskManagers registered with the running cluster at
the last observation, backing the Kubernetes scale subresource |
| `conditions` | `FlinkDeployment` | Kubernetes-standard
conditions derived from the status, see [Conditions](#conditions)
|
### Reconciliation Status
diff --git a/docs/content/docs/custom-resource/reference.md
b/docs/content/docs/custom-resource/reference.md
index 4348451ec..51bf7b3e1 100644
--- a/docs/content/docs/custom-resource/reference.md
+++ b/docs/content/docs/custom-resource/reference.md
@@ -554,4 +554,4 @@ This serves as a full reference for the FlinkDeployment,
FlinkSessionJob, FlinkS
| Parameter | Type | Docs |
| ----------| ---- | ---- |
| labelSelector | java.lang.String | TaskManager label selector. |
-| replicas | int | Number of TaskManager replicas if defined in the spec. |
+| replicas | int | Number of TaskManagers registered with the running Flink
cluster, as of the operator's last successful observation. |
diff --git a/docs/content/docs/custom-resource/status-and-lifecycle.md
b/docs/content/docs/custom-resource/status-and-lifecycle.md
index 8c3167c03..ba288d9c0 100644
--- a/docs/content/docs/custom-resource/status-and-lifecycle.md
+++ b/docs/content/docs/custom-resource/status-and-lifecycle.md
@@ -61,7 +61,7 @@ The status is the operator's write-back half of the contract:
users declare the
| `observedGeneration` | Both | The generation of the
resource the operator last observed
|
| `jobManagerDeploymentStatus` | `FlinkDeployment` | The observed state of the
JobManager's Kubernetes Deployment, see [JobManager Deployment
Status](#jobmanager-deployment-status)
|
| `clusterInfo` | `FlinkDeployment` | The Flink version and
revision of the running cluster, its total CPU and memory, the last observed
state size, and the operator's cluster health bookkeeping |
-| `taskManager` | `FlinkDeployment` | The TaskManager label
selector and replica count backing the Kubernetes scale subresource, populated
in application mode only |
+| `taskManager` | `FlinkDeployment` | The TaskManager label
selector and the number of TaskManagers registered with the running cluster at
the last observation, backing the Kubernetes scale subresource |
| `conditions` | `FlinkDeployment` | Kubernetes-standard
conditions derived from the status, see [Conditions](#conditions)
|
### Reconciliation Status
diff --git
a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/TaskManagerInfo.java
b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/TaskManagerInfo.java
index c5973af54..d1503b00b 100644
---
a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/TaskManagerInfo.java
+++
b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/TaskManagerInfo.java
@@ -38,6 +38,9 @@ public class TaskManagerInfo {
/** TaskManager label selector. */
@LabelSelector private String labelSelector;
- /** Number of TaskManager replicas if defined in the spec. */
+ /**
+ * Number of TaskManagers registered with the running Flink cluster, as of
the operator's last
+ * successful observation.
+ */
@StatusReplicas private int replicas = 0;
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java
index 54e1573bf..91d63a477 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java
@@ -24,6 +24,7 @@ import org.apache.flink.kubernetes.operator.api.spec.JobSpec;
import org.apache.flink.kubernetes.operator.api.spec.JobState;
import org.apache.flink.kubernetes.operator.api.status.FlinkDeploymentStatus;
import
org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatus;
+import org.apache.flink.kubernetes.operator.api.status.TaskManagerInfo;
import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
import
org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
import
org.apache.flink.kubernetes.operator.exception.MissingJobManagerException;
@@ -80,13 +81,20 @@ public abstract class AbstractFlinkDeploymentObserver
private void observeClusterInfo(FlinkResourceContext<FlinkDeployment> ctx)
{
var flinkApp = ctx.getResource();
try {
- Map<String, String> clusterInfo =
+ var clusterInfo =
ctx.getFlinkService()
.getClusterInfo(
ctx.getObserveConfig(),
flinkApp.getStatus().getJobStatus().getJobId());
- flinkApp.getStatus().getClusterInfo().putAll(clusterInfo);
+
flinkApp.getStatus().getClusterInfo().putAll(clusterInfo.getInfo());
logger.debug("ClusterInfo: {}",
flinkApp.getStatus().getClusterInfo());
+
+ var labelSelector =
+
FlinkUtils.getTaskManagerLabelSelector(flinkApp.getMetadata().getName());
+ flinkApp.getStatus()
+ .setTaskManager(
+ new TaskManagerInfo(
+ labelSelector,
clusterInfo.getTaskManagerReplicas()));
} catch (Exception e) {
logger.warn("Exception while fetching cluster info", e);
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
index 94140c3ff..282deb193 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
@@ -50,6 +50,8 @@ import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
@@ -169,12 +171,12 @@ public class ReconciliationUtils {
if (target instanceof FlinkDeployment) {
// For application deployments we update the taskmanager
info
- ((FlinkDeploymentStatus) status)
- .setTaskManager(
- getTaskManagerInfo(
- target.getMetadata().getName(),
- conf,
- stateAfterReconcile));
+ var deploymentStatus = (FlinkDeploymentStatus) status;
+ deploymentStatus.setTaskManager(
+ getTaskManagerInfo(
+ target.getMetadata().getName(),
+ stateAfterReconcile,
+ deploymentStatus.getTaskManager()));
}
reconciliationStatus.serializeAndSetLastReconciledSpec(clonedSpec, target);
if (spec.getJob().getState() == JobState.SUSPENDED) {
@@ -240,10 +242,10 @@ public class ReconciliationUtils {
}
private static TaskManagerInfo getTaskManagerInfo(
- String name, Configuration conf, JobState jobState) {
- var labelSelector = "component=taskmanager,app=" + name;
+ String name, JobState jobState, @Nullable TaskManagerInfo current)
{
if (jobState == JobState.RUNNING) {
- return new TaskManagerInfo(labelSelector,
FlinkUtils.getNumTaskManagers(conf));
+ int replicas = current != null ? current.getReplicas() : 0;
+ return new
TaskManagerInfo(FlinkUtils.getTaskManagerLabelSelector(name), replicas);
} else {
return new TaskManagerInfo("", 0);
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
index 99753a7e2..693d7b985 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
@@ -803,8 +803,7 @@ public abstract class AbstractFlinkService implements
FlinkService {
}
@Override
- public Map<String, String> getClusterInfo(Configuration conf, @Nullable
String jobId)
- throws Exception {
+ public ClusterInfo getClusterInfo(Configuration conf, @Nullable String
jobId) throws Exception {
Map<String, String> clusterInfo = new HashMap<>();
populateFlinkVersion(conf, clusterInfo);
@@ -819,7 +818,7 @@ public abstract class AbstractFlinkService implements
FlinkService {
populateStateSize(conf, jobId, clusterInfo);
- return clusterInfo;
+ return new ClusterInfo(clusterInfo, taskManagerReplicas);
}
private void populateFlinkVersion(Configuration conf, Map<String, String>
clusterInfo)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/ClusterInfo.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/ClusterInfo.java
new file mode 100644
index 000000000..908e39606
--- /dev/null
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/ClusterInfo.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.service;
+
+import lombok.Value;
+
+import java.util.Map;
+
+/** Snapshot of the observed Flink cluster returned by {@link
FlinkService#getClusterInfo}. */
+@Value
+public class ClusterInfo {
+ Map<String, String> info;
+ int taskManagerReplicas;
+}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
index b1a078fbb..d6f7a966d 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
@@ -117,7 +117,7 @@ public interface FlinkService {
void disposeSavepoint(String savepointPath, Configuration conf) throws
Exception;
- Map<String, String> getClusterInfo(Configuration conf, @Nullable String
jobId) throws Exception;
+ ClusterInfo getClusterInfo(Configuration conf, @Nullable String jobId)
throws Exception;
PodList getJmPodList(FlinkDeployment deployment, Configuration conf);
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java
index dbd68ba4e..b0a0f0259 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java
@@ -351,6 +351,10 @@ public class FlinkUtils {
|| haMode.equalsIgnoreCase("kubernetes");
}
+ public static String getTaskManagerLabelSelector(String clusterId) {
+ return "component=taskmanager,app=" + clusterId;
+ }
+
public static int getNumTaskManagers(Configuration conf) {
int parallelism = conf.get(CoreOptions.DEFAULT_PARALLELISM);
return getNumTaskManagers(conf, parallelism);
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
index b17fb624a..a8eaa9fae 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
@@ -49,8 +49,10 @@ import
org.apache.flink.kubernetes.operator.observer.CheckpointStatsResult;
import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
import org.apache.flink.kubernetes.operator.service.AbstractFlinkService;
import org.apache.flink.kubernetes.operator.service.CheckpointHistoryWrapper;
+import org.apache.flink.kubernetes.operator.service.ClusterInfo;
import org.apache.flink.kubernetes.operator.service.SuspendMode;
import
org.apache.flink.kubernetes.operator.standalone.StandaloneKubernetesConfigOptionsInternal;
+import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.runtime.client.DuplicateJobSubmissionException;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.clusterframework.ApplicationStatus;
@@ -156,6 +158,9 @@ public class TestingFlinkService extends
AbstractFlinkService {
@Getter private int desiredReplicas = 0;
@Getter private int cancelJobCallCount = 0;
+ @Setter private Integer taskManagerReplicas = null;
+ @Setter private boolean taskManagerReplicasError = false;
+
@Getter private Configuration submittedConf;
@Setter
@@ -717,12 +722,18 @@ public class TestingFlinkService extends
AbstractFlinkService {
}
@Override
- public Map<String, String> getClusterInfo(Configuration conf, String jobId)
- throws TimeoutException {
+ public ClusterInfo getClusterInfo(Configuration conf, String jobId) throws
TimeoutException {
if (!isPortReady) {
throw new TimeoutException("JM port is unavailable");
}
- return CLUSTER_INFO;
+ if (taskManagerReplicasError) {
+ throw new RuntimeException("Failed to fetch TaskManager count");
+ }
+ int replicas =
+ taskManagerReplicas != null
+ ? taskManagerReplicas
+ : FlinkUtils.getNumTaskManagers(conf);
+ return new ClusterInfo(CLUSTER_INFO, replicas);
}
@Override
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
index 7581fce6b..982919963 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
@@ -39,6 +39,7 @@ import
org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import
org.apache.flink.kubernetes.operator.reconciler.deployment.AbstractFlinkResourceReconciler;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.kubernetes.operator.utils.IngressUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.util.SerializedThrowable;
@@ -213,13 +214,15 @@ public class FlinkDeploymentControllerTest {
}
appCluster.getSpec().getJob().setState(JobState.RUNNING);
+ flinkService.setTaskManagerReplicas(3);
verifyReconcileNormalLifecycle(appCluster);
var jobs = flinkService.listJobs();
assertEquals(1, jobs.size());
assertEquals("s0", jobs.get(0).f0);
assertEquals(
new TaskManagerInfo(
- "component=taskmanager,app=" +
appCluster.getMetadata().getName(), 1),
+
FlinkUtils.getTaskManagerLabelSelector(appCluster.getMetadata().getName()),
+ 3),
appCluster.getStatus().getTaskManager());
}
@@ -378,7 +381,8 @@ public class FlinkDeploymentControllerTest {
assertEquals("s0", jobs.get(0).f0);
assertEquals(
new TaskManagerInfo(
- "component=taskmanager,app=" +
appCluster.getMetadata().getName(), 1),
+
FlinkUtils.getTaskManagerLabelSelector(appCluster.getMetadata().getName()),
+ 0),
appCluster.getStatus().getTaskManager());
assertEquals("s0",
appCluster.getStatus().getJobStatus().getUpgradeSavepointPath());
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
index 31493b501..67265eaf7 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
@@ -31,6 +31,7 @@ import
org.apache.flink.kubernetes.operator.api.status.JobStatus;
import org.apache.flink.kubernetes.operator.api.status.ReconciliationState;
import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.api.status.TaskManagerInfo;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import
org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
@@ -89,6 +90,38 @@ public class ApplicationObserverTest extends
OperatorTestBase {
deployment.getStatus().getJobStatus().setJobId(jobId);
}
+ @Test
+ public void observeReportsActualTaskManagerReplicas() throws Exception {
+ Configuration conf =
+ configManager.getDeployConfig(deployment.getMetadata(),
deployment.getSpec());
+
+ observer.observe(deployment, TestUtils.createEmptyContext());
+ flinkService.submitApplicationCluster(deployment.getSpec().getJob(),
conf, false);
+ ReconciliationUtils.updateStatusForDeployedSpec(deployment, new
Configuration());
+
+ flinkService.setTaskManagerReplicas(5);
+ observer.observe(deployment, readyContext);
+ observer.observe(deployment, readyContext);
+ assertEquals(
+ JobManagerDeploymentStatus.READY,
+ deployment.getStatus().getJobManagerDeploymentStatus());
+
+ var labelSelector =
+
FlinkUtils.getTaskManagerLabelSelector(deployment.getMetadata().getName());
+ assertEquals(
+ new TaskManagerInfo(labelSelector, 5),
deployment.getStatus().getTaskManager());
+
+ flinkService.setTaskManagerReplicas(3);
+ observer.observe(deployment, readyContext);
+ assertEquals(
+ new TaskManagerInfo(labelSelector, 3),
deployment.getStatus().getTaskManager());
+
+ flinkService.setTaskManagerReplicasError(true);
+ observer.observe(deployment, readyContext);
+ assertEquals(
+ new TaskManagerInfo(labelSelector, 3),
deployment.getStatus().getTaskManager());
+ }
+
@Test
public void observeApplicationCluster() throws Exception {
Configuration conf =
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserverTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserverTest.java
index cfb1910cb..e5fbb8b4e 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserverTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserverTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
import
org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatus;
import org.apache.flink.kubernetes.operator.api.status.ReconciliationState;
+import org.apache.flink.kubernetes.operator.api.status.TaskManagerInfo;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.observer.TestObserverAdapter;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
@@ -101,6 +102,29 @@ public class SessionObserverTest extends OperatorTestBase {
deployment.getStatus().getJobManagerDeploymentStatus());
}
+ @Test
+ public void observeReportsActualTaskManagerReplicas() {
+ FlinkDeployment deployment = TestUtils.buildSessionCluster();
+ ReconciliationUtils.updateStatusForDeployedSpec(deployment, new
Configuration());
+
+ flinkService.setTaskManagerReplicas(5);
+ observer.observe(deployment, readyContext);
+ observer.observe(deployment, readyContext);
+ assertEquals(
+ JobManagerDeploymentStatus.READY,
+ deployment.getStatus().getJobManagerDeploymentStatus());
+
+ var labelSelector =
+
FlinkUtils.getTaskManagerLabelSelector(deployment.getMetadata().getName());
+ assertEquals(
+ new TaskManagerInfo(labelSelector, 5),
deployment.getStatus().getTaskManager());
+
+ flinkService.setTaskManagerReplicas(2);
+ observer.observe(deployment, readyContext);
+ assertEquals(
+ new TaskManagerInfo(labelSelector, 2),
deployment.getStatus().getTaskManager());
+ }
+
@Test
public void observeAlreadyUpgraded() {
var kubernetesDeployment = TestUtils.createDeployment(true);
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
index abac03963..baf275a78 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
@@ -1254,14 +1254,16 @@ public class AbstractFlinkServiceTest {
AbstractFlinkService.FIELD_NAME_TOTAL_MEMORY,
"" + MemorySize.ofMebiBytes(1000).getBytes() * 2);
- assertEquals(expectedEntries, flinkService.getClusterInfo(conf, null));
+ var clusterInfo = flinkService.getClusterInfo(conf, null);
+ assertEquals(1, clusterInfo.getTaskManagerReplicas());
+ assertEquals(expectedEntries, clusterInfo.getInfo());
assertEquals(
ImmutableMap.<String, String>builder()
.putAll(expectedEntries)
.put(AbstractFlinkService.FIELD_NAME_STATE_SIZE,
"42424242")
.build(),
- flinkService.getClusterInfo(conf,
JobID.generate().toHexString()));
+ flinkService.getClusterInfo(conf,
JobID.generate().toHexString()).getInfo());
}
@Test
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ReconciliationUtilsTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ReconciliationUtilsTest.java
index f5afefa16..6cbcce4cb 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ReconciliationUtilsTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ReconciliationUtilsTest.java
@@ -21,6 +21,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
import org.apache.flink.kubernetes.operator.api.spec.JobState;
import org.apache.flink.kubernetes.operator.api.status.ReconciliationState;
+import org.apache.flink.kubernetes.operator.api.status.TaskManagerInfo;
import org.apache.flink.kubernetes.operator.api.utils.BaseTestUtils;
import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
@@ -64,6 +65,20 @@ public class ReconciliationUtilsTest {
assertNotEquals(0, updateControl.getScheduleDelay().get());
}
+ @Test
+ public void taskManagerReplicasPreservedOnRunningSpecUpdate() {
+ FlinkDeployment app = BaseTestUtils.buildApplicationCluster();
+ app.getSpec().getJob().setState(JobState.RUNNING);
+
app.getStatus().getReconciliationStatus().setState(ReconciliationState.DEPLOYED);
+
+ var labelSelector =
FlinkUtils.getTaskManagerLabelSelector(app.getMetadata().getName());
+ app.getStatus().setTaskManager(new TaskManagerInfo(labelSelector, 7));
+
+ ReconciliationUtils.updateStatusForDeployedSpec(app, new
Configuration());
+
+ assertEquals(new TaskManagerInfo(labelSelector, 7),
app.getStatus().getTaskManager());
+ }
+
@Test
public void testObservedGenerationStatus() throws Exception {
FlinkDeployment app = BaseTestUtils.buildApplicationCluster();
