From 0a1dacd2226d0a4b996d64359007b952c8fbd33c Mon Sep 17 00:00:00 2001
From: Luca Castelli <luca.castelli@shopify.com>
Date: Tue, 18 Feb 2025 12:01:49 -0500
Subject: [PATCH] [FLINK-37320] [Observer] FINISHED finite streaming jobs
 incorrectly being set to RECONCILING

---
 .../observer/deployment/AbstractFlinkDeploymentObserver.java | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java
index 38d4b831c..713e132ad 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java
+++ b/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) {