humit0 opened a new issue, #54491: URL: https://github.com/apache/airflow/issues/54491
### Apache Airflow version 3.0.4 ### If "Other Airflow 2 version" selected, which one? _No response_ ### What happened? Asset-Aware scheduling DAG is intermittently triggered twice for the same asset event when running two schedulers. ### What you think should happen instead? DAG should only trigger once for the same asset event when running multiple schedulers. ### How to reproduce Since the problem appears to be caused by a race condition, it is difficult to reproduce. However, you can try by having one DAG generate an asset event via its `outlets` using a cron expression, and scheduling another DAG to receive that asset event. Then, run two schedulers. ### Operating System Debian GNU/Linux 12 (bookworm) ### Versions of Apache Airflow Providers apache-airflow-providers-amazon==9.11.0 apache-airflow-providers-apache-hive==9.1.2 apache-airflow-providers-apache-spark==5.3.2 apache-airflow-providers-celery==3.12.2 apache-airflow-providers-common-compat==1.7.3 apache-airflow-providers-common-io==1.6.2 apache-airflow-providers-common-sql==1.27.4 apache-airflow-providers-fab==2.3.1 apache-airflow-providers-http==5.3.3 apache-airflow-providers-jdbc==5.2.2 apache-airflow-providers-mysql==6.3.3 apache-airflow-providers-postgres==6.2.2 apache-airflow-providers-redis==4.2.0 apache-airflow-providers-smtp==2.1.2 apache-airflow-providers-standard==1.5.0 ### Deployment Official Apache Airflow Helm Chart ### Deployment details _No response_ ### Anything else? Examination of the asset-based scheduling logic indicates that the process operates as follows: 1. It fetches all records from the `asset_dag_run_queue`. https://github.com/apache/airflow/blob/367d8680af355b492f256ab86aa738f9ee292f2f/airflow-core/src/airflow/models/dag.py#L2146-L2148 2. It calculates the list of DAGs that need to be run based on these records. (with row level lock) https://github.com/apache/airflow/blob/367d8680af355b492f256ab86aa738f9ee292f2f/airflow-core/src/airflow/models/dag.py#L2204-L2207 3. It creates DagRuns for the determined DAGs. https://github.com/apache/airflow/blob/367d8680af355b492f256ab86aa738f9ee292f2f/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1626-L1638 4. Finally, it deletes the corresponding records from the `asset_dag_run_queue` matching the dag_id. https://github.com/apache/airflow/blob/367d8680af355b492f256ab86aa738f9ee292f2f/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1641 The problem appears to stem from the step where records are fetched from the `asset_dag_run_queue`. It seems that no row-level lock is applied during this read, which can lead to a race condition. When two schedulers are active, it's possible for one scheduler to read the `asset_dag_run_queue` records before the other scheduler has deleted the records it is processing for the same dag_id. This scenario appears to be the cause of the duplicate DAG triggers. It seems that the issue could be resolved by applying a row-level lock when fetching all records from the `asset_dag_run_queue`, but I'm not sure if this is the best approach. ### Are you willing to submit PR? - [ ] 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) -- 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]
