[GitHub] [airflow] uranusjr commented on a diff in pull request #28871: KubenetesExecutor sends state even when successful

[GitBox <gi...@apache.org>]

uranusjr commented on code in PR #28871:
URL: https://github.com/apache/airflow/pull/28871#discussion_r1067709600


##########
airflow/executors/kubernetes_executor.py:
##########
@@ -751,19 +751,25 @@ def _change_state(self, key: TaskInstanceKey, state: str | None, pod_id: str, na
         if TYPE_CHECKING:
             assert self.kube_scheduler
 
-        if state != State.RUNNING:
-            if self.kube_config.delete_worker_pods:
-                if state != State.FAILED or self.kube_config.delete_worker_pods_on_failure:
-                    self.kube_scheduler.delete_pod(pod_id, namespace)
-                    self.log.info("Deleted pod: %s in namespace %s", str(key), str(namespace))
-            else:
-                self.kube_scheduler.patch_pod_executor_done(pod_id=pod_id, namespace=namespace)
-                self.log.info("Patched pod %s in namespace %s to mark it as done", str(key), str(namespace))
-            try:
-                self.running.remove(key)
-            except KeyError:
-                self.log.debug("Could not find key: %s", str(key))
-        self.event_buffer[key] = state, None
+        if state == State.RUNNING:
+            self.event_buffer[key] = state, None
+            return
+
+        if self.kube_config.delete_worker_pods:
+            if state != State.FAILED or self.kube_config.delete_worker_pods_on_failure:
+                self.kube_scheduler.delete_pod(pod_id, namespace)
+                self.log.info("Deleted pod: %s in namespace %s", str(key), str(namespace))
+        else:
+            self.kube_scheduler.patch_pod_executor_done(pod_id=pod_id, namespace=namespace)
+            self.log.info("Patched pod %s in namespace %s to mark it as done", str(key), str(namespace))
+
+        if key in self.running:
+            self.running.remove(key)
+            # We do get multiple events once the pod hits a terminal state, and we only want to
+            # do this once, so only do it when we remove the task from running
+            self.event_buffer[key] = state, None
+        else:
+            self.log.debug("TI key not in running, not adding to event_buffer: %s", str(key))

Review Comment:
   ```suggestion
           try:
               self.running.remove(key)
           except KeyError:
               self.log.debug("TI key not in running, not adding to event_buffer: %s", str(key))
           else:
               # We get multiple events once the pod hits a terminal state, and we only want to
               # do this once, so only do it when we remove the task from running
               self.event_buffer[key] = state, None
   ```



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org