namanjain24-sudo opened a new pull request, #73395:
URL: https://github.com/apache/airflow/pull/73395
`closes: #73006`
## What happened
The task runner installs a `SIGTERM` handler that calls `on_kill()` and
returns:
```python
def _on_term(signum, frame):
pid = os.getpid()
if pid != parent_pid:
return
ti.task.on_kill()
```
Nothing raises, so `execute()` resumes from wherever the signal interrupted
it and runs
to its end. For an operator whose `on_kill` destroys the work it is waiting
on, the
operator then returns normally and the task instance is committed
**`success`** even
though the work never happened.
`KubernetesPodOperator` is the reported case: `on_kill` deletes the pod, and
`cleanup()`
— the only place that raises for a pod that did not reach `Succeeded` —
returns early
because `_killed` is set. That guard comes from #36749, whose premise is
that something
else is already marking the task failed. That holds when a user marks a task
failed in
the UI. It does not hold when the `SIGTERM` comes from a pod eviction, a
drained node or
a restarted worker, so nothing fails the task at all.
`AirflowTaskTerminated` is already caught in `_run_task_and_map_outcome`,
but grepping
the tree shows it is never raised anywhere — the raise was dropped when the
handler was
ported in #50141, and only the `except` survived.
Reproduced on `main` at `61db43fba9` with the operator shape from the issue:
the task
instance ends `state=TaskInstanceState.SUCCESS` and `execute()` is observed
running past
the point where the signal arrived.
## What is in this PR
`_on_term` raises `AirflowTaskTerminated` after `on_kill()`, with two guards:
- **Only while the task's own body is on the stack.** The supervisor also
sends `SIGTERM`
once a task has already reported success and is sitting in
`task_success_overtime`, and
also when the server says the task should no longer be running. Raising
there would turn
a finished task into a failed one and lose a result the server has already
accepted, so
outside that window the handler keeps today's behaviour exactly. The
window is closed in
a `finally`, so the paths that leave `_execute_task` by raising a
`BaseException`
— `AirflowTaskTimeout`, `SystemExit` — leave it too.
- **Once.** A second `SIGTERM` arriving while the first is still unwinding
is ignored.
Without this, an `on_kill` that is itself interrupted re-enters the
handler: the test
for it records **478** `on_kill` calls on `main` against 1 with this
change.
An `on_kill` that raises no longer hides the termination either — the raise
is in a
`finally`, so the task still ends as `AirflowTaskTerminated` rather than as
whatever
`on_kill` failed with.
This is deliberately in the task runner rather than in
`KubernetesPodOperator`. The
missing raise is the regression, and it affects any operator whose `on_kill`
tears down
the work in flight; fixing `cleanup()` alone would leave that hole open for
the rest.
## Testing
`TestSigtermTerminatesTask` in
`task-sdk/tests/task_sdk/execution_time/test_task_runner.py`,
six cases: a `SIGTERM` mid-`execute`; the `KubernetesPodOperator` shape,
with an `on_kill`
that sets `_killed` and a `cleanup()` that therefore declines to raise; a
second `SIGTERM`
delivered from inside `on_kill`; an `on_kill` that raises; and two negative
controls — a
`SIGTERM` delivered after `execute()` has returned, which must stay
`success`, and a task
that gets no signal at all.
Ablation, with only `task_runner.py` reverted to `main` and the tests kept
(they live in a
different file, so the revert does not delete them): the four positive cases
fail and the
two controls pass. On `main` the first case reports `state=SUCCESS` with
`execute()` having
run to its end, the pod-shaped one reports `SUCCESS`, and the re-signalling
one reports 478
`on_kill` calls.
`test_task_runner.py` and `test_supervisor.py` were run with and without the
change and the
failure sets compared: **no regressions**. 15 tests fail identically either
way in this
local environment (8 `TestTriggerDagRunOperator`, 7 in
`test_supervisor.py`); the only
difference between the two runs is the four new tests.
`ruff check`, `ruff format --check` and `mypy` are clean on both changed
files. Three
pre-push hooks could not run locally — `generate-openapi-spec`,
`check-default-configuration`
and provider `mypy` all need a Breeze CI image, and the local Docker daemon
fails to build
one; they touch no file in this PR and will run in CI.
## Behaviour change
A task whose process is sent `SIGTERM` while `execute()` is running now ends
**failed**
instead of finishing normally. That is the point of the fix, but it is a
visible change for
anyone whose tasks were quietly being marked successful after an eviction.
`AirflowTaskTerminated` is classified as terminal today, so such a task
fails without
consuming a retry. Whether an infrastructure kill should instead be
retry-eligible is a
real question — it is what #73238 is about — but changing that
classification also changes
the UI "mark failed" path, so it is left alone here.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes
Generative AI tooling was used while authoring this PR, following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions).
Every claim above — the reproduction, the ablation and the before/after
numbers — was produced by running the tests, not by the model asserting them.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
