You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "kaxil (via GitHub)" <gi...@apache.org> on 2026/07/30 07:46:32 UTC
[PR] Fix task callbacks being skipped when `TriggerDagRunOperator` gets a 404 [airflow]
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix task callbacks being skipped when `TriggerDagRunOperator` gets a 404 [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #70719:
URL: https://github.com/apache/airflow/pull/70719#discussion_r3685453869
##########
task-sdk/tests/task_sdk/execution_time/test_task_runner.py:
##########
@@ -980,6 +988,214 @@ def test_defer_task_queue_assignment(
)
+@pytest.mark.parametrize(
+ ("should_retry", "expected_state"),
+ [
+ (True, TaskInstanceState.UP_FOR_RETRY),
+ (False, TaskInstanceState.FAILED),
+ ],
+)
+def test_defer_with_unserializable_kwargs_honours_retries_and_callbacks(
+ should_retry, expected_state, create_runtime_ti, mock_supervisor_comms
+):
+ """
+ A task that defers with a non-serializable ``next_kwargs`` value must fail like any
+ other task, rather than taking the whole run down.
+
+ ``_defer_task`` runs ``serde_serialize`` on the deferral kwargs, which raises
+ ``TypeError`` for anything it has no serializer for (a file handle, a client object,
+ a lambda). That raise happens inside ``run()``'s ``except TaskDeferred`` handler, so
+ before the fix it escaped ``run()`` without evaluating retries or running callbacks --
+ the same defect as https://github.com/apache/airflow/issues/70683, reached without
+ involving the API server at all.
Review Comment:
Test docstring references a GitHub issue URL/number. Per Airflow testing guidelines, avoid issue numbers in test docstrings; keep the rationale but remove the issue identifier.
This issue also appears on line 5415 of the same file.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix task callbacks being skipped when `TriggerDagRunOperator` gets a 404 [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #70719:
URL: https://github.com/apache/airflow/pull/70719#discussion_r3710563913
##########
task-sdk/tests/task_sdk/execution_time/test_task_runner.py:
##########
@@ -980,6 +988,214 @@ def test_defer_task_queue_assignment(
)
+@pytest.mark.parametrize(
+ ("should_retry", "expected_state"),
+ [
+ (True, TaskInstanceState.UP_FOR_RETRY),
+ (False, TaskInstanceState.FAILED),
+ ],
+)
+def test_defer_with_unserializable_kwargs_honours_retries_and_callbacks(
+ should_retry, expected_state, create_runtime_ti, mock_supervisor_comms
+):
+ """
+ A task that defers with a non-serializable ``next_kwargs`` value must fail like any
+ other task, rather than taking the whole run down.
+
+ ``_defer_task`` runs ``serde_serialize`` on the deferral kwargs, which raises
+ ``TypeError`` for anything it has no serializer for (a file handle, a client object,
+ a lambda). That raise happens inside ``run()``'s ``except TaskDeferred`` handler, so
+ before the fix it escaped ``run()`` without evaluating retries or running callbacks --
+ the same defect as https://github.com/apache/airflow/issues/70683, reached without
+ involving the API server at all.
Review Comment:
+1
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix task callbacks being skipped when `TriggerDagRunOperator` gets a 404 [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 merged PR #70719:
URL: https://github.com/apache/airflow/pull/70719
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix task callbacks being skipped when `TriggerDagRunOperator` gets a 404 [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #70719:
URL: https://github.com/apache/airflow/pull/70719#issuecomment-5178145615
### Backport successfully created: v3-3-test
Note: As of [Merging PRs targeted for Airflow 3.X](https://github.com/apache/airflow/blob/main/dev/README_AIRFLOW3_DEV.md#merging-prs-targeted-for-airflow-3x)
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.
In matter of doubt please ask in [#release-management](https://apache-airflow.slack.com/archives/C03G9H97MM2) Slack channel.
<table>
<tr>
<th>Status</th>
<th>Branch</th>
<th>Result</th>
</tr>
<tr>
<td>✅</td>
<td>v3-3-test</td>
<td><a href="https://github.com/apache/airflow/pull/71083"><img src="https://img.shields.io/badge/PR-71083-blue" alt="PR Link"></a></td>
</tr>
</table>
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix task callbacks being skipped when `TriggerDagRunOperator` gets a 404 [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #70719:
URL: https://github.com/apache/airflow/pull/70719#discussion_r3685637322
##########
task-sdk/tests/task_sdk/execution_time/test_task_runner.py:
##########
@@ -980,6 +988,214 @@ def test_defer_task_queue_assignment(
)
+@pytest.mark.parametrize(
+ ("should_retry", "expected_state"),
+ [
+ (True, TaskInstanceState.UP_FOR_RETRY),
+ (False, TaskInstanceState.FAILED),
+ ],
+)
+def test_defer_with_unserializable_kwargs_honours_retries_and_callbacks(
+ should_retry, expected_state, create_runtime_ti, mock_supervisor_comms
+):
+ """
+ A task that defers with a non-serializable ``next_kwargs`` value must fail like any
+ other task, rather than taking the whole run down.
+
+ ``_defer_task`` runs ``serde_serialize`` on the deferral kwargs, which raises
+ ``TypeError`` for anything it has no serializer for (a file handle, a client object,
+ a lambda). That raise happens inside ``run()``'s ``except TaskDeferred`` handler, so
+ before the fix it escaped ``run()`` without evaluating retries or running callbacks --
+ the same defect as https://github.com/apache/airflow/issues/70683, reached without
+ involving the API server at all.
Review Comment:
Yup, I'll remove it
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org