This is an automated email from the ASF dual-hosted git repository.
ashb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 1e404fac4ac Stop the scheduler crashing when a queued run's Dag fails
to deserialize (#74499)
1e404fac4ac is described below
commit 1e404fac4ace22467a2e3680523c2fd0d60b3d19
Author: Adam Postek <[email protected]>
AuthorDate: Fri Oct 9 15:22:16 2026 +0000
Stop the scheduler crashing when a queued run's Dag fails to deserialize
(#74499)
A queued run whose Dag cannot be deserialized (for example a custom
timetable
calling Variable.get on a missing Variable) raised out of
_start_queued_dagruns
and crashed the scheduler, which then crashed again on every restart. Log
and
skip that run instead, as _schedule_all_dag_runs already does for running
runs.
---
airflow-core/newsfragments/74499.bugfix.rst | 1 +
.../src/airflow/jobs/scheduler_job_runner.py | 8 ++++-
airflow-core/tests/unit/jobs/test_scheduler_job.py | 40 +++++++++++++++++++++-
3 files changed, 47 insertions(+), 2 deletions(-)
diff --git a/airflow-core/newsfragments/74499.bugfix.rst
b/airflow-core/newsfragments/74499.bugfix.rst
new file mode 100644
index 00000000000..c32f1614959
--- /dev/null
+++ b/airflow-core/newsfragments/74499.bugfix.rst
@@ -0,0 +1 @@
+Fix the scheduler crashing, and crashing again on every restart, when a queued
Dag run's Dag cannot be deserialized, for example because a custom timetable
reads a missing Variable. The run is now logged and skipped until its Dag loads.
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index 8bd4085a7fe..b13ceefb9a1 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -3052,7 +3052,13 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
dag_id = dag_run.dag_id
run_id = dag_run.run_id
backfill_id = dag_run.backfill_id
- dag = dag_run.dag = cached_get_dag(dag_run)
+ try:
+ dag = dag_run.dag = cached_get_dag(dag_run)
+ except DBAPIError:
+ raise
+ except Exception:
+ self.log.exception("Failed to load Dag %s for queued run %s,
skipping it", dag_id, run_id)
+ continue
if not dag:
self.log.error("DAG '%s' not found in serialized_dag table",
dag_run.dag_id)
continue
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index 8a128fba211..0633e4d2d5f 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -58,7 +58,7 @@ from airflow.callbacks.callback_requests import (
from airflow.callbacks.database_callback_sink import DatabaseCallbackSink
from airflow.dag_processing.collection import AssetModelOperation,
DagModelOperation
from airflow.dag_processing.dagbag import DagBag, sync_bag_to_db
-from airflow.exceptions import AirflowException, TaskNotFound
+from airflow.exceptions import AirflowException, DeserializationError,
TaskNotFound
from airflow.executors.base_executor import BaseExecutor
from airflow.executors.executor_constants import MOCK_EXECUTOR
from airflow.executors.executor_loader import ExecutorLoader
@@ -7070,6 +7070,44 @@ class TestSchedulerJob:
for msg in error_messages
)
+ def test_start_queued_dagruns_does_not_crash_on_dag_deserialization_error(
+ self, dag_maker, caplog, session
+ ):
+ """A queued run whose Dag cannot be deserialized must not crash the
scheduler.
+
+ A custom timetable that raises in ``deserialize`` (for example
``Variable.get`` on a missing
+ Variable) used to escape ``_start_queued_dagruns`` into the scheduler
loop, so every restart hit
+ the same queued run again and no Dag was scheduled at all.
+ """
+ with dag_maker(dag_id="bad_dag", schedule="@once"):
+ EmptyOperator(task_id="bad_task")
+ dag_maker.create_dagrun(state=DagRunState.QUEUED)
+
+ with dag_maker(dag_id="good_dag", schedule="@once"):
+ EmptyOperator(task_id="good_task")
+ dag_maker.create_dagrun(state=DagRunState.QUEUED)
+ session.flush()
+
+ self.job_runner = SchedulerJobRunner(job=Job(),
executors=[self.null_exec])
+ get_dag_for_run = self.job_runner.scheduler_dag_bag.get_dag_for_run
+
+ def fail_for_bad_dag(dag_run, session):
+ if dag_run.dag_id == "bad_dag":
+ raise DeserializationError("bad_dag") from
RuntimeError("Variable not found")
+ return get_dag_for_run(dag_run=dag_run, session=session)
+
+ caplog.clear()
+ with (
+ caplog.at_level("ERROR",
logger="airflow.jobs.scheduler_job_runner"),
+ patch.object(self.job_runner.scheduler_dag_bag, "get_dag_for_run",
side_effect=fail_for_bad_dag),
+ ):
+ self.job_runner._start_queued_dagruns(session)
+ session.flush()
+
+ states = dict(session.execute(select(DagRun.dag_id,
DagRun.state)).all())
+ assert states == {"good_dag": DagRunState.RUNNING, "bad_dag":
DagRunState.QUEUED}
+ assert any(r.levelno >= logging.ERROR and "bad_dag" in r.getMessage()
for r in caplog.records)
+
def test_schedule_all_dag_runs_reraises_db_errors(self, dag_maker,
session):
"""Test that _schedule_all_dag_runs does not catch DBAPIError, allowing
it to propagate to @retry_db_transaction for proper retry handling.
