[FLINK-37320] [Observer] FINISHED finite streaming jobs incorrectly being set to RECONCILING

Author: luca-p-castelli

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java

@@ -152,7 +152,10 @@ protected void observeJmDeployment(FlinkResourceContext<FlinkDeployment> ctx) {
         }
 
         deploymentStatus.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.MISSING);
-        deploymentStatus.getJobStatus().setState(JobStatus.RECONCILING);
+
+        if (!ReconciliationUtils.isJobInTerminalState(deploymentStatus)) {
+            deploymentStatus.getJobStatus().setState(JobStatus.RECONCILING);
+        }
 
         if (previousJmStatus != JobManagerDeploymentStatus.MISSING
                 && previousJmStatus != JobManagerDeploymentStatus.ERROR) {

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java

@@ -876,4 +876,26 @@ public void validateLastReconciledClearedOnInitialFailure() {
         observer.observe(deployment, TestUtils.createEmptyContext());
         assertTrue(reconStatus.isBeforeFirstDeployment());
     }
+
+    @Test
+    public void jobStatusNotOverwrittenWhenTerminal() throws Exception {
+        Configuration conf =
+                configManager.getDeployConfig(deployment.getMetadata(), deployment.getSpec());
+        flinkService.submitApplicationCluster(deployment.getSpec().getJob(), conf, false);
+        bringToReadyStatus(deployment);
+
+        deployment
+                .getStatus()
+                .getJobStatus()
+                .setState(org.apache.flink.api.common.JobStatus.FINISHED);
+
+        // Simulate missing deployment
+        var emptyContext = TestUtils.createEmptyContext();
+        deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.MISSING);
+        observer.observe(deployment, emptyContext);
+
+        assertEquals(
+                org.apache.flink.api.common.JobStatus.FINISHED,
+                deployment.getStatus().getJobStatus().getState());
+    }
 }