Duplicate DAG triggering for asset events with multiple schedulers

[humit0]

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.

   airflow/airflow-core/src/airflow/models/dag.py

   Lines 2146 to 2148 in 367d868

   	adrq_by_dag: dict[str, list[AssetDagRunQueue]] = defaultdict(list)
   	for r in session.scalars(select(AssetDagRunQueue)):
   	adrq_by_dag[r.target_dag_id].append(r)

2. It calculates the list of DAGs that need to be run based on these records. (with row level lock)

   airflow/airflow-core/src/airflow/models/dag.py

   Lines 2204 to 2207 in 367d868

   	return (
   	session.scalars(with_row_locks(query, of=cls, session=session, skip_locked=True)),
   	triggered_date_by_dag,
   	)

3. It creates DagRuns for the determined DAGs.

   airflow/airflow-core/src/airflow/jobs/scheduler_job_runner.py

   Lines 1626 to 1638 in 367d868

   	dag_run = dag.create_dagrun(
   	run_id=DagRun.generate_run_id(
   	run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=triggered_date
   	),
   	logical_date=None,
   	data_interval=None,
   	run_after=triggered_date,
   	run_type=DagRunType.ASSET_TRIGGERED,
   	triggered_by=DagRunTriggeredByType.ASSET,
   	state=DagRunState.QUEUED,
   	creating_job_id=self.job.id,
   	session=session,
   	)

4. Finally, it deletes the corresponding records from the asset_dag_run_queue matching the dag_id.

   airflow/airflow-core/src/airflow/jobs/scheduler_job_runner.py

   Line 1641 in 367d868

   session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_run.dag_id))

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

• I agree to follow this project's Code of Conduct

[ashb]

What database are you using please @humit0

[humit0]

@ashb I'm using postgresql!