apache / apache/airflow

Use dynamic task mapping in TriggerDagRunOperator may generate the same run_id

Open
#28,868 5 comments 2 reactions 0 assignees View on GitHub
kind:feature
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.