kaxil opened a new pull request, #70719: URL: https://github.com/apache/airflow/pull/70719
Fixes #70683 ## Problem When `TriggerDagRunOperator` targets a Dag that does not exist, the task dies without running `on_failure_callback` / `on_retry_callback`, without firing task-instance listeners, and without evaluating the task's `retry_policy`. To the user the task just goes red, and none of their failure handling runs. The cause is not specific to `TriggerDagRunOperator`. [`run()`](https://github.com/apache/airflow/blob/658bd8a288/task-sdk/src/airflow/sdk/execution_time/task_runner.py#L1571-L1693) has a single `try` with a flat list of `except` clauses, and several of those clauses call out to the API server. An exception raised *inside* an `except` clause is not offered to that clause's siblings, so it escapes `run()` entirely, skipping the retry decision in `_handle_current_task_failed()` and everything in `finalize()`. For the reported case the chain is: 1. The execution API returns 404 when no non-stale Dag matches ([`dag_runs.py#L107`](https://github.com/apache/airflow/blob/658bd8a288/airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py#L107)). 2. `DagRunOperations.trigger` only special-cases 409, so the 404 re-raises ([`client.py#L927`](https://github.com/apache/airflow/blob/658bd8a288/task-sdk/src/airflow/sdk/api/client.py#L927)). 3. The supervisor converts it to an `API_SERVER_ERROR` response ([`supervisor.py#L945`](https://github.com/apache/airflow/blob/658bd8a288/task-sdk/src/airflow/sdk/execution_time/supervisor.py#L945)). 4. `CommsDecoder._from_frame` raises `AirflowRuntimeError` ([`comms.py#L359`](https://github.com/apache/airflow/blob/658bd8a288/task-sdk/src/airflow/sdk/execution_time/comms.py#L359)). That raise happens inside [`except DagRunTriggerException`](https://github.com/apache/airflow/blob/658bd8a288/task-sdk/src/airflow/sdk/execution_time/task_runner.py#L1624), 39 lines above the [clause that already handles `AirflowRuntimeError`](https://github.com/apache/airflow/blob/658bd8a288/task-sdk/src/airflow/sdk/execution_time/task_runner.py#L1663). Python never consults it. ## Solution Wrap the handler chain in an outer `try` whose `except BaseException` routes the exception through `_handle_current_task_failed()`, the same path every other failure takes. The pre-existing `finally` block moves to the outer `try` so it still sends the terminal-state message after the new handler binds `msg` and `state`. Most of the diff is indentation. `git diff -w` shows the whole logical change: ```python + except BaseException as e: + # An exception raised *inside* one of the handlers above is not matched by + # its sibling ``except`` clauses, so it would escape ``run()`` entirely -- + # skipping the retry decision in ``_handle_current_task_failed`` and the + # callbacks/listeners in ``finalize()``. + log.exception("Task failed with exception") + msg, state = _handle_current_task_failed(ti, e, log, context) + error = e ``` ## Why fix the whole chain rather than the 404 Three other handlers can raise the same way, and all of them lose callbacks today: | Clause | What can raise | |---|---| | `except DownstreamTasksSkipped` | sends `SkipDownstreamTasks` to the API server | | `except DagRunTriggerException` | three comms sends plus an XCom push | | `except TaskDeferred` | `trigger.serialize()` runs user code, then `serde_serialize` | | `except TaskAwaitingInput` | `serde_serialize` | A 404-only fix would leave the other three broken. `_handle_current_task_failed()` does not talk to the API server, so it is safe to call from the outer handler. `BaseException` rather than `Exception` keeps the new clause consistent with the pre-existing inner `except BaseException`, which already converts a body-level `KeyboardInterrupt` into a task failure. Narrowing it would also stop catching `SystemExit`, which the chain explicitly handles. ## What changes for users On `main` today the supervisor's exit-code fallback already recovers `UP_FOR_RETRY` from the non-zero exit, so the task does still retry. What was being lost, and now works: - `on_failure_callback` and `on_retry_callback` run - `on_task_instance_failed` listeners fire - `email_on_failure` / `email_on_retry` are sent - A `retry_policy` is evaluated, so it can force `FAIL` or supply a custom delay and reason - The `RetryTask` message carries `end_date`, retry delay and retry reason instead of being skipped - `ti_failures`, `operator_failures` and `ti.finish` stats are recorded - `dag.test()` and `run_task_in_process` no longer crash out of the CLI The issue reports retries dropped entirely, which was accurate for 3.1.0: the `_should_retry` exit-code fallback landed later in #55767 and first shipped in 3.1.2. ## Gotchas If the *first* trigger attempt partially succeeds (the Dag run is created, then a later call in the same handler fails), the task is now retried where it previously died. On retry `TriggerDagRunOperator` mints a run id from `run_after or utcnow()` when no explicit `trigger_run_id` is set, so that retry can create a second Dag run. This behaviour is unchanged by this PR: the supervisor's exit-code fallback already produced the same retry on `main`. Pinning `trigger_run_id` makes the retry idempotent via the existing 409 handling. -- 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]
