Re: [PR] [FLINK-37448] Report actual TaskManager count in status.taskManager.replicas [flink-kubernetes-operator]

["Dennis-Mircea (via GitHub)" <gi...@apache.org>]

Dennis-Mircea commented on code in PR #1146:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1146#discussion_r3511079146


##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java:
##########
@@ -87,6 +88,14 @@ private void observeClusterInfo(FlinkResourceContext<FlinkDeployment> ctx) {
                                     flinkApp.getStatus().getJobStatus().getJobId());
             flinkApp.getStatus().getClusterInfo().putAll(clusterInfo);
             logger.debug("ClusterInfo: {}", flinkApp.getStatus().getClusterInfo());
+
+            // Report the actual number of TaskManagers registered with the running cluster rather
+            // than a value derived from the spec, so the status reflects dynamically changing
+            // deployments (e.g. standalone reactive mode). See FLINK-37448.
+            var replicas = ctx.getFlinkService().getTaskManagerReplicas(ctx.getObserveConfig());

Review Comment:
   By adding this implementation here, it will make a subsequent unnecessary REST call, as the `ctx.getFlinkService().getClusterInfo(...)` already does on REST call.. This should be integrated as part of `getClusterInfo(...)` method.



##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java:
##########
@@ -317,9 +317,11 @@ public void verifyUpgradeFromSavepointLegacyMode(FlinkVersion flinkVersion) thro
         var jobs = flinkService.listJobs();
         assertEquals(1, jobs.size());
         assertEquals("s0", jobs.get(0).f0);
+        // The actual replica count is populated during observation from the running cluster; right
+        // after reconciliation (cluster not yet observed) it stays 0 (FLINK-37448).

Review Comment:
   No need for this comment.



##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java:
##########
@@ -87,6 +88,14 @@ private void observeClusterInfo(FlinkResourceContext<FlinkDeployment> ctx) {
                                     flinkApp.getStatus().getJobStatus().getJobId());
             flinkApp.getStatus().getClusterInfo().putAll(clusterInfo);
             logger.debug("ClusterInfo: {}", flinkApp.getStatus().getClusterInfo());
+
+            // Report the actual number of TaskManagers registered with the running cluster rather
+            // than a value derived from the spec, so the status reflects dynamically changing
+            // deployments (e.g. standalone reactive mode). See FLINK-37448.

Review Comment:
   Too much unnecessary comments verbosity. This comment can be removed entirely.



##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java:
##########
@@ -239,11 +237,12 @@ private static void updateLastReconciledJobSpec(
         }
     }
 
-    private static TaskManagerInfo getTaskManagerInfo(
-            String name, Configuration conf, JobState jobState) {
-        var labelSelector = "component=taskmanager,app=" + name;
+    private static TaskManagerInfo getTaskManagerInfo(String name, JobState jobState) {
         if (jobState == JobState.RUNNING) {
-            return new TaskManagerInfo(labelSelector, FlinkUtils.getNumTaskManagers(conf));
+            // The actual replica count is populated during observation from the running cluster
+            // (see AbstractFlinkDeploymentObserver#observeClusterInfo). Here we only set the label
+            // selector; the replica count stays 0 until the cluster is observed.

Review Comment:
   Unnecessary comment verbosity.



##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java:
##########
@@ -239,11 +237,12 @@ private static void updateLastReconciledJobSpec(
         }
     }
 
-    private static TaskManagerInfo getTaskManagerInfo(
-            String name, Configuration conf, JobState jobState) {
-        var labelSelector = "component=taskmanager,app=" + name;
+    private static TaskManagerInfo getTaskManagerInfo(String name, JobState jobState) {
         if (jobState == JobState.RUNNING) {
-            return new TaskManagerInfo(labelSelector, FlinkUtils.getNumTaskManagers(conf));
+            // The actual replica count is populated during observation from the running cluster
+            // (see AbstractFlinkDeploymentObserver#observeClusterInfo). Here we only set the label
+            // selector; the replica count stays 0 until the cluster is observed.
+            return new TaskManagerInfo(FlinkUtils.getTaskManagerLabelSelector(name), 0);

Review Comment:
   This now resets replicas to 0 when RUNNING, and it stays 0 until the next observe populates it. In steady state this is fine, but right after every upgrade/spec change the scale subresource reports 0 until the next observe loop. This doesn't look ok.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org