TriggerDagRunOperator can fail with DagRunAlreadyExists after ambiguous execution API trigger retry
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 9h
- Merged PRs (30d)
- 472
Description
### Apache Airflow version
main branch
### What happened?
`TriggerDagRunOperator` can fail with `DagRunAlreadyExists` even though the child Dag run was created successfully.
In the Airflow 3 task-sdk path, `DagRunOperations.trigger()` sends `POST /execution/dag-runs/{dag_id}/{run_id}` through the generic execution API retry layer. If the server creates the Dag run but the client sees an ambiguous transport or request error, the retry can POST the same run ID again and receive `409 Conflict`.
The task runner then treats that as a real pre-existing run, marks the parent task failed, and does not write the `trigger_run_id` XCom.
### What you think should happen instead?
A transport-level ambiguity after a trigger POST should not be converted into a duplicate-run failure when the requested Dag run now exists.
### How to reproduce
1. Mock `POST /dag-runs/{dag_id}/{run_id}` so the server-side run is created but the client sees `httpx.RequestError`.
2. Return an existing Dag run from `GET /dag-runs/{dag_id}/{run_id}`.
3. The trigger operation should treat this as success for that run ID instead of surfacing `DAGRUN_ALREADY_EXISTS`.
### Code pointers
`task-sdk/src/airflow/sdk/api/client.py`
`task-sdk/src/airflow/sdk/execution_time/task_runner.py`
### Are you willing to submit PR?
Yes.
Contributor guide
Assessment
This issue has not been assessed yet.