[FLINK-32033][Kubernetes-Operator] Fix Lifecycle status in case of MISSING/ERROR JM status with unrecoverable error

Author: nishita-pattanayak

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/CommonStatus.java

@@ -93,25 +93,16 @@ public ResourceLifecycleState getLifecycleState() {
         // Check for unrecoverable deployments that should be marked as FAILED
         if (this instanceof FlinkDeploymentStatus) {
             FlinkDeploymentStatus deploymentStatus = (FlinkDeploymentStatus) this;
-            var jmStatus = deploymentStatus.getJobManagerDeploymentStatus();
-
-            // ERROR deployments are in terminal error state and should always be FAILED
-            if (jmStatus == JobManagerDeploymentStatus.ERROR) {
+            var jmDeployStatus = deploymentStatus.getJobManagerDeploymentStatus();
+
+            // ERROR/MISSING deployments are in terminal error state
+            // [Configmaps deleted -> require manual restore]  and should always be FAILED
+            if ((jmDeployStatus == JobManagerDeploymentStatus.MISSING
+                            || jmDeployStatus == JobManagerDeploymentStatus.ERROR)
+                    && StringUtils.isNotEmpty(error)
+                    && error.contains("configmaps have been deleted")) {
                 return ResourceLifecycleState.FAILED;
             }
-
-            // MISSING deployments should be FAILED if they're clearly unrecoverable
-            if (jmStatus == JobManagerDeploymentStatus.MISSING) {
-                // Mark as FAILED if error message clearly indicates deployment failure (any time)
-                if (StringUtils.isNotEmpty(error)) {
-                    return ResourceLifecycleState.FAILED;
-                }
-                // Also mark as FAILED if stable deployment is missing without any error
-                // (indicating it was deleted externally)
-                else if (reconciliationStatus.isLastReconciledSpecStable()) {
-                    return ResourceLifecycleState.FAILED;
-                }
-            }
         }
 
         if (reconciliationStatus.getState() == ReconciliationState.ROLLED_BACK) {

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/ResourceLifecycleMetricsTest.java

@@ -72,28 +72,8 @@ public void lifecycleStateTest() {
         ReconciliationUtils.updateStatusForDeployedSpec(application, new Configuration());
         assertEquals(DEPLOYED, application.getStatus().getLifecycleState());
 
-        application.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
         application.getStatus().getReconciliationStatus().markReconciledSpecAsStable();
-        assertEquals(
-                STABLE,
-                application.getStatus().getLifecycleState(),
-                "JobManager Deployment is in DEPLOYING state, hence application is STABLE");
-
-        application
-                .getStatus()
-                .setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYED_NOT_READY);
-        application.getStatus().getReconciliationStatus().markReconciledSpecAsStable();
-        assertEquals(
-                STABLE,
-                application.getStatus().getLifecycleState(),
-                "JobManager Deployment is in DEPLOYED_NOT_READY state, hence application is STABLE");
-
-        application.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
-        application.getStatus().getReconciliationStatus().markReconciledSpecAsStable();
-        assertEquals(
-                STABLE,
-                application.getStatus().getLifecycleState(),
-                "JobManager Deployment is in READY state, hence application is STABLE");
+        assertEquals(STABLE, application.getStatus().getLifecycleState());
 
         application.getStatus().setError("errr");
         assertEquals(STABLE, application.getStatus().getLifecycleState());
@@ -369,35 +349,28 @@ public void testUnrecoverableDeploymentLifecycleState() {
         application.getStatus().getReconciliationStatus().markReconciledSpecAsStable();
 
         application.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.ERROR);
-        application.getStatus().setError(null);
-        assertEquals(
-                FAILED,
-                application.getStatus().getLifecycleState(),
-                "ERROR deployment should always be FAILED (terminal error state)");
-
-        application.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.ERROR);
-        application.getStatus().setError("JobManager deployment failed to start");
+        application
+                .getStatus()
+                .setError(
+                        "JobManager deployment is missing and HA data is not available to make stateful upgrades. "
+                                + "It is possible that the job has finished or terminally failed, or the configmaps have been deleted. "
+                                + "Manual restore required.");
         assertEquals(
                 FAILED,
                 application.getStatus().getLifecycleState(),
-                "ERROR deployment with error message should also be FAILED");
+                "ERROR deployment with `configmaps have been deleted` error should always be FAILED (terminal error state)");
 
         application.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.MISSING);
         application
                 .getStatus()
-                .setError("JobManager deployment was deleted and cannot be recovered");
+                .setError(
+                        "HA metadata not available to restore from last state. "
+                                + "It is possible that the job has finished or terminally failed, or the configmaps have been deleted. ");
         assertEquals(
                 FAILED,
                 application.getStatus().getLifecycleState(),
                 "MISSING deployment with error should be FAILED");
 
-        application.getStatus().setError(null);
-        application.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.MISSING);
-        assertEquals(
-                FAILED,
-                application.getStatus().getLifecycleState(),
-                "MISSING deployment with stable reconciliation should be FAILED");
-
         application.getStatus().setError(null);
         application.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.MISSING);
         // Reset to DEPLOYED state (not stable yet) to simulate ongoing deployment