You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "hkc-8010 (via GitHub)" <gi...@apache.org> on 2026/04/04 06:25:27 UTC
[I] Deadline breach callback fires multiple times with multiple scheduler replicas (race condition in handle_miss) [airflow]
hkc-8010 opened a new issue, #64710:
URL: https://github.com/apache/airflow/issues/64710
## Description
When Airflow is running with multiple scheduler replicas (the default on Astronomer Hosted, where 2 scheduler replicas always run), the deadline check loop in `scheduler_job_runner.py` has a **race condition** that causes duplicate `CallbackTrigger` entries to be created for the same missed deadline. This results in the deadline breach callback firing **multiple times** — producing duplicate alerts/notifications.
With a single triggerer replica the duplicates may go unnoticed (they fire in rapid succession from the same pod). With **two or more triggerer replicas**, each triggerer pod claims one of the duplicate triggers and fires its callback independently, making the problem clearly visible to users.
---
## Steps to Reproduce
1. Use Airflow 3 with a DAG that defines a `deadline` using `DeadlineAlert` and a callback (e.g. `AsyncCallback`):
```python
from datetime import timedelta
from airflow.sdk.definitions.deadline import DeadlineAlert, DeadlineReference
from airflow.sdk import AsyncCallback
with DAG(
dag_id="crm-braze-cdi-daily",
schedule="@daily",
deadline=DeadlineAlert(
reference=DeadlineReference.DAGRUN_QUEUED_AT,
interval=timedelta(hours=1),
callback=AsyncCallback(path="my_package.callbacks.on_deadline_breach"),
),
) as dag:
...
```
2. Run Airflow with **2 scheduler replicas** (the default on Astro Hosted).
3. Run Airflow with **2 triggerer replicas** (e.g. `ASTRO_TRIGGERER_REPLICAS=2`).
4. Allow a DAG run to exceed its deadline.
**Expected:** The deadline breach callback fires **once**.
**Actual:** The deadline breach callback fires **twice** — once per triggerer replica.
---
## Root Cause Analysis
### 1. The scheduler deadline check loop lacks row-level locking
In `scheduler_job_runner.py`, the scheduling loop checks for missed deadlines on every iteration:
```python
# airflow/jobs/scheduler_job_runner.py, lines 1626-1635
with create_session() as session:
# Only retrieve expired deadlines that haven't been processed yet.
# `missed` is False by default until the handler sets it.
for deadline in session.scalars(
select(Deadline)
.where(Deadline.deadline_time < datetime.now(timezone.utc))
.where(~Deadline.missed) # ← NO SELECT FOR UPDATE
.options(selectinload(Deadline.callback), selectinload(Deadline.dagrun))
):
deadline.handle_miss(session)
```
There is **no `with_for_update()`** (pessimistic locking) on this query. Under PostgreSQL's default `READ COMMITTED` isolation, two scheduler replicas executing this query concurrently can **both read the same `Deadline` row** with `missed=False` before either has committed its transaction.
### 2. `handle_miss()` creates a new `Trigger` record each time it is called
```python
# airflow/models/deadline.py, lines 216-264
def handle_miss(self, session: Session):
...
if isinstance(self.callback, TriggererCallback):
self.callback.queue() # ← creates a NEW Trigger row in the DB
session.add(self.callback)
session.flush()
...
self.missed = True
session.add(self)
```
`TriggererCallback.queue()` calls `Trigger.from_object(CallbackTrigger(...))`, inserting a **new row** into the `trigger` table:
```python
# airflow/models/callback.py, lines 214-224
def queue(self):
from airflow.models.trigger import Trigger
from airflow.triggers.callback import CallbackTrigger
self.trigger = Trigger.from_object(
CallbackTrigger(
callback_path=self.data["path"],
callback_kwargs=self.data["kwargs"],
)
)
super().queue()
```
### 3. Race condition sequence (2 scheduler replicas)
```
Time → Scheduler A Scheduler B
T1 reads Deadline X (missed=False)
T1 reads Deadline X (missed=False)
T2 calls handle_miss() → Trigger 1 created, missed=True, COMMIT
T3 calls handle_miss() → Trigger 2 created, missed=True, COMMIT
```
At T3, Scheduler B's session still holds the stale `missed=False` value from T1 (READ COMMITTED doesn't re-read within the same transaction). B's update of `missed=True` succeeds silently (idempotent), but **Trigger 2 already exists in the database**.
### 4. Two triggerer replicas pick up one trigger each
`Trigger.assign_unassigned()` distributes unassigned triggers across available triggerer instances. With 2 triggerer replicas, each pod claims one of the two `CallbackTrigger` rows and executes the deadline breach callback independently → **2 notifications sent**.
With only **1 triggerer replica**, both triggers are picked up by the same pod and fire back-to-back. This is still a bug (fires twice), but it is much harder to notice because the notifications arrive nearly simultaneously.
---
## Proposed Fix
Add `.with_for_update(skip_locked=True)` to the deadline query in the scheduler loop. This ensures that when Scheduler A is processing a deadline, Scheduler B will skip it (rather than both processing it simultaneously):
```python
# scheduler_job_runner.py
with create_session() as session:
for deadline in session.scalars(
select(Deadline)
.where(Deadline.deadline_time < datetime.now(timezone.utc))
.where(~Deadline.missed)
.with_for_update(skip_locked=True) # ← ADD THIS
.options(selectinload(Deadline.callback), selectinload(Deadline.dagrun))
):
deadline.handle_miss(session)
```
`skip_locked=True` is preferred over `nowait=True` because it prevents the second scheduler from erroring out — it simply skips the row that is already locked by the first scheduler, which is the correct behaviour.
Alternatively, a **unique constraint** on `(dagrun_id, deadline_alert_id)` in the `deadline` table combined with an upsert pattern would also prevent the duplicate trigger creation, but the `SELECT FOR UPDATE / SKIP LOCKED` approach is lower risk and more consistent with how similar patterns are handled elsewhere in Airflow (e.g., trigger assignment).
---
## Impact
| Scenario | Behaviour |
|---|---|
| 1 scheduler, 1 triggerer | No visible duplicate (correct) |
| 2 schedulers, 1 triggerer | 2 triggers created; same pod fires callback twice (may be invisible) |
| 2 schedulers, 2 triggerers | 2 triggers created; each pod fires callback once → **user sees 2 notifications** |
The bug is latent whenever multiple scheduler replicas run (the default on Astronomer Hosted and most HA Airflow deployments). Adding a second triggerer makes it clearly visible because the two duplicate triggers are now distributed across two independent pods.
---
## Environment
- **Airflow version:** 3.0.x / 3.1.x (confirmed in `apache/airflow` `main` branch as of 2026-04-02)
- **Database:** PostgreSQL (READ COMMITTED isolation level)
- **Scheduler replicas:** 2 (standard Astro Hosted configuration)
- **Triggerer replicas:** 2 (set via `ASTRO_TRIGGERER_REPLICAS=2`)
- **Executor:** AstroExecutor
---
## Related Issues / PRs
- [apache/airflow#64620](https://github.com/apache/airflow/issues/64620) — Triggerer `TriggerCommsDecoder` `asyncio.Lock` race condition (separate but related triggerer stability issue on the same deployment)
--
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] Deadline breach callback fires multiple times with multiple scheduler replicas (race condition in handle_miss) [airflow]
Posted by "hkc-8010 (via GitHub)" <gi...@apache.org>.
hkc-8010 commented on issue #64710:
URL: https://github.com/apache/airflow/issues/64710#issuecomment-4188978903
Tested this PR on my Airflow local environment with Celery executor running multiple replicas of scheduler and triggerer, and it's working as expected. Below is the DAG that I used for testing. Thanks @shivaam.
```
from __future__ import annotations
from datetime import datetime, timedelta
from airflow.providers.slack.notifications.slack_webhook import SlackWebhookNotifier
from airflow.sdk import dag, task
from airflow.sdk.definitions.deadline import AsyncCallback, DeadlineAlert, DeadlineReference
SLACK_WEBHOOK_CONN_ID = "cre_testing_channel"
# Sleep longer than the deadline so the DagRun is still running when the alert fires.
SLEEP_SECONDS_PAST_DEADLINE = 180
@dag(
dag_id="example_deadline_callback_slack",
description="Deadline alert to Slack if the DagRun is still running 2 minutes after queue",
schedule=None,
start_date=datetime(2024, 1, 1),
catchup=False,
tags=["example", "deadline", "slack"],
deadline=DeadlineAlert(
reference=DeadlineReference.DAGRUN_QUEUED_AT,
interval=timedelta(minutes=1),
callback=AsyncCallback(
SlackWebhookNotifier,
kwargs={
"slack_webhook_conn_id": SLACK_WEBHOOK_CONN_ID,
"text": (
":alarm_clock: *Deadline missed* — DAG `{{ dag_run.dag_id }}` had not finished "
"within 1 minutes of being queued.\n"
"*Deadline time:* {{ deadline.deadline_time }}\n"
"*Dag run:* {{ dag_run }}"
),
},
),
),
)
def example_deadline_callback_slack():
@task
def run_longer_than_deadline() -> None:
import logging
import time
log = logging.getLogger(__name__)
log.info("Sleeping %s s so the deadline can fire while this run is still active.", SLEEP_SECONDS_PAST_DEADLINE)
time.sleep(SLEEP_SECONDS_PAST_DEADLINE)
run_longer_than_deadline()
dag_instance = example_deadline_callback_slack()
if __name__ == "__main__":
dag_instance.test()
```
--
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] Deadline breach callback fires multiple times with multiple scheduler replicas (race condition in handle_miss) [airflow]
Posted by "laurenhall1526-netizen (via GitHub)" <gi...@apache.org>.
laurenhall1526-netizen commented on issue #64710:
URL: https://github.com/apache/airflow/issues/64710#issuecomment-4187249414
Any updates on this
On Sat, Apr 4, 2026, 3:20 AM Ash Berlin-Taylor ***@***.***>
wrote:
> *ashb* left a comment (apache/airflow#64710)
> <https://github.com/apache/airflow/issues/64710#issuecomment-4186750964>
>
> The with locked also needs the of=Deadline (i think that's the kwarg) to
> only lock that table, otherwise it will lock the ones were join to which we
> don't want
>
> —
> Reply to this email directly, view it on GitHub
> <https://github.com/apache/airflow/issues/64710#issuecomment-4186750964>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/B7IHA45HLNIM66ADZF3HRYT4UDAUZAVCNFSM6AAAAACXMNH5SCVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHM2DCOBWG42TAOJWGQ>
> .
> You are receiving this because you are subscribed to this thread.Message
> ID: ***@***.***>
>
--
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] Deadline breach callback fires multiple times with multiple scheduler replicas (race condition in handle_miss) [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on issue #64710:
URL: https://github.com/apache/airflow/issues/64710#issuecomment-4186750964
The with locked also needs the `of=Deadline` (i think that's the kwarg) to only lock that table, otherwise it will lock the ones were join to which we don't want
--
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] Deadline breach callback fires multiple times with multiple scheduler replicas (race condition in handle_miss) [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on issue #64710:
URL: https://github.com/apache/airflow/issues/64710#issuecomment-4188179685
Opened #64737 with a fix using FOR UPDATE SKIP LOCKED per @ashb suggestion.
--
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] Deadline breach callback fires multiple times with multiple scheduler replicas (race condition in handle_miss) [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk closed issue #64710: Deadline breach callback fires multiple times with multiple scheduler replicas (race condition in handle_miss)
URL: https://github.com/apache/airflow/issues/64710
--
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