This is an automated email from the ASF dual-hosted git repository. vatsrahul1001 pushed a commit to branch fix-scheduler-deser-crash-v3-4-test in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 7eea26a7844a3d2acaabeea4ef0873b740903c51 Author: Rahul Vats <[email protected]> AuthorDate: Tue Oct 27 16:06:52 2026 +0530 [v3-4-test] Guard scheduler against Dag deserialization failures (#75263) --- .../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)
