Rename Kafka Listener to Kafka Event Producer and update the documentation

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

@@ -50,7 +50,7 @@ class TestGetPlugins:
                     "edge_executor",
                     "hitl_review",
                     "hive",
-                    "kafka_listener",
+                    "kafka_event_producer",
                     "plugin-a",
                     "plugin-b",
                     "plugin-c",

providers/apache/kafka/docs/configurations-ref.rst

@@ -24,27 +24,41 @@
 Highlighted configurations
 ===========================
 
-The ``[kafka_listener]`` section configures the ``KafkaListenerPlugin``,
+The ``[kafka_event_producer]`` section configures the ``KafkaEventProducerPlugin``,
 which publishes Airflow DagRun and TaskInstance state-change events to a
 Kafka topic. DagRun and TaskInstance events are separated and enabled by
 distinct flags. Both event-type flags default to ``False``.
 
-.. _configuration_kafka_listener_activation:kafka:
+.. _configuration_kafka_event_producer_use_cases:kafka:
 
-Activating the listener
------------------------
+Common use-cases
+----------------
+
+  * Consume the Kafka events by an external observability or analytics tool and gather info about the state
+    of multiple Airflow instances without polling their metadata DBs.
+  * Based on the state of a DagRun, trigger a downstream external system/pipeline (notifications, alerting,
+    cross-team handoffs) without direct interaction with Airflow.
+  * Coordinate Dags across multiple Airflow instances over a shared Kafka service.
+
+    * For example, team_A with Airflow instance_A has a deferred task which is triggered
+      when a task from team_B with Airflow instance_B finishes.
+
+.. _configuration_kafka_event_producer_activation:kafka:
+
+Activating the plugin
+---------------------
 
 To enable event publishing you need to
 
   * enable at least one event-type flag
-  * point the listener at an Airflow Kafka connection via ``kafka_config_id``
+  * point the plugin at an Airflow Kafka connection via ``kafka_config_id``
     (defaults to ``kafka_default``) that carries the broker address and any
     other confluent-kafka client options on its extras
   * have a pre-existing kafka topic
 
 .. code-block:: ini
 
-    [kafka_listener]
+    [kafka_event_producer]
     dag_run_events_enabled = True
     task_instance_events_enabled = True
     kafka_config_id = kafka_events
@@ -68,20 +82,20 @@ Environment-variable equivalents:
 
 .. code-block:: ini
 
-    AIRFLOW__KAFKA_LISTENER__DAG_RUN_EVENTS_ENABLED=True
-    AIRFLOW__KAFKA_LISTENER__TASK_INSTANCE_EVENTS_ENABLED=True
-    AIRFLOW__KAFKA_LISTENER__KAFKA_CONFIG_ID=kafka_events
-    AIRFLOW__KAFKA_LISTENER__TOPIC=airflow.events
+    AIRFLOW__KAFKA_EVENT_PRODUCER__DAG_RUN_EVENTS_ENABLED=True
+    AIRFLOW__KAFKA_EVENT_PRODUCER__TASK_INSTANCE_EVENTS_ENABLED=True
+    AIRFLOW__KAFKA_EVENT_PRODUCER__KAFKA_CONFIG_ID=kafka_events
+    AIRFLOW__KAFKA_EVENT_PRODUCER__TOPIC=airflow.events
 
 The two event flags are independent, users can opt-in to get only DagRun
 event messages or only TaskInstance event messages or both.
 
 The topic must already exist on the broker, it's not auto-created. On a missing
-topic, broker connection failure, or any other producer init error, the listener
+topic, broker connection failure, or any other producer init error, the plugin
 doesn't fail, instead it logs a warning and retries the init after ``topic_check_retry_interval``
-seconds (default ``60``). Once the topic is created on the broker the listener will pick it up.
+seconds (default ``60``). Once the topic is created on the broker the plugin will pick it up.
 
-.. _configuration_kafka_listener_filtering:kafka:
+.. _configuration_kafka_event_producer_filtering:kafka:
 
 Filtering events
 ----------------
@@ -92,7 +106,7 @@ comma-separated list of ``fnmatch`` glob patterns; an empty list means
 
 .. code-block:: ini
 
