You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "itz-puneet (via GitHub)" <gi...@apache.org> on 2026/09/06 23:29:23 UTC
[PR] Fail KubernetesPodOperator when on_kill interrupted the pod [airflow]
itz-puneet opened a new pull request, #72606:
URL: https://github.com/apache/airflow/pull/72606
closes: #71202
### The bug
Under `KubernetesExecutor` there are two pods: the per-task pod the executor creates, and the child
pod `KubernetesPodOperator.execute()` creates for the user's workload. When both are interrupted
within about a second of each other, the task instance is recorded as **success** even though the
child pod never finished. A second user confirmed the same behaviour on their production clusters in
the issue thread.
The sequence:
1. SIGTERM reaches the task process, and the runner calls `on_kill()`.
2. `on_kill()` sets `self._killed` and deletes the child pod.
3. The wait inside `execute_sync` — `await_pod_completion`, following container logs — returns
*normally* rather than raising, because the pod it was watching has simply gone away.
4. The `finally` block runs `post_complete_action` → `cleanup`, and `cleanup` returns early on
`self._killed` before reaching any of its failure signalling.
5. `execute_sync` falls through to its `return`, so the operator reports success.
`cleanup`'s early return is correct on its own terms — it exists so the pod is not deleted a second
time, since a duplicate delete can raise and trigger a spurious retry. The gap is that skipping the
deletion also skips the only place that would have failed the task.
### The fix
Raise from `execute_sync` when the body completed while `_killed` is set.
The placement matters and is the reason this is not done inside `cleanup`: `cleanup` is called from a
`finally` block, so raising there would replace an exception that is already propagating. When the
body failed on its own, that failure is what the operator should surface — it already fails the task,
with a more specific reason. Putting the check after the `finally` means it only fires on the path
that would otherwise have returned success. `cleanup` keeps its early return unchanged, so the
double-delete it guards against still cannot happen.
### Tests
Two regression tests in `test_pod.py`:
- `test_execute_sync_fails_when_on_kill_ran_during_the_wait` — drives the exact sequence above, with
the wait returning normally after `on_kill()` fires. Fails on `main` (the call returns instead of
raising), passes with this change.
- `test_execute_sync_keeps_the_original_error_when_on_kill_ran` — the body raises after `on_kill()`;
asserts the original exception still propagates rather than being replaced. This is what pins the
placement of the check.
Neither test needs a cluster.
---
**Gen-AI disclosure:** I used an AI coding assistant while working on this. I traced the interrupted
path through `execute_sync`, `cleanup` and `on_kill` myself, modelled the control flow before and
after the change to confirm the success path is the only one that changes behaviour, and reviewed
the final diff before submitting.
--
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
Re: [PR] Fail KubernetesPodOperator when on_kill interrupted the pod [airflow]
Posted by "flyovers-bot (via GitHub)" <gi...@apache.org>.
flyovers-bot commented on PR #72606:
URL: https://github.com/apache/airflow/pull/72606#issuecomment-5577263177
<!-- flyovers:flyover -->
[](https://flyovers.dev/flyover/633561c1-78ed-47a3-b4f3-b676a4be62c0)
**[Watch the flyover →](https://flyovers.dev/flyover/633561c1-78ed-47a3-b4f3-b676a4be62c0)** · 3:43 · 6 chapters · 2 files
An animated walkthrough of this diff: context, the problem, the change, implementation care, code walkthrough, and impact.
<sub>aa59bcf · updated automatically by flyovers</sub>
--
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
Re: [PR] Fail KubernetesPodOperator when on_kill interrupted the pod [airflow]
Posted by "itz-puneet (via GitHub)" <gi...@apache.org>.
itz-puneet commented on code in PR #72606:
URL: https://github.com/apache/airflow/pull/72606#discussion_r3945645816
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -886,6 +886,17 @@ def execute_sync(self, context: Context):
pod=pod_to_clean, remote_pod=self.remote_pod, context=context, result=result
)
+ if self._killed:
+ # on_kill() ran while the block above was waiting on the pod, and that wait
+ # returned normally because the pod it was watching simply went away. The
+ # workload never finished, so falling through here would finalise the task
+ # instance as success. This check sits after the finally block on purpose: if
+ # the body raised, that exception propagates untouched and already fails the
+ # task with its own reason.
+ raise AirflowException(
+ f"Pod {self.pod and self.pod.metadata.name} was interrupted before it completed."
+ )
Review Comment:
Good catch — the hook is real and both files were over their cap (pod.py allows 3 and had 4; the test module allows none). Added a dedicated `PodInterrupted(AirflowException)` alongside the other pod exceptions in aa59bcf, and the message now falls back to `self.name` so it cannot read "Pod None".
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##########
@@ -823,6 +823,89 @@ def test_process_pod_deletion(self, delete_pod_mock, on_finish_action, pod_phase
assert result == should_delete
+ def _interrupted_pod_operator(self, **kwargs):
+ return 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,
+ **kwargs,
+ )
+
+ @staticmethod
+ def _running_pod():
+ pod = MagicMock()
+ pod.metadata.name = "sleep-worker"
+ pod.metadata.namespace = "default"
+ pod.status.phase = PodPhase.RUNNING
+ return pod
+
+ @patch(HOOK_CLASS, new=MagicMock)
+ @patch(KUB_OP_PATH.format("get_or_create_pod"))
+ @patch(KUB_OP_PATH.format("find_pod"))
+ @patch(KUB_OP_PATH.format("await_pod_completion"))
+ def test_execute_sync_fails_when_on_kill_ran_during_the_wait(
+ self, await_pod_completion_mock, find_pod_mock, get_or_create_pod_mock
+ ):
+ """A pod interrupted by on_kill must not finalise the task instance as success.
+
+ Under KubernetesExecutor the task pod and the KPO child pod can be interrupted
+ within about a second of each other. SIGTERM reaches the task process, the runner
+ calls on_kill(), which deletes the child, and the wait in execute_sync then returns
+ normally because the pod it was watching has gone away. cleanup() skips its usual
+ failure signalling once _killed is set, so execute_sync used to fall through and
+ return, and the task was recorded as success (apache/airflow#71202).
Review Comment:
Removed the issue reference from the docstring in aa59bcf — it lives in the commit message and PR description instead.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##########
@@ -823,6 +823,89 @@ def test_process_pod_deletion(self, delete_pod_mock, on_finish_action, pod_phase
assert result == should_delete
+ def _interrupted_pod_operator(self, **kwargs):
+ return 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,
+ **kwargs,
+ )
+
+ @staticmethod
+ def _running_pod():
+ pod = MagicMock()
+ pod.metadata.name = "sleep-worker"
+ pod.metadata.namespace = "default"
+ pod.status.phase = PodPhase.RUNNING
+ return pod
+
+ @patch(HOOK_CLASS, new=MagicMock)
+ @patch(KUB_OP_PATH.format("get_or_create_pod"))
+ @patch(KUB_OP_PATH.format("find_pod"))
+ @patch(KUB_OP_PATH.format("await_pod_completion"))
+ def test_execute_sync_fails_when_on_kill_ran_during_the_wait(
+ self, await_pod_completion_mock, find_pod_mock, get_or_create_pod_mock
+ ):
+ """A pod interrupted by on_kill must not finalise the task instance as success.
+
+ Under KubernetesExecutor the task pod and the KPO child pod can be interrupted
+ within about a second of each other. SIGTERM reaches the task process, the runner
+ calls on_kill(), which deletes the child, and the wait in execute_sync then returns
+ normally because the pod it was watching has gone away. cleanup() skips its usual
+ failure signalling once _killed is set, so execute_sync used to fall through and
+ return, and the task was recorded as success (apache/airflow#71202).
+ """
+ k = self._interrupted_pod_operator()
+ running_pod = self._running_pod()
+ get_or_create_pod_mock.return_value = running_pod
+ find_pod_mock.return_value = running_pod
+ self.await_pod_mock.return_value = running_pod
+
+ # The wait returns rather than raising: the log stream simply ended when the
+ # child pod was deleted out from under it.
+ await_pod_completion_mock.side_effect = lambda pod: k.on_kill()
+
+ context = create_context(k)
+ context["ti"].xcom_push = MagicMock()
+
+ with pytest.raises(AirflowException, match="was interrupted before it completed"):
Review Comment:
Updated in aa59bcf: the interrupted test now expects `PodInterrupted` by type, and the "body already failed" test raises `RuntimeError`, which also keeps that file clear of the AirflowException cap and makes the stronger point that any in-flight exception survives.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##########
@@ -823,6 +823,89 @@ def test_process_pod_deletion(self, delete_pod_mock, on_finish_action, pod_phase
assert result == should_delete
+ def _interrupted_pod_operator(self, **kwargs):
+ return 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,
+ **kwargs,
+ )
+
+ @staticmethod
+ def _running_pod():
+ pod = MagicMock()
+ pod.metadata.name = "sleep-worker"
+ pod.metadata.namespace = "default"
+ pod.status.phase = PodPhase.RUNNING
+ return pod
Review Comment:
Leaving this one as is: the surrounding module builds pod mocks with bare `MagicMock()` throughout, and `spec_set` on `V1Pod` plus its nested `status`/`metadata` objects would make these two tests inconsistent with the rest of the file. Happy to switch if a maintainer would rather standardise on it.
--
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
Re: [PR] Fail KubernetesPodOperator when on_kill interrupted the pod [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #72606:
URL: https://github.com/apache/airflow/pull/72606#discussion_r3945629636
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -886,6 +886,17 @@ def execute_sync(self, context: Context):
pod=pod_to_clean, remote_pod=self.remote_pod, context=context, result=result
)
+ if self._killed:
+ # on_kill() ran while the block above was waiting on the pod, and that wait
+ # returned normally because the pod it was watching simply went away. The
+ # workload never finished, so falling through here would finalise the task
+ # instance as success. This check sits after the finally block on purpose: if
+ # the body raised, that exception propagates untouched and already fails the
+ # task with its own reason.
+ raise AirflowException(
+ f"Pod {self.pod and self.pod.metadata.name} was interrupted before it completed."
+ )
Review Comment:
This adds a new direct `raise AirflowException(...)`, which is disallowed by Airflow's "no new AirflowException" rule (enforced by prek). Use a standard exception (or a dedicated provider exception) instead, and avoid emitting `Pod None` when `self.pod` is unset.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##########
@@ -823,6 +823,89 @@ def test_process_pod_deletion(self, delete_pod_mock, on_finish_action, pod_phase
assert result == should_delete
+ def _interrupted_pod_operator(self, **kwargs):
+ return 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,
+ **kwargs,
+ )
+
+ @staticmethod
+ def _running_pod():
+ pod = MagicMock()
+ pod.metadata.name = "sleep-worker"
+ pod.metadata.namespace = "default"
+ pod.status.phase = PodPhase.RUNNING
+ return pod
Review Comment:
New `MagicMock()` instances are created without a spec, which can mask attribute/typing mistakes in these tests. Consider using `spec_set` for the mocked pod and its nested objects so unexpected attribute access fails fast.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##########
@@ -823,6 +823,89 @@ def test_process_pod_deletion(self, delete_pod_mock, on_finish_action, pod_phase
assert result == should_delete
+ def _interrupted_pod_operator(self, **kwargs):
+ return 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,
+ **kwargs,
+ )
+
+ @staticmethod
+ def _running_pod():
+ pod = MagicMock()
+ pod.metadata.name = "sleep-worker"
+ pod.metadata.namespace = "default"
+ pod.status.phase = PodPhase.RUNNING
+ return pod
+
+ @patch(HOOK_CLASS, new=MagicMock)
+ @patch(KUB_OP_PATH.format("get_or_create_pod"))
+ @patch(KUB_OP_PATH.format("find_pod"))
+ @patch(KUB_OP_PATH.format("await_pod_completion"))
+ def test_execute_sync_fails_when_on_kill_ran_during_the_wait(
+ self, await_pod_completion_mock, find_pod_mock, get_or_create_pod_mock
+ ):
+ """A pod interrupted by on_kill must not finalise the task instance as success.
+
+ Under KubernetesExecutor the task pod and the KPO child pod can be interrupted
+ within about a second of each other. SIGTERM reaches the task process, the runner
+ calls on_kill(), which deletes the child, and the wait in execute_sync then returns
+ normally because the pod it was watching has gone away. cleanup() skips its usual
+ failure signalling once _killed is set, so execute_sync used to fall through and
+ return, and the task was recorded as success (apache/airflow#71202).
+ """
+ k = self._interrupted_pod_operator()
+ running_pod = self._running_pod()
+ get_or_create_pod_mock.return_value = running_pod
+ find_pod_mock.return_value = running_pod
+ self.await_pod_mock.return_value = running_pod
+
+ # The wait returns rather than raising: the log stream simply ended when the
+ # child pod was deleted out from under it.
+ await_pod_completion_mock.side_effect = lambda pod: k.on_kill()
+
+ context = create_context(k)
+ context["ti"].xcom_push = MagicMock()
+
+ with pytest.raises(AirflowException, match="was interrupted before it completed"):
Review Comment:
This assertion expects `AirflowException`, but the operator-side fix should not introduce a new direct `raise AirflowException(...)`. If the operator is switched to a standard exception (e.g. `RuntimeError`), update this test accordingly so it fails on main and passes with the fix.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py:
##########
@@ -823,6 +823,89 @@ def test_process_pod_deletion(self, delete_pod_mock, on_finish_action, pod_phase
assert result == should_delete
+ def _interrupted_pod_operator(self, **kwargs):
+ return 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,
+ **kwargs,
+ )
+
+ @staticmethod
+ def _running_pod():
+ pod = MagicMock()
+ pod.metadata.name = "sleep-worker"
+ pod.metadata.namespace = "default"
+ pod.status.phase = PodPhase.RUNNING
+ return pod
+
+ @patch(HOOK_CLASS, new=MagicMock)
+ @patch(KUB_OP_PATH.format("get_or_create_pod"))
+ @patch(KUB_OP_PATH.format("find_pod"))
+ @patch(KUB_OP_PATH.format("await_pod_completion"))
+ def test_execute_sync_fails_when_on_kill_ran_during_the_wait(
+ self, await_pod_completion_mock, find_pod_mock, get_or_create_pod_mock
+ ):
+ """A pod interrupted by on_kill must not finalise the task instance as success.
+
+ Under KubernetesExecutor the task pod and the KPO child pod can be interrupted
+ within about a second of each other. SIGTERM reaches the task process, the runner
+ calls on_kill(), which deletes the child, and the wait in execute_sync then returns
+ normally because the pod it was watching has gone away. cleanup() skips its usual
+ failure signalling once _killed is set, so execute_sync used to fall through and
+ return, and the task was recorded as success (apache/airflow#71202).
Review Comment:
Test docstrings should not include issue numbers; this docstring embeds `apache/airflow#71202`, which violates the project testing guidelines.
--
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