samraj2k commented on code in PR #74915:
URL: https://github.com/apache/airflow/pull/74915#discussion_r4370508417
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1003,15 +1036,65 @@ def _refresh_dag_bundles(self, known_files: dict[str,
set[DagFileInfo]]):
self._resort_file_queue()
self._add_new_files_to_queue(known_files=known_files)
+ def _reconcile_bundles(self, known_files: dict[str, set[DagFileInfo]]) ->
None:
+ if not is_bundle_config_reload_enabled():
+ return
+ try:
+ dag_bundle_manager = DagBundlesManager()
+ bundle_names = set(dag_bundle_manager.get_all_bundle_names())
+ except Exception:
+ self.log.exception("Could not load Dag bundles; keeping the
current bundles")
+ return
+
+ loaded_bundles = {bundle.name: bundle for bundle in self._dag_bundles}
+ if self.bundle_names_to_parse:
+ bundle_names.intersection_update(self.bundle_names_to_parse)
+ removed_names = loaded_bundles.keys() - bundle_names
+ bundle_instances: dict[str, BaseDagBundle] = {}
+ for name in sorted(bundle_names - loaded_bundles.keys()):
+ try:
+ bundle_instances[name] = dag_bundle_manager.get_bundle(name)
+ except Exception:
+ self.log.exception("Could not create Dag bundle %s; will
retry", name)
+
+ try:
+ # Other processors can overwrite active state even when our
configuration is unchanged.
+ dag_bundle_manager.sync_bundles_to_db(
+ deactivate_missing=not self.bundle_names_to_parse,
+ bundle_instances=bundle_instances,
+ )
+ except Exception:
+ self.log.exception("Could not reconcile Dag bundles; keeping the
current bundles")
Review Comment:
Thanks, have added check in sync_bundles_to_db() method that all teams
exist before updating bundle rows. If a team does not exist we keep the
previous configs and existing bundles active.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
