This is an automated email from the ASF dual-hosted git repository.
kaxil 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 ca31bc2e11f Catch general Exception when initializing a dag bundle
(#71355)
ca31bc2e11f is described below
commit ca31bc2e11f93970d672be87ee61f025b79380bb
Author: Amogh Desai <[email protected]>
AuthorDate: Mon Aug 10 14:06:22 2026 +0530
Catch general Exception when initializing a dag bundle (#71355)
---
airflow-core/src/airflow/dag_processing/manager.py | 3 +-
.../tests/unit/dag_processing/test_manager.py | 33 ++++++++++++++++++++++
2 files changed, 34 insertions(+), 2 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/manager.py
b/airflow-core/src/airflow/dag_processing/manager.py
index a4d4917958c..e8a8a2c5036 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -56,7 +56,6 @@ from airflow.dag_processing.bundles.base import (
from airflow.dag_processing.bundles.manager import DagBundlesManager
from airflow.dag_processing.collection import update_dag_parsing_results_in_db
from airflow.dag_processing.processor import DagFileParsingResult,
DagFileProcessorProcess
-from airflow.exceptions import AirflowException
from airflow.models.asset import remove_references_to_deleted_dags
from airflow.models.dag import DagModel
from airflow.models.dagbag import DagPriorityParsingRequest
@@ -853,7 +852,7 @@ class DagFileProcessorManager(LoggingMixin):
try:
bundle.initialize()
any_refreshed = True
- except AirflowException as e:
+ except Exception as e:
self.log.exception("Error initializing bundle %s: %s",
bundle.name, e)
continue
try:
diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py
b/airflow-core/tests/unit/dag_processing/test_manager.py
index b136a4d9e8f..4599c8ea583 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -3251,6 +3251,39 @@ class TestDagFileProcessorManager:
bundle.refresh.assert_not_called()
+ def
test_refresh_dag_bundles_initialize_non_airflow_exception_skips_bundle(self):
+ """
+ A bundle whose initialize() raises a non AirflowException must be
skipped, not
+ left to propagate and abort refresh for every other bundle.
+ """
+ manager = DagFileProcessorManager(max_runs=1)
+ failing_bundle = self._make_refresh_bundle()
+ failing_bundle.name = "failing_bundle"
+ failing_bundle.is_initialized = False
+ failing_bundle.initialize.side_effect = FileNotFoundError("Repository
path not found")
+
+ healthy_bundle = self._make_refresh_bundle()
+ healthy_bundle.name = "healthy_bundle"
+
+ manager._dag_bundles = [failing_bundle, healthy_bundle]
+ manager._force_refresh_bundles = set()
+
+ with (
+ mock.patch.object(
+ manager, "get_bundle_state",
return_value=BundleState(last_refreshed=None, version=None)
+ ),
+ mock.patch.object(manager, "update_bundle_state"),
+ mock.patch.object(manager, "_find_files_in_bundle",
return_value=[]),
+ mock.patch.object(manager, "deactivate_deleted_dags"),
+ mock.patch.object(manager, "clear_orphaned_import_errors"),
+ mock.patch.object(manager, "handle_removed_files"),
+ mock.patch.object(manager, "_resort_file_queue"),
+ mock.patch.object(manager, "_add_new_files_to_queue"),
+ ):
+ manager._refresh_dag_bundles({})
+
+ healthy_bundle.refresh.assert_called_once()
+
def
test_refresh_dag_bundles_update_bundle_state_failure_still_scans_files(self):
"""A failure in update_bundle_state() logs but does not skip file
scanning.
