Respect task retries for signal killed tasks

[amoghrajesh]

closes: #55753

Problem

When tasks are killed by system signals (SIGKILL for OOM, SIGTERM for worker restarts), they immediately go to FAILED state instead of respecting the task retries set and going to UP_FOR_RETRY state. This creates unexpected behavior where exception based failures respect retries but signal based failures don't.

Root Cause

The supervisor's final_state property only checked the exit code and didn't consider retry eligibility for signal based failures. While exception based failures properly checked should_retry which is set by the API server when a task is run for the first time a.k.a, in its run context, signal based failures ignored this logic entirely.

Testing

DAG used:

from airflow import DAG
from datetime import datetime

from airflow.providers.standard.operators.python import PythonOperator


def func():
    a = "asd"
    while True:
        a += a*100000

with DAG(
    dag_id="oom_example",
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
    doc_md=__doc__,
    tags=["oom"],
) as dag:
    hello_task = PythonOperator(
        task_id="oom_task",
        python_callable=func,
        retries=3
    )

Earlier:

The dag immediately moved into FAILED state without any retries;

After:

The dag moves into up for retry state as appropriate:

See retries:

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[ashb]

SIGABRT and SIGSEGV should not retry

Why? That's not what I would expect to do (and also I suspect not what Airflow 2 did?)

[kaxil]

Update: Nvm, thanks to @rawwar figured out that K8s/cgroups has configs to kill only the process

---

When tasks are killed by system signals (SIGKILL for OOM, SIGTERM for worker restarts), they immediately go to FAILED state instead of respecting the task retries set and going to UP_FOR_RETRY state. This creates unexpected behavior where exception based failures respect retries but signal based failures don't.

How common is this scenario (excluding manually killing task process)? Since the supervisor and task processes are running in the same container, wouldn't an OOM condition typically kill the entire container rather than just the individual task process?

In the more common case where the entire container gets OOM-killed:

1. The supervisor process would also die
2. Heartbeat to the scheduler would fail
3. Scheduler would receive a FAILED executor event and handle retries through the normal process_executor_events() → handle_failure() path

[ashb]

@kaxil Also people do occasionally run Airflow outside of Kubernetes you know 😉

[amoghrajesh]

I'll get to the comments on this later, not needed for 3.1, can wait till 3.1.1

[amoghrajesh]

@ashb replied to some of your comments, could you take a look when possible?

[rawwar]

Verified by running a task that gets killed due to oom and the task went to "up_for_retry"

Just wanted to let it run all tries. It did fail at the end:

[rawwar]

Thanks!

[github-actions]

Backport successfully created: v3-1-test

Status	Branch	Result
✅	v3-1-test