You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Mercury2699 (via GitHub)" <gi...@apache.org> on 2026/06/04 23:54:29 UTC
[I] Duplicate DAG runs for PartitionedAssetTimetable with multiple schedulers (HA) — race not covered by #60773 [airflow]
Mercury2699 opened a new issue, #68045:
URL: https://github.com/apache/airflow/issues/68045
# Duplicate DAG runs for PartitionedAssetTimetable with multiple schedulers (HA) — race not covered by #60773
## Apache Airflow version
3.2.1
## What happened?
Asset-triggered DAGs using `PartitionedAssetTimetable` or `CronPartitionTimetable` are intermittently triggered twice for the same asset event when running with 2 schedulers (HA mode).
PR #60773 ("Add row lock to ADRQ before Dag run creation") fixed this race for **non-partitioned** asset scheduling by adding `with_row_locks` on `AssetDagRunQueue` reads. However, the **partitioned asset** code path in `scheduler_job_runner.py` appears to have a separate or parallel DagRun-creation flow that is not protected by the same row lock. The result is that two schedulers can both read the same unprocessed partitioned asset event and each independently create a DagRun.
This is confirmed reproducible on MWAA 3.2.1 (image `2026_05_12_23_43_Airflow_Image_3_2_1`) which includes #60773 (shipped in 3.2.0).
## What you think should happen instead?
A single partitioned asset event should trigger exactly one DAG run, regardless of how many schedulers are running.
The partitioned-asset scheduling path should receive the same row-lock treatment that #60773 applied to the non-partitioned path.
## How to reproduce
### Prerequisites
- Airflow 3.2.1 with **2 schedulers** (HA mode)
- PostgreSQL metadata database
### DAGs
**Producer** — emits a partitioned asset event every 5 minutes via `CronPartitionTimetable`:
```python
import time
from datetime import datetime, timedelta
import pendulum
from airflow.sdk import DAG, Asset, CronPartitionTimetable
from airflow.providers.standard.operators.python import PythonOperator
ASSET_SHARED = Asset("test_v7b_shared_events")
def simulate_processing(**context):
dag_run = context["dag_run"]
partition_key = getattr(dag_run, "partition_key", None) or getattr(dag_run, "logical_date", None)
print(f"Processing partition: {partition_key}")
time.sleep(10)
with DAG(
dag_id="test_partitioned_asset_producer",
start_date=pendulum.now("UTC").subtract(hours=2),
schedule=CronPartitionTimetable("*/5 * * * *", timezone="UTC"),
catchup=False,
max_active_runs=1,
tags=["repro", "partitioned_asset_race"],
) as dag:
PythonOperator(
task_id="process_data",
python_callable=simulate_processing,
outlets=[ASSET_SHARED],
)
```
**Consumer Alpha** — uses `PartitionedAssetTimetable`:
```python
import time
from datetime import datetime
from airflow.sdk import DAG, Asset, PartitionedAssetTimetable
from airflow.providers.standard.operators.python import PythonOperator
ASSET_SHARED = Asset("test_v7b_shared_events")
def process_event(**context):
print(f"Consumer ALPHA - Run ID: {context['dag_run'].run_id}")
time.sleep(5)
with DAG(
dag_id="test_partitioned_asset_consumer_alpha",
start_date=datetime(2024, 1, 1),
schedule=PartitionedAssetTimetable(assets=ASSET_SHARED),
catchup=False,
tags=["repro", "partitioned_asset_race", "consumer"],
) as dag:
PythonOperator(task_id="process_event", python_callable=process_event)
```
**Consumer Beta** — identical schedule, different DAG ID:
```python
import time
from datetime import datetime
from airflow.sdk import DAG, Asset, PartitionedAssetTimetable
from airflow.providers.standard.operators.python import PythonOperator
ASSET_SHARED = Asset("test_v7b_shared_events")
def process_event(**context):
print(f"Consumer BETA - Run ID: {context['dag_run'].run_id}")
time.sleep(5)
with DAG(
dag_id="test_partitioned_asset_consumer_beta",
start_date=datetime(2024, 1, 1),
schedule=PartitionedAssetTimetable(assets=ASSET_SHARED),
catchup=False,
tags=["repro", "partitioned_asset_race", "consumer"],
) as dag:
PythonOperator(task_id="process_event", python_callable=process_event)
```
### Steps
1. Deploy all 3 DAGs with **2 schedulers** (HA).
2. Unpause all DAGs.
3. Let run for 1+ hours.
4. Check consumer DAG run history for duplicates.
### Key conditions
- Must have **2+ schedulers** (HA mode)
- Must have **2+ consumer DAGs** on the same asset (single consumer does NOT trigger the bug)
- Must use **`PartitionedAssetTimetable`** — plain `Asset` schedule (non-partitioned) does NOT reproduce (covered by #60773)
- The issue is intermittent (~40ms race window between schedulers)
### Evidence
Duplicate run IDs from production reproduction (timestamps identical, hashes differ):
```
asset_triggered__2026-06-02T10:30:14.149756+00:00_EMV7bEJl
asset_triggered__2026-06-02T10:30:14.265600+00:00_ePLMw1Xd
```
Scheduler logs show both `job_id=3` and `job_id=6` independently creating consumer runs within 40ms of each other from the same producer completion event.
## Root cause analysis
PR #60773 added `with_row_locks` to the ADRQ fetch in `_create_dag_runs_asset_triggered`. However, when `PartitionedAssetTimetable` is used, the DagRun creation appears to flow through a different code path (likely involving `AssetPartitionDagRun` / partition-specific logic) that does not acquire the same exclusive lock before creating the run.
The fix should extend the row-lock strategy from #60773 to cover the partitioned-asset scheduling path in `scheduler_job_runner.py`.
## Related issues
- #63507 — Duplicate DAG runs for asset-triggered scheduling (closed by #60773, non-partitioned)
- #54491 — Original report (Airflow 3.0.4, non-partitioned)
- #60773 — The fix PR (row lock on ADRQ, merged Mar 17 2026)
- #59183, #61831, #61433, #62441 — Related partition concurrency fixes in 3.2.0/3.2.1
## Are you willing to submit PR?
- Yes I am willing to submit a PR!
--
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: commits-unsubscribe@airflow.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Duplicate DAG runs for PartitionedAssetTimetable with multiple schedulers (HA) — race not covered by #60773 [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on issue #68045:
URL: https://github.com/apache/airflow/issues/68045#issuecomment-4626955502
Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Duplicate DAG runs for PartitionedAssetTimetable with multiple schedulers (HA) — race not covered by #60773 [airflow]
Posted by "Mercury2699 (via GitHub)" <gi...@apache.org>.
Mercury2699 closed issue #68045: Duplicate DAG runs for PartitionedAssetTimetable with multiple schedulers (HA) — race not covered by #60773
URL: https://github.com/apache/airflow/issues/68045
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Duplicate DAG runs for PartitionedAssetTimetable with multiple schedulers (HA) — race not covered by #60773 [airflow]
Posted by "Mercury2699 (via GitHub)" <gi...@apache.org>.
Mercury2699 commented on issue #68045:
URL: https://github.com/apache/airflow/issues/68045#issuecomment-4674736752
Resolved by PR #64571 ("AIP-76: Hold Dag run until all upstream partitions arrive", merged Jun 5 2026). That PR rewrote `_create_dagruns_for_partitioned_asset_dags` with `with_row_locks(of=AssetPartitionDagRun, skip_locked=True, key_share=False)` built in, which prevents the HA race condition described here.
The fix is on `main` and will ship in the next Airflow release (3.3.0 or a 3.2.x backport). PR #68061 (which targeted this issue) is now redundant.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org