Use dynamic task mapping in TriggerDagRunOperator may generate the same run_id
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 10h
- Merged PRs (30d)
- 483
Description
### Apache Airflow version
2.5.0
### What happened
I use TriggerDagRunOperator to generate a large number of dag instances(>= 100), and it sometimes causes a bug that is
``` log
Failed to execute job 7159001 for task dynamic_dags ((psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "dag_run_dag_id_run_id_key"
DETAIL: Key (dag_id, run_id)=(dynamic_dags, manual__2023-01-11T22:28:03.286419+00:00) already exists.
```
``` python
TriggerDagRunOperator.partial(task_id='dynamic_dags', trigger_dag_id='trigger_dag', wait_for_completion=True).expand(conf=[{"id": 1}, {"id": 2}, {"id": 3}, ......])
```
### What you think should happen instead
I expect the `run_id` to be always unique when I use dynamic task mapping to generate instances.
### How to reproduce
1. Create a dag called `trigger_dag`.
2. Create a dag called `test_dag`, which runs TriggerDagRunOperator with the dynamic task to generate a large number of `trigger_dag`.
3. Sometimes, you will see the task failed due to the same `run_id`.
* I use the local executor and parallelism = 16, max_active_tasks_per_dag = 12, and max_active_runs_per_dag = 12
### Operating System
Ubuntu 20.04.4 LTS
### Versions of Apache Airflow Providers
_No response_
### Deployment
Virtualenv installation
### Deployment details
_No response_
### Anything else
_No response_
### Are you willing to submit PR?
- [X] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Research direction
Start with TriggerDagRunOperator and the dynamic task-mapping execution path, then reproduce the failure using many mapped triggers and the reported concurrency settings. Trace how run_id values are generated and persisted, and add coverage showing that concurrently mapped triggers do not collide; done means the duplicate-key failure is prevented.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- backend, data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100