seanghaeli opened a new issue, #74677:
URL: https://github.com/apache/airflow/issues/74677
### What happened
When the scheduler creates a Dag run, it saves the Dag run and a deadline
for each of the Dag's deadline alerts to the database in one go. It handles the
alerts one at a time, in the order they're listed on the Dag.
If one alert's code undoes the scheduler's unsaved changes (for example, a
custom deadline reference that catches a database error and calls
`session.rollback()`), the new Dag run is undone along with it. Then:
- **If a later alert uses `DeadlineReference.FIXED_DATETIME`,** it still
adds a deadline for the Dag run that was just undone. Saving fails because that
Dag run no longer exists, and the scheduler exits. Restarting it crashes it
again within a second, and no Dag gets new runs in the meantime.
- **Otherwise,** the Dag run is quietly not created, but the scheduler moves
on as if it had been. It only tries again when the Dag file is re-read (every
30 seconds by default), and fails the same way each time. The Dag never runs.
#68917 made a failing alert safe for ordinary errors, like a missing
Variable. This is the one kind of failure it doesn't cover.
### What you think should happen instead
A failing alert should never crash the scheduler or stop the Dag run from
being created.
### How to reproduce
Tested on main (`417a33b5432`) in Breeze with Postgres, LocalExecutor, and
the real scheduler, dag-processor and api-server.
Plugin, `plugins/my_references.py`:
```python
from sqlalchemy import text
from airflow.plugins_manager import AirflowPlugin
from airflow.sdk import BaseDeadlineReference, deadline_reference
@deadline_reference
class RollbackReference(BaseDeadlineReference):
"""Reads a time from a table. On a database error it rolls back, then
re-raises."""
def _evaluate_with(self, *, session, **kwargs):
try:
return session.execute(text("SELECT due FROM
my_sla_table")).scalar_one()
except Exception:
session.rollback()
raise
async def notify(**kwargs):
print("deadline missed", kwargs)
class MyReferencesPlugin(AirflowPlugin):
name = "my_references"
deadline_references = [RollbackReference]
```
Dag (`my_sla_table` doesn't exist, so the first alert always fails):
```python
from datetime import datetime, timedelta
from my_references import notify
from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import DAG, AsyncCallback, DeadlineAlert, DeadlineReference
with DAG(
dag_id="rollback_fixed",
schedule="@daily",
start_date=datetime(2026, 1, 1),
catchup=False,
deadline=[
DeadlineAlert(
reference=DeadlineReference.RollbackReference,
interval=timedelta(hours=1),
callback=AsyncCallback(notify),
),
DeadlineAlert(
reference=DeadlineReference.FIXED_DATETIME(datetime(2030, 1, 1)),
interval=timedelta(0),
callback=AsyncCallback(notify),
),
],
):
BashOperator(task_id="wait", bash_command="sleep 600")
```
Unpause it and let the scheduler create the Dag run. Saving fails with an
`IntegrityError` on `deadline_dagrun_id_fkey` and the scheduler exits.
Change the second alert to `DeadlineReference.DAGRUN_LOGICAL_DATE` (or
remove it) and the scheduler stays up, but no Dag run is ever created.
### Anything else
- **Possible fix:** before handling each alert, check whether the Dag run
has already been undone, and if so skip the rest of the alerts (in code: skip
when `orm_dagrun not in session`).
- **Why it matters for #70714:** once #70714 merges, `DAGRUN_LOGICAL_DATE`
and `DAGRUN_QUEUED_AT` alerts behave like `FIXED_DATETIME` here, so the crash
covers the most common references too.
--
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]
