Find Pod Before Cleanup In KubernetesPodOperator Execution

airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py

@@ -400,6 +400,8 @@ def execute(self, context: 'Context'):
                 pod_request_obj=self.pod_request_obj,
                 context=context,
             )
+            # finding remote pod to ensure it exists
+            remote_pod = self.find_pod(self.pod.metadata.namespace, context=context)
             self.await_pod_start(pod=self.pod)
 
             if self.get_logs:
@@ -437,16 +439,18 @@ def cleanup(self, pod: k8s.V1Pod, remote_pod: k8s.V1Pod):
                 with _suppress(Exception):
                     for event in self.pod_manager.read_pod_events(pod).items:
                         self.log.error("Pod Event: %s - %s", event.reason, event.message)
-            with _suppress(Exception):
-                self.process_pod_deletion(pod)
+            if remote_pod is not None:
+                with _suppress(Exception):
+                    self.process_pod_deletion(pod)
             error_message = get_container_termination_message(remote_pod, self.BASE_CONTAINER_NAME)
             error_message = "\n" + error_message if error_message else ""
             raise AirflowException(
                 f'Pod {pod and pod.metadata.name} returned a failure:{error_message}\n{remote_pod}'
             )
         else:
-            with _suppress(Exception):
-                self.process_pod_deletion(pod)
+            if remote_pod is not None:
+                with _suppress(Exception):
+                    self.process_pod_deletion(pod)
 
     def process_pod_deletion(self, pod):
         if self.is_delete_operator_pod:

airflow/providers/cncf/kubernetes/utils/pod_manager.py

@@ -87,7 +87,7 @@ def get_container_termination_message(pod: V1Pod, container_name: str):
         container_statuses = pod.status.container_statuses
         container_status = next(iter([x for x in container_statuses if x.name == container_name]), None)
         return container_status.state.terminated.message if container_status else None
-    except AttributeError:
+    except (AttributeError, TypeError):
         return None
 
 

kubernetes_tests/test_kubernetes_pod_operator.py

@@ -660,7 +660,7 @@ def test_envs_from_secrets(self, await_pod_completion_mock, create_pod):
             do_xcom_push=False,
         )
         # THEN
-        await_pod_completion_mock.return_value = None
+        await_pod_completion_mock.side_effect = AirflowException
         context = create_context(k)
         with pytest.raises(AirflowException):
             k.execute(context)

tests/providers/cncf/kubernetes/operators/test_kubernetes_pod.py

@@ -266,7 +266,10 @@ def test_image_pull_policy_correctly_set(self):
         assert pod.spec.containers[0].image_pull_policy == "Always"
 
     @mock.patch("airflow.providers.cncf.kubernetes.utils.pod_manager.PodManager.delete_pod")
-    def test_pod_delete_even_on_launcher_error(self, delete_pod_mock):
+    @mock.patch("airflow.providers.cncf.kubernetes.operators.kubernetes_pod.KubernetesPodOperator.find_pod")
+    def test_pod_delete_even_on_launcher_error(self, find_pod_mock, delete_pod_mock):
+        remote_pod_mock = MagicMock()
+        find_pod_mock.return_value = remote_pod_mock
         k = KubernetesPodOperator(
             namespace="default",
             image="ubuntu:16.04",
@@ -286,6 +289,30 @@ def test_pod_delete_even_on_launcher_error(self, delete_pod_mock):
             k.execute(context=context)
         assert delete_pod_mock.called
 
+    @mock.patch("airflow.providers.cncf.kubernetes.utils.pod_manager.PodManager.delete_pod")
+    @mock.patch("airflow.providers.cncf.kubernetes.operators.kubernetes_pod.KubernetesPodOperator.find_pod")
+    def test_pod_not_deleting_non_existing_pod(self, find_pod_mock, delete_pod_mock):
+
+        find_pod_mock.return_value = None
+        k = KubernetesPodOperator(
+            namespace="default",
+            image="ubuntu:16.04",
+            cmds=["bash", "-cx"],
+            arguments=["echo 10"],
+            labels={"foo": "bar"},
+            name="test",
+            task_id="task",
+            in_cluster=False,
+            do_xcom_push=False,
+            cluster_context="default",
+            is_delete_operator_pod=True,
+        )
+        self.create_mock.side_effect = AirflowException("fake failure")
+        with pytest.raises(AirflowException):
+            context = create_context(k)
+            k.execute(context=context)
+        delete_pod_mock.assert_not_called()
+
     @pytest.mark.parametrize('randomize', [True, False])
     def test_provided_pod_name(self, randomize):
         name_base = "test"
@@ -790,7 +817,6 @@ def test_previous_pods_ignored_for_reattached(self):
             task_id="task",
         )
         self.run_pod(k)
-        k.client.list_namespaced_pod.assert_called_once()
         _, kwargs = k.client.list_namespaced_pod.call_args
         assert 'already_checked!=True' in kwargs['label_selector']