What happened
When a deferrable operator's trigger yields a terminal TaskFailedEvent (a BaseTaskEndEvent with task_instance_state=FAILED) to fail a task without resuming it on a worker, the task is always marked failed terminally and its on_failure_callback runs, even when the task still has retries remaining.
A task that fails while running on a worker with retries left goes to up_for_retry and runs on_retry_callback. A task that fails through its trigger does not. Deferrable tasks that fail via the trigger silently lose their retries and emit a false on_failure_callback.
What you think should happen instead
A trigger-emitted failure should respect the task's retry configuration, the same way a worker-side failure does: if the task is eligible to retry, go up_for_retry and run on_retry_callback; only when retries are exhausted, fail terminally and run on_failure_callback.
How to reproduce
Defer a task with retries=1 whose trigger yields TaskFailedEvent(). It goes straight to failed and on_failure_callback fires on the first attempt, instead of retrying.
The cause is the BaseTaskEndEvent handler in airflow-core/src/airflow/models/trigger.py, which calls set_state(event.task_instance_state) and sends the callback with task_callback_type=event.task_instance_state without any retry-eligibility check.
Context
The scheduler executor-event path already routes by retry-eligibility (#56586) and the DAG processor honors task_callback_type (UP_FOR_RETRY -> on_retry_callback). The trigger end-event path was not updated to match.
Are you willing to submit a PR?
Yes
What happened
When a deferrable operator's trigger yields a terminal
TaskFailedEvent(aBaseTaskEndEventwithtask_instance_state=FAILED) to fail a task without resuming it on a worker, the task is always markedfailedterminally and itson_failure_callbackruns, even when the task still has retries remaining.A task that fails while running on a worker with retries left goes to
up_for_retryand runson_retry_callback. A task that fails through its trigger does not. Deferrable tasks that fail via the trigger silently lose their retries and emit a falseon_failure_callback.What you think should happen instead
A trigger-emitted failure should respect the task's retry configuration, the same way a worker-side failure does: if the task is eligible to retry, go
up_for_retryand runon_retry_callback; only when retries are exhausted, fail terminally and runon_failure_callback.How to reproduce
Defer a task with
retries=1whose trigger yieldsTaskFailedEvent(). It goes straight tofailedandon_failure_callbackfires on the first attempt, instead of retrying.The cause is the
BaseTaskEndEventhandler inairflow-core/src/airflow/models/trigger.py, which callsset_state(event.task_instance_state)and sends the callback withtask_callback_type=event.task_instance_statewithout any retry-eligibility check.Context
The scheduler executor-event path already routes by retry-eligibility (#56586) and the DAG processor honors
task_callback_type(UP_FOR_RETRY -> on_retry_callback). The trigger end-event path was not updated to match.Are you willing to submit a PR?
Yes