Move listeners module to shared library for client server separation (#59883)

Extract the listeners infrastructure to `shared/listeners/` library to eliminate cross dependencies between airflow-core and task-sdk.

- ListenerManager and hookimpl marker now in shared library
- Hook specs split by callers:
  - shared: lifecycle, taskinstance (called from both sdk and core)
  - core: dagrun, asset, importerrors (called only from core)
- sdk registers only specs it actually uses (lifecycle, taskinstance)
- core registers all specs for full listener support

Author: amoghrajesh

airflow-core/pyproject.toml

@@ -161,6 +161,9 @@ dependencies = [
     # Start of shared configuration dependencies
     "pyyaml>=6.0.3",
     # End of shared configuration dependencies
+    # Start of shared listeners dependencies
+    "pluggy>=1.5.0",
+    # End of shared listeners dependencies
 ]
 
 
@@ -235,6 +238,7 @@ exclude = [
 "../shared/secrets_backend/src/airflow_shared/secrets_backend" = "src/airflow/_shared/secrets_backend"
 "../shared/secrets_masker/src/airflow_shared/secrets_masker" = "src/airflow/_shared/secrets_masker"
 "../shared/timezones/src/airflow_shared/timezones" = "src/airflow/_shared/timezones"
+"../shared/listeners/src/airflow_shared/listeners" = "src/airflow/_shared/listeners"
 "../shared/plugins_manager/src/airflow_shared/plugins_manager" = "src/airflow/_shared/plugins_manager"
 
 [tool.hatch.build.targets.custom]
@@ -305,6 +309,7 @@ apache-airflow-devel-common = { workspace = true }
 shared_distributions = [
     "apache-airflow-shared-configuration",
     "apache-airflow-shared-dagnode",
+    "apache-airflow-shared-listeners",
     "apache-airflow-shared-logging",
     "apache-airflow-shared-module-loading",
     "apache-airflow-shared-observability",

airflow-core/src/airflow/_shared/listeners

@@ -0,0 +1 @@
+../../../../shared/listeners/src/airflow_shared/listeners

airflow-core/src/airflow/listeners/__init__.py

@@ -17,6 +17,6 @@
 # under the License.
 from __future__ import annotations
 
-from pluggy import HookimplMarker
+from airflow._shared.listeners import hookimpl
 
-hookimpl = HookimplMarker("airflow")
+__all__ = ["hookimpl"]

airflow-core/src/airflow/listeners/listener.py

@@ -17,72 +17,36 @@
 # under the License.
 from __future__ import annotations
 
-import logging
 from functools import cache
-from typing import TYPE_CHECKING
-
-import pluggy
 
+from airflow._shared.listeners.listener import ListenerManager
+from airflow._shared.listeners.spec import lifecycle, taskinstance
+from airflow.listeners.spec import asset, dagrun, importerrors
 from airflow.plugins_manager import integrate_listener_plugins
 
-if TYPE_CHECKING:
-    from pluggy._hooks import _HookRelay
-
-log = logging.getLogger(__name__)
-
-
-def _before_hookcall(hook_name, hook_impls, kwargs):
-    log.debug("Calling %r with %r", hook_name, kwargs)
-    log.debug("Hook impls: %s", hook_impls)
-
-
-def _after_hookcall(outcome, hook_name, hook_impls, kwargs):
-    log.debug("Result from %r: %s", hook_name, outcome.get_result())
-
-
-class ListenerManager:
-    """Manage listener registration and provides hook property for calling them."""
-
-    def __init__(self):
-        from airflow.listeners.spec import (
-            asset,
-            dagrun,
-            importerrors,
-            lifecycle,
-            taskinstance,
-        )
-
-        self.pm = pluggy.PluginManager("airflow")
-        self.pm.add_hookcall_monitoring(_before_hookcall, _after_hookcall)
-        self.pm.add_hookspecs(lifecycle)
-        self.pm.add_hookspecs(dagrun)
-        self.pm.add_hookspecs(asset)
-        self.pm.add_hookspecs(taskinstance)
-        self.pm.add_hookspecs(importerrors)
-
-    @property
-    def has_listeners(self) -> bool:
-        return bool(self.pm.get_plugins())
-
-    @property
-    def hook(self) -> _HookRelay:
-        """Return hook, on which plugin methods specified in spec can be called."""
-        return self.pm.hook
-
-    def add_listener(self, listener):
-        if self.pm.is_registered(listener):
-            return
-        self.pm.register(listener)
-
-    def clear(self):
-        """Remove registered plugins."""
-        for plugin in self.pm.get_plugins():
-            self.pm.unregister(plugin)
-
 
 @cache
 def get_listener_manager() -> ListenerManager:
-    """Get singleton listener manager."""
+    """
+    Get a listener manager for Airflow core.
+
+    Registers the following listeners:
+    - lifecycle: on_starting, before_stopping
+    - dagrun: on_dag_run_running, on_dag_run_success, on_dag_run_failed
+    - 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
+    """
     _listener_manager = ListenerManager()
+
+    _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
+
+
+__all__ = ["get_listener_manager", "ListenerManager"]

airflow-core/src/airflow/listeners/spec/asset.py

@@ -15,6 +15,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+
 from __future__ import annotations
 
 from typing import TYPE_CHECKING

airflow-core/src/airflow/listeners/spec/importerrors.py

@@ -15,6 +15,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+
 from __future__ import annotations
 
 from pluggy import HookspecMarker

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py

@@ -27,7 +27,6 @@
 
 from airflow._shared.timezones import timezone
 from airflow.api_fastapi.core_api.datamodels.dag_versions import DagVersionResponse
-from airflow.listeners.listener import get_listener_manager
 from airflow.models import DagModel, DagRun, Log
 from airflow.models.asset import AssetEvent, AssetModel
 from airflow.providers.standard.operators.empty import EmptyOperator
@@ -1285,12 +1284,6 @@ def test_patch_dag_run_bad_request(self, test_client):
         body = response.json()
         assert body["detail"][0]["msg"] == "Input should be 'queued', 'success' or 'failed'"
 
-    @pytest.fixture(autouse=True)
-    def clean_listener_manager(self):
-        get_listener_manager().clear()
-        yield
-        get_listener_manager().clear()
-
     @pytest.mark.parametrize(
         ("state", "listener_state"),
         [
@@ -1300,11 +1293,11 @@ def clean_listener_manager(self):
         ],
     )
     @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
-    def test_patch_dag_run_notifies_listeners(self, test_client, state, listener_state):
+    def test_patch_dag_run_notifies_listeners(self, test_client, state, listener_state, listener_manager):
         from unit.listeners.class_listener import ClassBasedListener
 
         listener = ClassBasedListener()
-        get_listener_manager().add_listener(listener)
+        listener_manager(listener)
         response = test_client.patch(f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}", json={"state": state})
         assert response.status_code == 200
         assert listener.state == listener_state

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py

@@ -33,7 +33,6 @@
 from airflow.dag_processing.dagbag import DagBag, sync_bag_to_db
 from airflow.jobs.job import Job
 from airflow.jobs.triggerer_job_runner import TriggererJobRunner
-from airflow.listeners.listener import get_listener_manager
 from airflow.models import DagRun, Log, TaskInstance
 from airflow.models.dag_version import DagVersion
 from airflow.models.hitl import HITLDetail
@@ -4084,12 +4083,6 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
     TASK_ID = "print_the_context"
     RUN_ID = "TEST_DAG_RUN_ID"
 
-    @pytest.fixture(autouse=True)
-    def clean_listener_manager(self):
-        get_listener_manager().clear()
-        yield
-        get_listener_manager().clear()
-
     @pytest.mark.parametrize(
         ("state", "listener_state"),
         [
@@ -4098,13 +4091,15 @@ def clean_listener_manager(self):
             ("skipped", []),
         ],
     )
-    def test_patch_task_instance_notifies_listeners(self, test_client, session, state, listener_state):
+    def test_patch_task_instance_notifies_listeners(
+        self, test_client, session, state, listener_state, listener_manager
+    ):
         from unit.listeners.class_listener import ClassBasedListener
 
         self.create_task_instances(session)
 
         listener = ClassBasedListener()
-        get_listener_manager().add_listener(listener)
+        listener_manager(listener)
         test_client.patch(
             self.ENDPOINT_URL,
             json={

airflow-core/tests/unit/assets/test_manager.py

@@ -30,7 +30,6 @@
 
 from airflow import settings
 from airflow.assets.manager import AssetManager
-from airflow.listeners.listener import get_listener_manager
 from airflow.models.asset import (
     AssetAliasModel,
     AssetDagRunQueue,
@@ -183,11 +182,11 @@ def test_register_asset_change_no_downstreams(self, session, mock_task_instance)
         assert session.scalar(select(func.count()).select_from(AssetDagRunQueue)) == 0
 
     def test_register_asset_change_notifies_asset_listener(
-        self, session, mock_task_instance, testing_dag_bundle
+        self, session, mock_task_instance, testing_dag_bundle, listener_manager
     ):
         asset_manager = AssetManager()
         asset_listener.clear()
-        get_listener_manager().add_listener(asset_listener)
+        listener_manager(asset_listener)
 
         bundle_name = "testing"
 
@@ -207,10 +206,10 @@ def test_register_asset_change_notifies_asset_listener(
         assert len(asset_listener.changed) == 1
         assert asset_listener.changed[0].uri == asset.uri
 
-    def test_create_assets_notifies_asset_listener(self, session):
+    def test_create_assets_notifies_asset_listener(self, session, listener_manager):
         asset_manager = AssetManager()
         asset_listener.clear()
-        get_listener_manager().add_listener(asset_listener)
+        listener_manager(asset_listener)
 
         asset = Asset(uri="test://asset1", name="test_asset_1")
 

airflow-core/tests/unit/dag_processing/test_collection.py

@@ -41,7 +41,6 @@
     update_dag_parsing_results_in_db,
 )
 from airflow.exceptions import SerializationError
-from airflow.listeners.listener import get_listener_manager
 from airflow.models import DagModel, DagRun
 from airflow.models.asset import (
     AssetActive,
@@ -321,12 +320,11 @@ def clean_db(self, session):
         clear_db_import_errors()
 
     @pytest.fixture(name="dag_import_error_listener")
-    def _dag_import_error_listener(self):
+    def _dag_import_error_listener(self, listener_manager):
         from unit.listeners import dag_import_error_listener
 
-        get_listener_manager().add_listener(dag_import_error_listener)
+        listener_manager(dag_import_error_listener)
         yield dag_import_error_listener
-        get_listener_manager().clear()
         dag_import_error_listener.clear()
 
     @mark_fab_auth_manager_test