Dennis-Mircea opened a new pull request, #1249: URL: https://github.com/apache/flink-kubernetes-operator/pull/1249
## What is the purpose of the change This PR adds a fix that marks the JobManager as `READY` only after its REST API responds. To achieve this, the code was restructured to ensure proper layering and separation of concerns. The implementation now has 3 distinct layers of observation: the JM Deployment (including its pods), the Flink Cluster process, and the Job. With this layering in place, the JobManager is now marked as `READY` by the Flink Cluster observation, while the JM Deployment observation can only advance as far as `DEPLOYED_NOT_READY`. For comparison, this is what was in place previously: <img width="389" height="301" alt="observeInternal-before" src="https://github.com/user-attachments/assets/32374258-09cb-40db-94ba-0420c091dfc6"; /> The Flink cluster was the white box because it had no observation step of its own. The Kubernetes step decided everything, as once the port accepted a connection, it set `DEPLOYED_NOT_READY`, and on the next observe (10s later by default) it set `READY` without asking Flink anything. The job observer then called the REST API as if the JobManager were up, and when that call timed out, the status was dropped back to `DEPLOYED_NOT_READY`. The cluster info was fetched only after READY was already set, so a failure there changed nothing. This scenario was presented already step by step within the JIRA. Now this is in place: <img width="389" height="301" alt="observeInternal-after" src="https://github.com/user-attachments/assets/c8a61cb5-3a17-458b-9cad-561fca84cc23"; /> Concretely (which is now translated 1-1 within the code structure): - The Kubernetes step (blue) looks at the Deployment and the JobManager pod. It can say the JobManager is missing, failed, starting, or started with its port open. It can never say `READY`. - The cluster step (orange) talks to the JobManager process through its REST API. If the call succeeds, the JobManager is `READY` and the version, revision and TaskManager count are stored. If the call fails, the Kubernetes step is asked again, to find out whether the pod is still there. - The job step (red) asks the JobMaster about its job: status, checkpoints, savepoints, state size, health. It runs only after the cluster step succeeded, so it never talks to a JobManager that is not answering. NOTE: The TaskManager pods are never observed on their own. In native mode, Flink's ResourceManager creates and replaces them, in standalone mode, the TaskManager Deployment does, and in both cases the operator only learns about them through the JobManager, as the registered TaskManager count returned by `/taskmanagers`. Two more things fall out of the split: - A session cluster no longer fetches its cluster info twice per loop. Before, `SessionObserver` fetched it once to see whether the REST API was reachable and threw the result away, then `observeClusterInfo` fetched it again to store it. Now the single fetch is the readiness check, and its result is what gets stored. - For application clusters, what belongs to the cluster (version, revision, TaskManager count) and what belongs to the job (status, snapshots, state size, health) are now observed in separate steps. The job status observer no longer needs to know about the Kubernetes Deployment, so its `onTimeout` hook is gone. The same split applied to `getClusterInfo`, which used to fetch the job's checkpoint statistics, without a timeout, as part of the cluster call. ## Brief change log - `AbstractFlinkDeploymentObserver`: `observeInternal` runs the Kubernetes step or the REST step, then the job step when `READY`. `observeJmDeployment` no longer sets `READY`. New `observeFlinkCluster` owns `READY`: REST call succeeds, `READY` plus cluster info; REST call fails, WARN if we were `READY` (job RUNNING moves to RECONCILING), INFO otherwise, then `observeJmDeployment`. TaskManager count, CPU and memory are fetched best-effort after a successful call. - `observeJob` replaces the abstract `observeFlinkCluster`: `ApplicationObserver` observes job status, snapshots, state size and health there, `SessionObserver` marks the spec stable. - `JobStatusObserver`: `onTimeout` hook removed, with the `ApplicationJobObserver` subclass. The RUNNING to RECONCILING move on a lost REST API now happens in `observeFlinkCluster`; `ifRunningMoveToReconciling` stays for a job status call that fails after a successful probe. - `FlinkService`: `getClusterInfo(conf, jobId)` split into `getClusterInfo(conf)`, `getTaskManagerReplicas(conf)` and `getStateSize(conf, jobId)`, the last one with the client timeout it was missing. `ClusterInfo` removed. - `kubernetes.operator.observer.rest-ready.delay` description reworded. ## Verifying this change The stability of this change can be verified by all existing tests that are still passing + new ones that cover the missing cases. Moreover, the change was verified against a local minikube setup and below, the correct log sequencing can be observed: ```log 2026-11-06 16:21:37,921 WARN ecutor-flinkdeploymentcontroller-157 o.a.f.k.o.o.d.ApplicationObserver JobManager REST API stopped answering java.util.concurrent.TimeoutException at java.base/java.util.concurrent.CompletableFuture.timedGet(Unknown Source) at java.base/java.util.concurrent.CompletableFuture.get(Unknown Source) at org.apache.flink.kubernetes.operator.service.AbstractFlinkService.getClusterInfo(AbstractFlinkService.java:814) at org.apache.flink.kubernetes.operator.observer.deployment.AbstractFlinkDeploymentObserver.observeFlinkCluster(AbstractFlinkDeploymentObserver.java:105) at org.apache.flink.kubernetes.operator.observer.deployment.AbstractFlinkDeploymentObserver.observeInternal(AbstractFlinkDeploymentObserver.java:91) at org.apache.flink.kubernetes.operator.observer.AbstractFlinkResourceObserver.observe(AbstractFlinkResourceObserver.java:49) at org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:145) at org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:59) at io.javaoperatorsdk.operator.processing.Controller$1.execute(Controller.java:191) at io.javaoperatorsdk.operator.processing.Controller$1.execute(Controller.java:147) at org.apache.flink.kubernetes.operator.metrics.OperatorJosdkMetrics.timeControllerExecution(OperatorJosdkMetrics.java:82) at io.javaoperatorsdk.operator.processing.Controller.reconcile(Controller.java:146) at io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.reconcileExecution(ReconciliationDispatcher.java:164) at io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleReconcile(ReconciliationDispatcher.java:145) at io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleDispatch(ReconciliationDispatcher.java:111) at io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleExecution(ReconciliationDispatcher.java:76) at io.javaoperatorsdk.operator.processing.event.EventProcessor$ReconcilerExecutor.run(EventProcessor.java:578) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) at java.base/java.lang.Thread.run(Unknown Source) 2026-11-06 16:21:37,922 INFO ecutor-flinkdeploymentcontroller-157 o.a.f.k.o.o.d.ApplicationObserver Observing JobManager deployment. Previous status: READY 2026-11-06 16:21:37,926 INFO ecutor-flinkdeploymentcontroller-157 o.a.f.k.o.o.d.ApplicationObserver JobManager deployment port is ready, waiting for the Flink REST API... 2026-11-06 16:21:37,941 INFO ecutor-flinkdeploymentcontroller-157 .r.d.AbstractFlinkResourceReconciler Resource fully reconciled, nothing to do... 2026-11-06 16:21:57,965 INFO ecutor-flinkdeploymentcontroller-168 o.a.f.k.o.o.d.ApplicationObserver JobManager REST API is not ready yet, no answer within 10 s 2026-11-06 16:21:57,965 INFO ecutor-flinkdeploymentcontroller-168 o.a.f.k.o.o.d.ApplicationObserver Observing JobManager deployment. Previous status: DEPLOYED_NOT_READY 2026-11-06 16:21:57,970 INFO ecutor-flinkdeploymentcontroller-168 o.a.f.k.o.o.d.ApplicationObserver JobManager deployment port is ready, waiting for the Flink REST API... 2026-11-06 16:21:57,972 INFO ecutor-flinkdeploymentcontroller-168 .r.d.AbstractFlinkResourceReconciler Resource fully reconciled, nothing to do... ``` ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): no - The public API, i.e., is any changes to the `CustomResourceDescriptors`: no - Core observer or reconciler logic that is regularly executed: yes ## Documentation - Does this pull request introduce a new feature? no - If yes, how is the feature documented? not applicable --- ##### Was generative AI tooling used to co-author this PR? <!-- If generative AI tooling has been used in the process of authoring this PR, please change the checkbox below to `[X]` followed by the name of the tool, and uncomment the "Generated-by" line. See the ASF Generative Tooling Guidance for details: https://www.apache.org/legal/generative-tooling.html You are responsible for the quality and correctness of every change in this PR regardless of the tooling used. Low-effort AI-generated PRs will be closed. See AGENTS.md for the full guidance. --> - [ ] Yes (please specify the tool below) <!-- Generated-by: [Tool Name and Version] --> -- 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]