-    [kafka_listener]
+    [kafka_event_producer]
     dag_run_dag_id_allowlist = sales_*,marketing_*
     dag_run_dag_id_denylist = sales_internal_*
 

providers/apache/kafka/provider.yaml

@@ -131,14 +131,14 @@ queues:
   - airflow.providers.apache.kafka.queues.kafka.KafkaMessageQueueProvider
 
 plugins:
-  - name: kafka_listener
-    plugin-class: airflow.providers.apache.kafka.plugins.listener.KafkaListenerPlugin
+  - name: kafka_event_producer
+    plugin-class: airflow.providers.apache.kafka.plugins.event_producer.KafkaEventProducerPlugin
 
 config:
-  kafka_listener:
+  kafka_event_producer:
     description: |
-      Settings for the Kafka listener that publishes Airflow DagRun and
-      TaskInstance state-change events to a Kafka topic.
+      Settings for the Kafka event producer plugin that publishes Airflow
+      DagRun and TaskInstance state-change events to a Kafka topic.
     options:
       dag_run_events_enabled:
         description: |
@@ -161,7 +161,7 @@ config:
         default: "False"
       kafka_config_id:
         description: |
-          Airflow connection used to build the listener's Kafka producer.
+          Airflow connection used to build the plugin's Kafka producer.
           When unset, the producer hook falls back to its default
           connection (``kafka_default``).
         version_added: 1.14.1
@@ -170,8 +170,8 @@ config:
         default: ""
       topic:
         description: |
-          Topic the listener publishes events to. The topic must already
-          exist on the broker; the listener will not auto-create it.
+          Topic the plugin publishes events to. The topic must already
+          exist on the broker; the plugin will not auto-create it.
         version_added: 1.14.1
         type: string
         example: ~

providers/apache/kafka/pyproject.toml

@@ -131,7 +131,7 @@ apache-airflow-providers-standard = {workspace = true}
 provider_info = "airflow.providers.apache.kafka.get_provider_info:get_provider_info"
 
 [project.entry-points."airflow.plugins"]
-kafka_listener = "airflow.providers.apache.kafka.plugins.listener:KafkaListenerPlugin"
+kafka_event_producer = "airflow.providers.apache.kafka.plugins.event_producer:KafkaEventProducerPlugin"
 
 [tool.flit.module]
 name = "airflow.providers.apache.kafka"

providers/apache/kafka/src/airflow/providers/apache/kafka/get_provider_info.py

