This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 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 e3f5e6ff3af Skip scheduling tasks when their Dag cannot be
deserialized (#75263)
e3f5e6ff3af is described below
commit e3f5e6ff3affa813834fc3c5048eeff6e00688dd
Author: abhijeets25012-tech <[email protected]>
AuthorDate: Tue Oct 27 11:57:08 2026 +0530
Skip scheduling tasks when their Dag cannot be deserialized (#75263)
* Skip scheduling tasks when their Dag cannot be deserialized
Keep one Dag's deserialization failure from stopping scheduler progress for
unrelated Dags.
* Fix scheduler crash on DAG deserialization errors
Isolate deserialization failures to the affected DAG run instead of
starving the whole DAG, and let DB errors propagate so the scheduler
does not silently skip task instances.
---------
Co-authored-by: Abhijeet Raj Singh
<[email protected]>
---
.../src/airflow/jobs/scheduler_job_runner.py | 20 +++-
airflow-core/tests/unit/jobs/test_scheduler_job.py | 129 +++++++++++++++++++++
2 files changed, 145 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 01c80d53379..a8ccf1d8145 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -594,6 +594,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
task_instance: TI,
concurrency_map: ConcurrencyMap,
session: Session,
+ starved_dag_runs: set[tuple[str, str]],
starved_tasks: set[tuple[str, str]],
starved_tasks_task_dagrun_concurrency: set[tuple[str, str, str]],
) -> bool:
@@ -602,10 +603,20 @@ 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,
- )
+ if (dag_id, run_id) in starved_dag_runs:
+ return False
+
+ try:
+ serialized_dag = self.scheduler_dag_bag.get_dag_for_run(
+ dag_run=task_instance.dag_run,
+ session=session,
+ )
+ except DBAPIError:
+ raise
+ except Exception:
+ self.log.exception("Failed to load Dag %s for run %s, skipping
it", dag_id, run_id)
+ starved_dag_runs.add((dag_id, run_id))
+ return False
# If the DAG is missing, fail all scheduled TIs for this DAG.
if not serialized_dag:
@@ -924,6 +935,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
task_instance=task_instance,
concurrency_map=concurrency_map,
session=session,
+ starved_dag_runs=starved_dag_runs,
starved_tasks=starved_tasks,
starved_tasks_task_dagrun_concurrency=(starved_tasks_task_dagrun_concurrency),
)
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index aa2a9c7c6d3..5ebbb8be9dc 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -3632,6 +3632,135 @@ 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_starves_run_not_whole_dag_on_deserialization_error(
+ self, dag_maker, caplog, session
+ ):
+ """A deserialization failure must starve only the affected Dag run.
+
+ Starving ``(dag_id, task_id)`` would also block the same task in
healthy runs of the
+ same Dag, and every task of the broken run would try to load and log
the Dag again.
+ """
+ dag_id = (
+
"SchedulerJobTest.test_find_executable_task_instances_starves_run_not_whole_dag_"
+ "on_deserialization_error"
+ )
+ with dag_maker(dag_id=dag_id):
+ first_task = EmptyOperator(task_id="first_task",
priority_weight=2, max_active_tis_per_dag=1)
+ second_task = EmptyOperator(task_id="second_task",
priority_weight=1, max_active_tis_per_dag=1)
+
+ older_run = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED)
+ newer_run = dag_maker.create_dagrun_after(older_run,
run_type=DagRunType.SCHEDULED)
+
+ for operator in (first_task, second_task):
+ older_run.get_task_instance(operator.task_id,
session=session).state = TaskInstanceState.SCHEDULED
+ newer_run.get_task_instance(operator.task_id,
session=session).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_older_run(dag_run, session):
+ if dag_run.run_id == older_run.run_id:
+ raise DeserializationError(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_older_run
+ ),
+ ):
+ result = self.job_runner._select_task_instances_to_queue(
+ 1, make_pool_stats(), set(), session=session
+ )
+
+ assert [ti.run_id for ti in result] == [newer_run.run_id]
+ selected_task_id = result[0].task_id
+ session.expire_all()
+ assert (
+ newer_run.get_task_instance(selected_task_id,
session=session).state == TaskInstanceState.QUEUED
+ )
+ assert (
+ older_run.get_task_instance(first_task.task_id,
session=session).state
+ == TaskInstanceState.SCHEDULED
+ )
+ # The failure is logged once for the starved run, not once per task in
that run.
+ assert sum(dag_id in record.getMessage() for record in caplog.records)
== 1
+
+ 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()
+
@pytest.fixture
def interleaved_blocked_and_runnable_tis(self, dag_maker, session):
"""
