This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 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 78eae1f337f [v3-4-test] Guard scheduler against Dag deserialization
failures (#75263) (#75745)
78eae1f337f is described below
commit 78eae1f337f8af2856ce9f1818944c58add8be0d
Author: Rahul Vats <[email protected]>
AuthorDate: Tue Oct 27 16:15:40 2026 +0530
[v3-4-test] Guard scheduler against Dag deserialization failures (#75263)
(#75745)
---
.../src/airflow/jobs/scheduler_job_runner.py | 17 ++++--
airflow-core/tests/unit/jobs/test_scheduler_job.py | 71 ++++++++++++++++++++++
2 files changed, 84 insertions(+), 4 deletions(-)
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index 3a58e9cf4a2..91304448aee 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -591,10 +591,19 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
task_id = task_instance.task_id
run_id = task_instance.run_id
- serialized_dag = self.scheduler_dag_bag.get_dag_for_run(
- dag_run=task_instance.dag_run,
- session=session,
- )
+ try:
+ serialized_dag = self.scheduler_dag_bag.get_dag_for_run(
+ dag_run=task_instance.dag_run,
+ session=session,
+ )
+ except DBAPIError:
+ # DB errors must propagate so the surrounding transaction-retry
handling runs.
+ raise
+ except Exception:
+ # A single Dag that fails to deserialize must not stop the
scheduler from
+ # scheduling other Dags; skip this task instance instead of
crashing the loop.
+ self.log.exception("Failed to load Dag %s for run %s, skipping
it", dag_id, run_id)
+ return False
# If the DAG is missing, fail all scheduled TIs for this DAG.
if not serialized_dag:
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index 16b02cbc072..37b440655ee 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -3631,6 +3631,77 @@ class TestSchedulerJob:
session.rollback()
+ def
test_find_executable_task_instances_skips_dag_that_cannot_be_deserialized(
+ self, dag_maker, caplog, session
+ ):
+ bad_dag_id = (
+
"SchedulerJobTest.test_find_executable_task_instances_skips_dag_that_cannot_be_deserialized-bad"
+ )
+ good_dag_id = (
+
"SchedulerJobTest.test_find_executable_task_instances_skips_dag_that_cannot_be_deserialized-good"
+ )
+
+ with dag_maker(dag_id=bad_dag_id):
+ bad_task = EmptyOperator(task_id="bad_task", priority_weight=2,
max_active_tis_per_dag=1)
+ bad_run_1 = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED)
+ bad_run_2 = dag_maker.create_dagrun_after(bad_run_1,
run_type=DagRunType.SCHEDULED)
+
+ with dag_maker(dag_id=good_dag_id):
+ good_task = EmptyOperator(task_id="good_task", priority_weight=1)
+ good_run = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED)
+
+ bad_running_ti = bad_run_1.get_task_instance(bad_task.task_id,
session=session)
+ bad_scheduled_ti = bad_run_2.get_task_instance(bad_task.task_id,
session=session)
+ good_scheduled_ti = good_run.get_task_instance(good_task.task_id,
session=session)
+ bad_running_ti.state = TaskInstanceState.RUNNING
+ bad_scheduled_ti.state = TaskInstanceState.SCHEDULED
+ good_scheduled_ti.state = TaskInstanceState.SCHEDULED
+ 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_id:
+ raise DeserializationError(bad_dag_id) 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),
+ ):
+ result = self.job_runner._select_task_instances_to_queue(
+ 1, make_pool_stats(), set(), session=session
+ )
+
+ assert [ti.key for ti in result] == [good_scheduled_ti.key]
+ assert bad_scheduled_ti.state == TaskInstanceState.SCHEDULED
+ assert any(bad_dag_id in record.getMessage() for record in
caplog.records)
+
+ session.rollback()
+
+ def
test_find_executable_task_instances_reraises_db_errors_from_dag_load(self,
dag_maker, session):
+ """DBAPIError while loading a Dag must propagate for transaction retry
handling."""
+ from sqlalchemy.exc import DBAPIError
+
+ with dag_maker(dag_id="db_error_dag_load"):
+ EmptyOperator(task_id="task", max_active_tis_per_dag=1)
+ run = dag_maker.create_dagrun(state=DagRunState.RUNNING)
+ run.get_task_instance("task", session=session).state =
TaskInstanceState.SCHEDULED
+ session.flush()
+
+ self.job_runner = SchedulerJobRunner(job=Job(),
executors=[self.null_exec])
+ with patch.object(
+ self.job_runner.scheduler_dag_bag,
+ "get_dag_for_run",
+ side_effect=DBAPIError("select 1", None, Exception("connection
lost")),
+ ):
+ with pytest.raises(DBAPIError):
+ self.job_runner._select_task_instances_to_queue(1,
make_pool_stats(), set(), session=session)
+
+ session.rollback()
+
def
test_find_executable_task_instances_task_concurrency_per_dagrun_for_first(self,
dag_maker):
scheduler_job = Job()
self.job_runner = SchedulerJobRunner(job=scheduler_job)
