From dacae2ed24c259876d4a20481e2850efdee2d6ca Mon Sep 17 00:00:00 2001 From: Sankalp Thakur Date: Thu, 13 Aug 2026 13:07:38 +0530 Subject: [PATCH 1/2] fix: fail KubernetesPodOperator when on_kill interrupts the child pod cleanup() returned early after on_kill(), so execute() looked like a clean success when SIGTERM hit the task process (spot drain / dual interrupt) even though the child pod never finished. Fixes #71202 Signed-off-by: Sankalp Thakur --- .../cncf/kubernetes/operators/pod.py | 16 ++++-- .../cncf/kubernetes/operators/test_pod.py | 53 +++++++++++++++++++ 2 files changed, 64 insertions(+), 5 deletions(-) diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py index 5a9df5645aaa8..a0affcafeea38 100644 --- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py +++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py @@ -1332,11 +1332,17 @@ 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 AirflowException( + 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) diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py index d259d87c6238a..2248bfffad877 100644 --- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py +++ b/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") From 4f8b7a59f7624fbe3b5a811abe00d3abdc8f0d71 Mon Sep 17 00:00:00 2001 From: Sankalp Thakur Date: Mon, 17 Aug 2026 13:32:50 +0530 Subject: [PATCH 2/2] Use PodInterrupted when KubernetesPodOperator cleanup sees on_kill check-no-new-airflow-exceptions rejects a fourth raise AirflowException in pod.py. ruff-format also wants the raise on a single line. Signed-off-by: Sankalp Thakur --- .../airflow/providers/cncf/kubernetes/operators/pod.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py index a0affcafeea38..5f45b2a63fc51 100644 --- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py +++ b/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. @@ -1337,9 +1341,7 @@ def cleanup( # retry. Do not return as success either: the workload did not finish. # See https://github.com/apache/airflow/issues/71202. if self._killed: - raise AirflowException( - f"Pod {pod and pod.metadata.name} was interrupted before it completed." - ) + 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: