Fail a task whose execute is interrupted by SIGTERM - #73809

Opened fresh from #73395, which was closed under the new open-PR-limit policy and then failed to reopen (both gh pr reopen and the UI button errored with a generic "Could not open the pull request") — following @potiuk's guidance on Slack to push/open fresh from the branch when reopening doesn't work.

closes: #73006

What happened

The task runner installs a SIGTERM handler that calls on_kill() and returns:

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?

• Yes

Generative AI tooling was used while authoring this PR, following the guidelines. Every claim above — the reproduction, the ablation and the before/after numbers — was produced by running the tests, not by the model asserting them.