samraj2k commented on code in PR #74915:
URL: https://github.com/apache/airflow/pull/74915#discussion_r4370485877
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -691,6 +825,10 @@ def get_bundle(
:return: The DAG bundle.
"""
+ if name not in self._bundle_config:
+ if name not in _get_cached_bundle_config_snapshot().names:
+ _refresh_bundle_config_snapshot(force=True)
Review Comment:
Thanks for pointing this out.
The force refresh is to handle the case:
DagProcessor parses new supplied bundle -> Scheduler schedules new tasks
from DAG of this new bundle -> The workers still not have the new config (due
to refresh interval)
To resolve this comment, havradded _failed_bundle_lookups as suggested by
you, which remembers unsuccessful lookups per manager and bundle name until the
refresh interval expires. This way we make sure we dont fail task with error
like bundle not found due to delays
--
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]
