o-nikolas commented on code in PR #69989:
URL: https://github.com/apache/airflow/pull/69989#discussion_r3616746261
##########
airflow-core/src/airflow/listeners/listener.py:
##########
@@ -36,17 +54,37 @@ def get_listener_manager() -> ListenerManager:
- taskinstance: on_task_instance_running, on_task_instance_success, etc.
- asset: on_asset_created, on_asset_changed, etc.
- importerrors: on_new_dag_import_error, on_existing_dag_import_error
+
+ :param team_name: In multi-team mode, the team whose (and the global)
plugin
+ listeners this manager should contain. ``None`` yields a manager with
only
+ global-plugin listeners (used by team-agnostic events such as
lifecycle,
+ asset, and import-error hooks). When multi-team mode is disabled this
+ argument is ignored and the manager holds every plugin's listeners.
+
+ Calls are cached per ``team_name``. The single positional delegation to
+ ``_build_listener_manager`` guarantees that ``get_listener_manager()``,
+ ``get_listener_manager(None)`` and
``get_listener_manager(team_name=None)`` all
+ return the identical (global) manager instance rather than distinct cache
+ entries.
"""
- _listener_manager = ListenerManager()
+ return _build_listener_manager(team_name)
- _listener_manager.add_hookspecs(lifecycle)
- _listener_manager.add_hookspecs(dagrun)
- _listener_manager.add_hookspecs(taskinstance)
- _listener_manager.add_hookspecs(asset)
- _listener_manager.add_hookspecs(importerrors)
- integrate_listener_plugins(_listener_manager)
- return _listener_manager
+def get_listener_manager_for_dag(dag_id: str, session: Session | None = None)
-> ListenerManager:
Review Comment:
Hey Ash, thanks for the review.
None of these `get_listener_manager_for_dag` are happening in the dag
processor or the sdk side at all, the scheduler and API server are calling this
method. Those are shared team spaces.
The scheduler does have a cached mapping of dag id to teams, so it is
possible for a few of the call sites to plumb that through (though you pay the
cost of more plumbing, so pick your poison). The API server does not have a
centralized dag id to team name cache, but the call volume of those APIs (and
thus their listeners) should be very small, I don't think we need to worry
about that at the moment. So dag id can still be passed along there.
What are your thoughts @ashb?
--
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]