@@ -103,13 +103,13 @@ def get_provider_info():
         "queues": ["airflow.providers.apache.kafka.queues.kafka.KafkaMessageQueueProvider"],
         "plugins": [
             {
-                "name": "kafka_listener",
-                "plugin-class": "airflow.providers.apache.kafka.plugins.listener.KafkaListenerPlugin",
+                "name": "kafka_event_producer",
+                "plugin-class": "airflow.providers.apache.kafka.plugins.event_producer.KafkaEventProducerPlugin",
             }
         ],
         "config": {
-            "kafka_listener": {
-                "description": "Settings for the Kafka listener that publishes Airflow DagRun and\nTaskInstance state-change events to a Kafka topic.\n",
+            "kafka_event_producer": {
+                "description": "Settings for the Kafka event producer plugin that publishes Airflow\nDagRun and TaskInstance state-change events to a Kafka topic.\n",
                 "options": {
                     "dag_run_events_enabled": {
                         "description": "Publish DagRun state-change events (``dag_run.running``,\n``dag_run.success``, ``dag_run.failed``). When False the\nDagRun listener is not registered.\n",
@@ -126,14 +126,14 @@ def get_provider_info():
                         "default": "False",
                     },
                     "kafka_config_id": {
-                        "description": "Airflow connection used to build the listener's Kafka producer.\nWhen unset, the producer hook falls back to its default\nconnection (``kafka_default``).\n",
+                        "description": "Airflow connection used to build the plugin's Kafka producer.\nWhen unset, the producer hook falls back to its default\nconnection (``kafka_default``).\n",
                         "version_added": "1.14.1",
                         "type": "string",
                         "example": "kafka_default",
                         "default": "",
                     },
                     "topic": {
-                        "description": "Topic the listener publishes events to. The topic must already\nexist on the broker; the listener will not auto-create it.\n",
+                        "description": "Topic the plugin publishes events to. The topic must already\nexist on the broker; the plugin will not auto-create it.\n",
                         "version_added": "1.14.1",
                         "type": "string",
                         "example": None,

providers/apache/kafka/src/airflow/providers/apache/kafka/plugins/listener.py → providers/apache/kafka/src/airflow/providers/apache/kafka/plugins/event_producer.py

@@ -40,7 +40,7 @@
 
 log = logging.getLogger(__name__)
 
-CONFIG_SECTION = "kafka_listener"
+CONFIG_SECTION = "kafka_event_producer"
 SCHEMA_VERSION = 1
 
 
@@ -152,7 +152,7 @@ def _task_instance_event_allowed(dag_id: str, task_id: str) -> bool:
     )
 
 
-# Listener-owned producer and topic state for the lifetime of the process.
+# Plugin-owned producer and topic state for the lifetime of the process.
 # The producer is built once via KafkaProducerHook and kept for the process lifetime;
 # the topic flags track whether the topic exists on the broker and whether we're
 # currently in a back-off window after a failed topic check.
@@ -179,7 +179,7 @@ def _reset_state_after_fork() -> None:
 
 
 def _get_producer() -> Producer | None:
-    """Build (once) and return the listener's producer, or ``None`` on init failure."""
+    """Build (once) and return the plugin's Kafka producer, or ``None`` on init failure."""
     global _producer
     if _producer is not None:
         return _producer
@@ -194,7 +194,7 @@ def _get_producer() -> Producer | None:
             hook_kwargs["kafka_config_id"] = _get_kafka_config_id()
         producer = KafkaProducerHook(**hook_kwargs).get_producer()
     except Exception as exc:
-        log.warning("Kafka listener: failed to initialize producer (%s).", exc)
+        log.warning("Kafka event producer: failed to initialize producer (%s).", exc)
         return None
 
     atexit.register(_flush_producer_at_exit, producer)
@@ -224,7 +224,7 @@ def _check_topic_exists() -> bool:
         topics = producer.list_topics(timeout=_get_topic_check_timeout()).topics
     except Exception as exc:
         log.warning(
-            "Kafka listener: topic check failed (%s). Will retry after %ds.",
+            "Kafka event producer: topic check failed (%s). Will retry after %ds.",
             exc,
             _get_topic_check_retry_interval(),
         )
@@ -233,7 +233,7 @@ def _check_topic_exists() -> bool:
 
     if _get_topic() not in topics:
         log.warning(
-            "Kafka listener: topic %r not found on the broker. Will retry after %ds. "
+            "Kafka event producer: topic %r not found on the broker. Will retry after %ds. "
             "Create the topic on the broker to enable publishing.",
             _get_topic(),
             _get_topic_check_retry_interval(),
@@ -242,7 +242,7 @@ def _check_topic_exists() -> bool:
         return False
 
     log.info(
-        "Kafka listener attached: pid=%s source=%r topic=%r",
+        "Kafka event producer attached: pid=%s source=%r topic=%r",
         os.getpid(),
         _get_source(),
         _get_topic(),
@@ -255,16 +255,16 @@ def _flush_producer_at_exit(producer: Producer) -> None:
     try:
         producer.flush(5)
     except Exception:
-        log.debug("Kafka listener: error flushing producer on exit", exc_info=True)
+        log.debug("Kafka event producer: error flushing producer on exit", exc_info=True)
 
 
 def _on_delivery(err, _msg) -> None:
     if err is None:
         return
-    log.warning("Kafka listener: delivery failed: %s", err)
+    log.warning("Kafka event producer: delivery failed: %s", err)
     # If the broker or local metadata says the topic is gone, the confirmation cached at
     # attach-time is stale. Flip it back so the next _check_topic_exists re-verifies with
-    # the broker, gated by the same cooldown used elsewhere. Lets the listener auto-recover
+    # the broker, gated by the same cooldown used elsewhere. Lets the plugin auto-recover
     # if the topic gets recreated instead of requiring a component restart.
     from confluent_kafka import KafkaError
 
@@ -291,7 +291,7 @@ def _produce_dr_message(event: str, dag_run: DagRun, msg: str) -> None:
             _get_dr_payload(dag_run, msg),
         )
     except Exception:
-        log.exception("Kafka listener: %s failed", event)
+        log.exception("Kafka event producer: %s failed", event)
 
 
 def _get_dr_payload(dag_run, msg) -> dict[str, Any]:
@@ -325,7 +325,7 @@ def _produce_ti_message(
             _get_ti_payload(task_instance, previous_state, error=error),
         )
     except Exception:
-        log.exception("Kafka listener: %s failed", event)
+        log.exception("Kafka event producer: %s failed", event)
 
 
 def _get_ti_payload(ti, previous_state, error=None) -> dict[str, Any]:
@@ -373,9 +373,12 @@ def _produce_message(event: str, dag_id: str, run_id: str, payload: dict[str, An
         )
         producer.poll(0)
     except Exception as ex:
-        log.warning("Kafka listener: failed to enqueue %s for %s/%s: %s", event, dag_id, run_id, ex)
+        log.warning("Kafka event producer: failed to enqueue %s for %s/%s: %s", event, dag_id, run_id, ex)
 
 
+# DagRunListener / TaskListener are pluggy hookimpl classes that plug into Airflow's
+# listener API (``AirflowPlugin.listeners``). These classes listen for Airflow events
+# and then produce a message for every event.
 class DagRunListener:
     """Publishes DagRun state-change event messages to Kafka."""
 
@@ -463,8 +466,8 @@ def _get_enabled_listeners() -> list[object]:
     return listeners
 
 
-class KafkaListenerPlugin(AirflowPlugin):
+class KafkaEventProducerPlugin(AirflowPlugin):
     """Publishes Airflow DagRun and TaskInstance event messages to a defined Kafka topic."""
 
-    name = "kafka_listener"
+    name = "kafka_event_producer"
     listeners = _get_enabled_listeners()

providers/apache/kafka/tests/integration/apache/kafka/plugins/test_listener.py → providers/apache/kafka/tests/integration/apache/kafka/plugins/test_event_producer.py

@@ -42,7 +42,7 @@
 
 client_config = {
     "bootstrap.servers": "broker:29092",
-    "group.id": "kafka-listener-integration-test",
+    "group.id": "kafka-event-producer-integration-test",
     "enable.auto.commit": False,
     "auto.offset.reset": "earliest",
 }
@@ -59,7 +59,7 @@ def _wait_for_assignment(consumer, timeout: float = 10.0) -> None:
 
 @pytest.mark.integration("kafka")
 @pytest.mark.backend("postgres")
-class TestEventListener:
+class TestEventProducer:
     test_dir = os.path.dirname(os.path.abspath(__file__))
     dag_folder = os.path.join(test_dir, "dags")
 
@@ -83,12 +83,12 @@ def setup_class(cls):
         os.environ["AIRFLOW__CORE__LOAD_EXAMPLES"] = "False"
         os.environ["AIRFLOW__CORE__UNIT_TEST_MODE"] = "False"
 
-        os.environ["AIRFLOW__KAFKA_LISTENER__DAG_RUN_EVENTS_ENABLED"] = "True"
-        os.environ["AIRFLOW__KAFKA_LISTENER__TASK_INSTANCE_EVENTS_ENABLED"] = "True"
-        os.environ["AIRFLOW__KAFKA_LISTENER__TOPIC"] = cls.TOPIC
-        os.environ["AIRFLOW__KAFKA_LISTENER__SOURCE"] = "dev-breeze"
+        os.environ["AIRFLOW__KAFKA_EVENT_PRODUCER__DAG_RUN_EVENTS_ENABLED"] = "True"
+        os.environ["AIRFLOW__KAFKA_EVENT_PRODUCER__TASK_INSTANCE_EVENTS_ENABLED"] = "True"
+        os.environ["AIRFLOW__KAFKA_EVENT_PRODUCER__TOPIC"] = cls.TOPIC
+        os.environ["AIRFLOW__KAFKA_EVENT_PRODUCER__SOURCE"] = "dev-breeze"
 
-        # Shared Kafka connection: used by the listener producer, the topic
+        # Shared Kafka connection: used by the event producer plugin, the topic
         # creation/deletion, and the consumer.
         kafka_default_conn = Connection(
             conn_id=cls.KAFKA_CONFIG_ID,