vsantwana commented on code in PR #1146:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1146#discussion_r4392270415
##########
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:
Agreed. `getClusterInfo` now returns the TaskManager count along with the
info map, so the observer makes no extra `/taskmanagers` call.
##########
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:
Removed.
##########
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:
Removed, along with the other comments this PR added to the tests.
##########
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:
Removed.
##########
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:
A RUNNING reconcile now keeps the last observed count, so an in-place change
such as scaling no longer drops the field to 0.
A full upgrade or a resume still reports 0 until the operator observes the
new cluster. Before it deploys, the reconciler records the target spec with the
job SUSPENDED, which clears the field, and at that point no TaskManagers have
registered with the new cluster yet. So 0 is the actual count there, and I
would rather keep it than report TaskManagers that are gone.
This does not disable an HPA: it takes the current replica count from the
scale subresource's `spec.replicas`, and reads `status.replicas` only for
per-pod averages of Object and External metrics, where 0 is handled. Is there
another consumer you have in mind that relies on the field staying non-zero?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
