Fix KubernetesPodOperator succeeding after SIGTERM/on_kill interrupt

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py

@@ -139,6 +139,10 @@ class FoundMoreThanOnePodFailure(AirflowException):
     """When during reconnect more than one matching pod was found."""
 
 
+class PodInterrupted(AirflowException):
+    """When on_kill already deleted the child pod before it completed."""
+
+
 class KubernetesPodOperator(BaseOperator):
     """
     Execute a task in a Kubernetes Pod.
@@ -1332,11 +1336,15 @@ def cleanup(
         xcom_result: dict | None = None,
         context: Context | None = None,
     ) -> None:
-        # Skip cleaning the pod in the following scenarios.
-        # 1. If a task got marked as failed, "on_kill" method would be called and the pod will be cleaned up
-        # there. Cleaning it up again will raise an exception (which might cause retry).
-        # 2. remote pod is null (ex: pod creation failed)
-        if self._killed or not remote_pod:
+        # If on_kill already ran (SIGTERM / task abort), the child pod was deleted there.
+        # Do not try to delete it again: a 404 from a second delete can cause a spurious
+        # retry. Do not return as success either: the workload did not finish.
+        # See https://github.com/apache/airflow/issues/71202.
+        if self._killed:
+            raise PodInterrupted(f"Pod {pod and pod.metadata.name} was interrupted before it completed.")
+        # remote pod is null (ex: pod creation failed). The original exception from
+        # create/await still propagates from execute_sync's try block.
+        if not remote_pod:
             return
 
         istio_enabled = self.is_istio_enabled(remote_pod)

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py

@@ -823,6 +823,41 @@ def test_process_pod_deletion(self, delete_pod_mock, on_finish_action, pod_phase
 
         assert result == should_delete
 
+    @patch(HOOK_CLASS, new=MagicMock)
+    @patch(KUB_OP_PATH.format("find_pod"))
+    @patch(KUB_OP_PATH.format("await_pod_completion"))
+    def test_execute_sync_on_kill_during_wait_is_not_success(self, await_pod_completion_mock, find_pod_mock):
+        """SIGTERM/on_kill while waiting on the child must fail the task, not succeed."""
+        k = KubernetesPodOperator(
+            namespace="default",
+            image="ubuntu:16.04",
+            cmds=["bash", "-cx"],
+            arguments=["sleep 120"],
+            name="sleep-worker",
+            task_id="task",
+            do_xcom_push=False,
+            get_logs=True,
+            log_pod_spec_on_failure=False,
+        )
+        running = MagicMock()
+        running.metadata.name = "sleep-worker"
+        running.metadata.namespace = "default"
+        running.status.phase = PodPhase.RUNNING
+        find_pod_mock.return_value = running
+        self.create_mock.return_value = running
+        self.await_pod_mock.return_value = running
+
+        def _sigterm_during_log_follow(pod):
+            # Same as task_runner's SIGTERM handler: on_kill() then execution continues.
+            k.on_kill()
+
+        await_pod_completion_mock.side_effect = _sigterm_during_log_follow
+        context = create_context(k)
+        context["ti"].xcom_push = MagicMock()
+
+        with pytest.raises(AirflowException, match="interrupted before it completed"):
+            k.execute(context=context)
+
     @pytest.mark.parametrize(
         "pod_phase",
         [
@@ -3612,6 +3647,24 @@ def test_cleanup_log_pod_spec_on_failure(self, log_pod_spec_on_failure, expect_m
         with pytest.raises(AirflowException, match=expect_match):
             k.cleanup(pod, pod)
 
+    def test_cleanup_raises_when_on_kill_already_ran(self):
+        """on_kill deletes the child pod; cleanup must not then report success.
+
+        Dual interrupt (executor task pod + KPO child) delivers SIGTERM into the
+        task process. task_runner calls on_kill(), which sets ``_killed`` and
+        deletes the child. cleanup() used to return immediately in that case,
+        so execute() finished normally and the TI was marked SUCCESS even though
+        the workload never completed (apache/airflow#71202).
+        """
+        k = KubernetesPodOperator(task_id="task", name="sleep-worker", log_pod_spec_on_failure=False)
+        k._killed = True
+        pod = k8s.V1Pod(
+            metadata=k8s.V1ObjectMeta(name="sleep-worker", namespace="default"),
+            status=V1PodStatus(phase=PodPhase.SUCCEEDED),
+        )
+        with pytest.raises(AirflowException, match="interrupted before it completed"):
+            k.cleanup(pod, pod)
+
     @patch(KUB_OP_PATH.format("_write_logs"))
     @patch("airflow.providers.cncf.kubernetes.operators.pod.KubernetesPodOperator.cleanup")
     @patch("airflow.providers.cncf.kubernetes.operators.pod.KubernetesPodOperator.find_pod")