This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new e4e96516c32 [v3-4-test] Catch lazy timetable errors when starting
queued runs (#76288) (#76292)
e4e96516c32 is described below
commit e4e96516c3236a44edd49e7e7a8850791cbfe5fe
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Nov 6 08:49:23 2026 +0000
[v3-4-test] Catch lazy timetable errors when starting queued runs (#76288)
(#76292)
(cherry picked from commit 1d3d8f533fb744d88bf7621b256c9a7b208266b0)
Signed-off-by: Adam <[email protected]>
Co-authored-by: Adam Postek <[email protected]>
---
.../src/airflow/jobs/scheduler_job_runner.py | 2 +
airflow-core/tests/unit/jobs/test_scheduler_job.py | 45 ++++++++++++++++++++++
2 files changed, 47 insertions(+)
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index bd9212973ed..49b16c7251d 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -3218,6 +3218,8 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
backfill_id = dag_run.backfill_id
try:
dag = dag_run.dag = cached_get_dag(dag_run)
+ if dag:
+ _ = dag.timetable
except DBAPIError:
raise
except Exception:
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index 6ed8be40735..3c9b6070a88 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -151,6 +151,7 @@ from airflow.sdk import (
from airflow.sdk.definitions._internal.loop import create_loop
from airflow.sdk.definitions.callback import AsyncCallback, SyncCallback
from airflow.sdk.definitions.timetables.assets import AssetTriggeredTimetable,
PartitionedAssetTimetable
+from airflow.serialization.definitions import dag as serialized_dag_definitions
from airflow.serialization.definitions.dag import SerializedDAG
from airflow.serialization.encoders import ensure_serialized_asset
from airflow.serialization.serialized_objects import LazyDeserializedDAG
@@ -7818,6 +7819,50 @@ class TestSchedulerJob:
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_start_queued_dagruns_does_not_crash_on_lazy_timetable_error(self,
dag_maker, caplog, session):
+ """The timetable is decoded lazily, so a failing one must be caught
when the Dag is loaded."""
+ 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
+ broken_timetable = {"__type": "broken"}
+ decode_timetable = serialized_dag_definitions.decode_timetable
+
+ def break_bad_dag_timetable(dag_run, session):
+ dag = get_dag_for_run(dag_run=dag_run, session=session)
+ if dag_run.dag_id == "bad_dag":
+ dag._encoded_timetable = broken_timetable
+ return dag
+
+ def fail_for_broken_timetable(encoded):
+ if encoded is broken_timetable:
+ raise RuntimeError("Variable not found")
+ return decode_timetable(encoded)
+
+ 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=break_bad_dag_timetable
+ ),
+ patch.object(
+ serialized_dag_definitions, "decode_timetable",
side_effect=fail_for_broken_timetable
+ ),
+ ):
+ 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.
