Schedulers are issuing abrupt pod deletes when there is a delay in schedulers' heartbeat

airflow/executors/kubernetes_executor_utils.py

@@ -129,7 +129,10 @@ def _run(
     ) -> str | None:
         self.log.info("Event: and now my watch begins starting at resource_version: %s", resource_version)
 
-        kwargs = {"label_selector": f"airflow-worker={scheduler_job_id}"}
+        # Schedulers are issuing abrupt pod deletes when there is a delay in schedulers' heartbeat
+        # https://github.com/apache/airflow/issues/31198
+        # Removed scheduler scheduler_job_id filter to receive events from all airflow worker pods
+        kwargs = {"label_selector": "airflow-worker"}
         if resource_version:
             kwargs["resource_version"] = resource_version
         if kube_config.kube_client_request_args:
@@ -143,6 +146,13 @@ def _run(
             self.log.debug("Event: %s had an event of type %s", task.metadata.name, event["type"])
             if event["type"] == "ERROR":
                 return self.process_error(event)
+            # Schedulers are issuing abrupt pod deletes when there is a delay in schedulers' heartbeat
+            # https://github.com/apache/airflow/issues/31198
+            # Added below scheduler_job_id condition to skip the events of pods created by other schedulers
+            labels = task.metadata.labels
+            if labels.get("airflow-worker", None) != scheduler_job_id:
+                last_resource_version = task.metadata.resource_version
+                continue
             annotations = task.metadata.annotations
             task_instance_related_annotations = {
                 "dag_id": annotations["dag_id"],

tests/executors/test_kubernetes_executor.py

@@ -1225,7 +1225,7 @@ def setup_method(self):
                 annotations={"airflow-worker": "bar", **self.core_annotations},
                 namespace="airflow",
                 resource_version="456",
-                labels={},
+                labels={"airflow-worker": "123"},
             ),
             status=k8s.V1PodStatus(phase="Pending"),
         )
@@ -1326,6 +1326,26 @@ def test_process_status_catchall(self):
         self._run()
         self.watcher.watcher_queue.put.assert_not_called()
 
+    @mock.patch.object(KubernetesJobWatcher, "process_status")
+    def test_process_status_called_for_matching_scheduler_id(self, mock_process_status):
+        self.pod.status.phase = "Running"
+        self.pod.metadata.labels = {"airflow-worker": "123"}
+        self.events.append({"type": "DELETED", "object": self.pod})
+
+        self._run()
+        mock_process_status.assert_called()
+
+    @mock.patch.object(KubernetesJobWatcher, "process_status")
+    def test_process_status_never_called_for_not_matching_scheduler_id(self, mock_process_status):
+        self.pod.status.phase = "Running"
+        self.pod.metadata.labels = {"airflow-worker": "124"}
+        self.events.append({"type": "DELETED", "object": self.pod})
+
+        self._run()
+        mock_process_status.assert_not_called()
+
+        self.pod.metadata.labels = {"airflow-worker": "123"}
+
     @mock.patch.object(KubernetesJobWatcher, "process_error")
     def test_process_error_event_for_410(self, mock_process_error):
         message = "too old resource version: 27272 (43334)"