TriggerDagRunOperator fails silently on 404 (DAG not found): on_failure_callback skipped and retries dropped

[manipatnam]

Under which category would you file this issue?

Task SDK

Apache Airflow version

Airflow 3.1.0

What happened and how to reproduce it?

When TriggerDagRunOperator attempts to trigger a non-existent DAG ID on Airflow 3.x (Task SDK architecture):

1. on_failure_callback / on_retry_callback are never executed.
2. Retries are completely ignored. Even if retries=3 is configured on the task, the task immediately fails on attempt 1/4 without attempting any retries.

Reproduction DAG

from datetime import datetime
from airflow.sdk import dag
from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator


def on_failure_cb(context):
    print("Failure callback executed!")


@dag(
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
)
def trigger_missing_dag():
    TriggerDagRunOperator(
        task_id="trigger_nonexistent",
        trigger_dag_id="this_dag_does_not_exist",
        retries=3,
        on_failure_callback=on_failure_cb,
    )


trigger_missing_dag()

What you think should happen instead?

Expected Behavior

1. If retries > 0, on_retry_callback (if defined) should run, and the task instance state should transition to UP_FOR_RETRY.
2. Once retries are exhausted, on_failure_callback should run, and the task instance state should transition to FAILED.

Actual Behavior

1. The task fails immediately on attempt 1/4 (marking state as FAILED in DB/UI).
2. on_failure_callback / on_retry_callback are never executed.

Operating System

No response

Deployment

Astronomer

Apache Airflow Provider(s)

standard

Versions of Apache Airflow Providers

No response

Official Helm Chart version

Not Applicable

Kubernetes Version

No response

Helm Chart configuration

No response

Docker Image customizations

No response

Anything else?

Technical Analysis & Root Cause

In airflow/sdk/execution_time/task_runner.py:

1. TriggerDagRunOperator.execute() raises DagRunTriggerException.
2. run() catches DagRunTriggerException and delegates to _handle_trigger_dag_run().
3. _handle_trigger_dag_run() calls SUPERVISOR_COMMS.send(TriggerDagRun(...)).
4. When the API server returns 404 Not Found, the supervisor returns an ErrorResponse(error=API_SERVER_ERROR, ...).
5. CommsDecoder._from_frame() receives this error and raises AirflowRuntimeError.
6. Exception Propagation Bug: Because AirflowRuntimeError is raised inside the except DagRunTriggerException handler block, Python propagates the exception out of run() without evaluating sibling except clauses in the same try block.
7. As a result:
   • _handle_current_task_failed(ti) (which checks task retries) is never called.
   • finalize() (which fires callbacks) is never called.
   • main() catches the unhandled exception and exits with code 1.
8. The Supervisor sees exit code 1, defaults final_state to TaskInstanceState.FAILED, and calls task_instances.finish(state=FAILED) on the API server.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct