diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index a4d4917958c11..e8a8a2c50366e 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.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 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]): 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 b136a4d9e8fe3..4599c8ea58342 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 @@ def test_refresh_dag_bundles_get_bundle_state_failure_skips_bundle(self): 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.