diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_plugins.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_plugins.py index 1d9ba2175fab7..38b2cc8fa9faf 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_plugins.py +++ b/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", diff --git a/providers/apache/kafka/docs/configurations-ref.rst b/providers/apache/kafka/docs/configurations-ref.rst index 80d0d373f8f2a..de52624721a6c 100644 --- a/providers/apache/kafka/docs/configurations-ref.rst +++ b/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_* diff --git a/providers/apache/kafka/provider.yaml b/providers/apache/kafka/provider.yaml index 8e23df573095b..a0ae3a8aebd15 100644 --- a/providers/apache/kafka/provider.yaml +++ b/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: ~ diff --git a/providers/apache/kafka/pyproject.toml b/providers/apache/kafka/pyproject.toml index 3c5ead9a4b80c..b2e3a1fa59880 100644 --- a/providers/apache/kafka/pyproject.toml +++ b/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" diff --git a/providers/apache/kafka/src/airflow/providers/apache/kafka/get_provider_info.py b/providers/apache/kafka/src/airflow/providers/apache/kafka/get_provider_info.py index d8c8510cb5ddb..31162c55425e1 100644 --- a/providers/apache/kafka/src/airflow/providers/apache/kafka/get_provider_info.py +++ b/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, diff --git a/providers/apache/kafka/src/airflow/providers/apache/kafka/plugins/listener.py b/providers/apache/kafka/src/airflow/providers/apache/kafka/plugins/event_producer.py similarity index 92% rename from providers/apache/kafka/src/airflow/providers/apache/kafka/plugins/listener.py rename to providers/apache/kafka/src/airflow/providers/apache/kafka/plugins/event_producer.py index 342b37139a094..20f6f1f6e449e 100644 --- a/providers/apache/kafka/src/airflow/providers/apache/kafka/plugins/listener.py +++ b/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() diff --git a/providers/apache/kafka/tests/integration/apache/kafka/plugins/test_listener.py b/providers/apache/kafka/tests/integration/apache/kafka/plugins/test_event_producer.py similarity index 92% rename from providers/apache/kafka/tests/integration/apache/kafka/plugins/test_listener.py rename to providers/apache/kafka/tests/integration/apache/kafka/plugins/test_event_producer.py index ce900f79b8734..81cbc4ca931e3 100644 --- a/providers/apache/kafka/tests/integration/apache/kafka/plugins/test_listener.py +++ b/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, diff --git a/providers/apache/kafka/tests/unit/apache/kafka/plugins/test_listener.py b/providers/apache/kafka/tests/unit/apache/kafka/plugins/test_event_producer.py similarity index 72% rename from providers/apache/kafka/tests/unit/apache/kafka/plugins/test_listener.py rename to providers/apache/kafka/tests/unit/apache/kafka/plugins/test_event_producer.py index 38bdc0566372f..8e4ed0fcf4ee3 100644 --- a/providers/apache/kafka/tests/unit/apache/kafka/plugins/test_listener.py +++ b/providers/apache/kafka/tests/unit/apache/kafka/plugins/test_event_producer.py @@ -24,8 +24,8 @@ from confluent_kafka import KafkaError from airflow.models import Connection -from airflow.providers.apache.kafka.plugins import listener -from airflow.providers.apache.kafka.plugins.listener import DagRunListener, TaskListener +from airflow.providers.apache.kafka.plugins import event_producer +from airflow.providers.apache.kafka.plugins.event_producer import DagRunListener, TaskListener from tests_common.test_utils.config import conf_vars from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS @@ -48,7 +48,7 @@ @pytest.fixture(autouse=True) def _default_kafka_conn(create_connection_without_db): - # The listener builds its producer through ``KafkaProducerHook``, which requires + # The plugin builds its producer through ``KafkaProducerHook``, which requires # a resolvable ``kafka_default`` connection. create_connection_without_db( Connection( @@ -63,24 +63,24 @@ def _default_kafka_conn(create_connection_without_db): def _clear_cached_values(): """Invalidates caches before each test run.""" # Reset config cache. - listener._dag_run_events_enabled.cache_clear() - listener._task_instance_events_enabled.cache_clear() - listener._get_topic.cache_clear() - listener._get_kafka_config_id.cache_clear() - listener._get_source.cache_clear() - listener._get_dag_run_dag_id_allowlist.cache_clear() - listener._get_dag_run_dag_id_denylist.cache_clear() - listener._get_task_instance_dag_id_allowlist.cache_clear() - listener._get_task_instance_dag_id_denylist.cache_clear() - listener._get_task_instance_task_id_allowlist.cache_clear() - listener._get_task_instance_task_id_denylist.cache_clear() - listener._get_topic_check_timeout.cache_clear() - listener._get_topic_check_retry_interval.cache_clear() + event_producer._dag_run_events_enabled.cache_clear() + event_producer._task_instance_events_enabled.cache_clear() + event_producer._get_topic.cache_clear() + event_producer._get_kafka_config_id.cache_clear() + event_producer._get_source.cache_clear() + event_producer._get_dag_run_dag_id_allowlist.cache_clear() + event_producer._get_dag_run_dag_id_denylist.cache_clear() + event_producer._get_task_instance_dag_id_allowlist.cache_clear() + event_producer._get_task_instance_dag_id_denylist.cache_clear() + event_producer._get_task_instance_task_id_allowlist.cache_clear() + event_producer._get_task_instance_task_id_denylist.cache_clear() + event_producer._get_topic_check_timeout.cache_clear() + event_producer._get_topic_check_retry_interval.cache_clear() # Reset the module-level producer and topic state. - listener._producer = None - listener._topic_exists = False - listener._topic_check_retry_after = 0.0 + event_producer._producer = None + event_producer._topic_exists = False + event_producer._topic_check_retry_after = 0.0 @pytest.fixture @@ -126,7 +126,7 @@ def _assert_common_message_fields(kafka_producer_mock, expected_event: str) -> d assert kwargs["key"] == f"{_DAG_ID}/{_DAG_RUN_ID}".encode() body = json.loads(kwargs["value"].decode("utf-8")) - assert body["schema_version"] == listener.SCHEMA_VERSION + assert body["schema_version"] == event_producer.SCHEMA_VERSION assert body["source"] == _SOURCE assert body["event"] == expected_event assert body["dag_id"] == _DAG_ID @@ -140,22 +140,22 @@ def _assert_common_message_fields(kafka_producer_mock, expected_event: str) -> d [ pytest.param( { - ("kafka_listener", "dag_run_events_enabled"): "True", - ("kafka_listener", "task_instance_events_enabled"): "True", + ("kafka_event_producer", "dag_run_events_enabled"): "True", + ("kafka_event_producer", "task_instance_events_enabled"): "True", }, [DagRunListener, TaskListener], id="both_listeners_enabled", ), pytest.param( { - ("kafka_listener", "dag_run_events_enabled"): "True", + ("kafka_event_producer", "dag_run_events_enabled"): "True", }, [DagRunListener], id="only_dr_events", ), pytest.param( { - ("kafka_listener", "task_instance_events_enabled"): "True", + ("kafka_event_producer", "task_instance_events_enabled"): "True", }, [TaskListener], id="only_ti_events", @@ -169,7 +169,7 @@ def _assert_common_message_fields(kafka_producer_mock, expected_event: str) -> d ) def test_get_enabled_listeners(configs, expected_listener_classes): with conf_vars(configs): - registered_listeners = listener._get_enabled_listeners() + registered_listeners = event_producer._get_enabled_listeners() # Both lists should have the same size. assert len(registered_listeners) == len(expected_listener_classes) @@ -187,9 +187,9 @@ def test_get_enabled_listeners(configs, expected_listener_classes): ) @conf_vars( { - ("kafka_listener", "dag_run_events_enabled"): "True", - ("kafka_listener", "topic"): _KAFKA_TOPIC, - ("kafka_listener", "source"): _SOURCE, + ("kafka_event_producer", "dag_run_events_enabled"): "True", + ("kafka_event_producer", "topic"): _KAFKA_TOPIC, + ("kafka_event_producer", "source"): _SOURCE, } ) def test_produce_dr_message( @@ -224,9 +224,9 @@ def test_produce_dr_message( ) @conf_vars( { - ("kafka_listener", "task_instance_events_enabled"): "True", - ("kafka_listener", "topic"): _KAFKA_TOPIC, - ("kafka_listener", "source"): _SOURCE, + ("kafka_event_producer", "task_instance_events_enabled"): "True", + ("kafka_event_producer", "topic"): _KAFKA_TOPIC, + ("kafka_event_producer", "source"): _SOURCE, } ) def test_produce_ti_message( @@ -261,15 +261,15 @@ class TestGetProducer: @pytest.fixture(autouse=True) def _producer_conf(self): - with conf_vars({("kafka_listener", "topic"): _KAFKA_TOPIC}): + with conf_vars({("kafka_event_producer", "topic"): _KAFKA_TOPIC}): yield def test_producer_cached_on_success(self): kafka_producer_mock = MagicMock() with patch(_PRODUCER_CLS, return_value=kafka_producer_mock) as producer_cls_mock: - producer1 = listener._get_producer() - producer2 = listener._get_producer() + producer1 = event_producer._get_producer() + producer2 = event_producer._get_producer() assert producer1 is kafka_producer_mock assert producer2 is producer1 @@ -277,10 +277,10 @@ def test_producer_cached_on_success(self): def test_init_failure_warns_and_returns_none(self, monkeypatch): log_warning_mock = MagicMock() - monkeypatch.setattr(listener.log, "warning", log_warning_mock) + monkeypatch.setattr(event_producer.log, "warning", log_warning_mock) with patch(_PRODUCER_CLS, side_effect=ValueError("boom")): - assert listener._get_producer() is None + assert event_producer._get_producer() is None log_warning_mock.assert_called_once() warning_format, exc_arg = log_warning_mock.call_args.args @@ -293,30 +293,30 @@ def test_atexit_register(self, monkeypatch): kafka_producer_mock = MagicMock() register_mock = MagicMock() - monkeypatch.setattr(listener.atexit, "register", register_mock) + monkeypatch.setattr(event_producer.atexit, "register", register_mock) with patch(_PRODUCER_CLS, return_value=kafka_producer_mock): - listener._get_producer() + event_producer._get_producer() - register_mock.assert_called_once_with(listener._flush_producer_at_exit, kafka_producer_mock) + register_mock.assert_called_once_with(event_producer._flush_producer_at_exit, kafka_producer_mock) - def test_hook_built_from_listener_config(self): - """The listener section's ``kafka_config_id`` is forwarded to the hook.""" + def test_hook_built_from_event_producer_config(self): + """The plugin section's ``kafka_config_id`` is forwarded to the hook.""" hook_mock = MagicMock() - with conf_vars({("kafka_listener", "kafka_config_id"): "kafka_events"}): + with conf_vars({("kafka_event_producer", "kafka_config_id"): "kafka_events"}): with patch(_PRODUCER_HOOK_CLS, return_value=hook_mock) as hook_cls_mock: - producer = listener._get_producer() + producer = event_producer._get_producer() hook_cls_mock.assert_called_once_with(kafka_config_id="kafka_events") assert producer is hook_mock.get_producer.return_value def test_hook_defaults_when_options_unset(self): - """Without listener options the hook falls back to its own defaults.""" + """Without plugin options the hook falls back to its own defaults.""" hook_mock = MagicMock() with patch(_PRODUCER_HOOK_CLS, return_value=hook_mock) as hook_cls_mock: - listener._get_producer() + event_producer._get_producer() # Unset options must not be passed to the hook at all (e.g. as ""), so the # hook's own defaults (the "kafka_default" connection) apply. @@ -328,7 +328,7 @@ class TestCheckTopicExists: @pytest.fixture(autouse=True) def _topic_conf(self): - with conf_vars({("kafka_listener", "topic"): _KAFKA_TOPIC}): + with conf_vars({("kafka_event_producer", "topic"): _KAFKA_TOPIC}): yield def test_topic_exists_kept_for_process_lifetime(self): @@ -336,9 +336,9 @@ def test_topic_exists_kept_for_process_lifetime(self): kafka_producer_mock.list_topics.return_value.topics.__contains__.return_value = True with patch(_PRODUCER_CLS, return_value=kafka_producer_mock): - assert listener._check_topic_exists() is True + assert event_producer._check_topic_exists() is True # Once confirmed, the broker is not queried again. - assert listener._check_topic_exists() is True + assert event_producer._check_topic_exists() is True kafka_producer_mock.list_topics.assert_called_once() @@ -347,29 +347,29 @@ def test_topic_doesnt_exist(self, monkeypatch): kafka_producer_mock.list_topics.return_value.topics.__contains__.return_value = False log_warning_mock = MagicMock() - monkeypatch.setattr(listener.log, "warning", log_warning_mock) + monkeypatch.setattr(event_producer.log, "warning", log_warning_mock) with patch(_PRODUCER_CLS, return_value=kafka_producer_mock): - assert listener._check_topic_exists() is False + assert event_producer._check_topic_exists() is False log_warning_mock.assert_called_once() # Format string + the two formatting args are passed positionally to log.warning. warning_format, topic_arg, retry_interval_arg = log_warning_mock.call_args.args assert "topic %r not found on the broker" in warning_format assert topic_arg == _KAFKA_TOPIC - assert retry_interval_arg == listener._get_topic_check_retry_interval() + assert retry_interval_arg == event_producer._get_topic_check_retry_interval() def test_list_topics_failure_warns_and_sets_cooldown(self, monkeypatch): kafka_producer_mock = MagicMock() kafka_producer_mock.list_topics.side_effect = ValueError("broker down") log_warning_mock = MagicMock() - monkeypatch.setattr(listener.log, "warning", log_warning_mock) + monkeypatch.setattr(event_producer.log, "warning", log_warning_mock) with patch(_PRODUCER_CLS, return_value=kafka_producer_mock): - assert listener._check_topic_exists() is False + assert event_producer._check_topic_exists() is False # Within the cooldown window the check is not retried. - assert listener._check_topic_exists() is False + assert event_producer._check_topic_exists() is False kafka_producer_mock.list_topics.assert_called_once() log_warning_mock.assert_called_once() @@ -377,7 +377,7 @@ def test_list_topics_failure_warns_and_sets_cooldown(self, monkeypatch): def test_retried_after_failure_cooldown(self, monkeypatch): time_now = [1000.0] - monkeypatch.setattr(listener.time, "monotonic", lambda: time_now[0]) + monkeypatch.setattr(event_producer.time, "monotonic", lambda: time_now[0]) kafka_producer_mock = MagicMock() # The topic doesn't exist yet. @@ -385,36 +385,36 @@ def test_retried_after_failure_cooldown(self, monkeypatch): with patch(_PRODUCER_CLS, return_value=kafka_producer_mock): # First check fails: the topic is missing. - assert listener._check_topic_exists() is False + assert event_producer._check_topic_exists() is False assert kafka_producer_mock.list_topics.call_count == 1 # The interval hasn't passed and the check isn't retried. time_now[0] += 1 - assert listener._check_topic_exists() is False + assert event_producer._check_topic_exists() is False assert kafka_producer_mock.list_topics.call_count == 1 # Past the interval. The check is retried and succeeds this time. kafka_producer_mock.list_topics.return_value.topics.__contains__.return_value = True - time_now[0] += listener._get_topic_check_retry_interval() + 1 - assert listener._check_topic_exists() is True + time_now[0] += event_producer._get_topic_check_retry_interval() + 1 + assert event_producer._check_topic_exists() is True assert kafka_producer_mock.list_topics.call_count == 2 def test_producer_init_failure_sets_cooldown(self, monkeypatch): time_now = [1000.0] - monkeypatch.setattr(listener.time, "monotonic", lambda: time_now[0]) + monkeypatch.setattr(event_producer.time, "monotonic", lambda: time_now[0]) with patch(_PRODUCER_CLS, side_effect=ValueError("boom")) as producer_cls_mock: - assert listener._check_topic_exists() is False + assert event_producer._check_topic_exists() is False assert producer_cls_mock.call_count == 1 # The interval hasn't passed and the producer initialization isn't retried. time_now[0] += 1 - assert listener._check_topic_exists() is False + assert event_producer._check_topic_exists() is False assert producer_cls_mock.call_count == 1 # Past the interval. The producer initialization is retried. - time_now[0] += listener._get_topic_check_retry_interval() + 1 - assert listener._check_topic_exists() is False + time_now[0] += event_producer._get_topic_check_retry_interval() + 1 + assert event_producer._check_topic_exists() is False assert producer_cls_mock.call_count == 2 @pytest.mark.parametrize( @@ -434,27 +434,27 @@ def test_delivery_report_invalidates_topic_confirmation_only_on_missing_topic( """ time_now = [1000.0] - monkeypatch.setattr(listener.time, "monotonic", lambda: time_now[0]) + monkeypatch.setattr(event_producer.time, "monotonic", lambda: time_now[0]) kafka_producer_mock = MagicMock() kafka_producer_mock.list_topics.return_value.topics.__contains__.return_value = True with patch(_PRODUCER_CLS, return_value=kafka_producer_mock): - assert listener._check_topic_exists() is True - assert listener._topic_exists is True + assert event_producer._check_topic_exists() is True + assert event_producer._topic_exists is True err = MagicMock() err.code.return_value = kafka_error - listener._on_delivery(err, MagicMock()) + event_producer._on_delivery(err, MagicMock()) - assert listener._topic_exists is topic_exists_expected + assert event_producer._topic_exists is topic_exists_expected if not topic_exists_expected: # Cooldown holds off the immediate re-check. - assert listener._check_topic_exists() is False + assert event_producer._check_topic_exists() is False # After the cooldown, the check re-verifies; the topic is still on the broker, # so confirmation is restored. - time_now[0] += listener._get_topic_check_retry_interval() + 1 - assert listener._check_topic_exists() is True + time_now[0] += event_producer._get_topic_check_retry_interval() + 1 + assert event_producer._check_topic_exists() is True class TestFilters: @@ -464,9 +464,9 @@ class TestFilters: def _filters_conf(self): with conf_vars( { - ("kafka_listener", "dag_run_events_enabled"): "True", - ("kafka_listener", "task_instance_events_enabled"): "True", - ("kafka_listener", "topic"): _KAFKA_TOPIC, + ("kafka_event_producer", "dag_run_events_enabled"): "True", + ("kafka_event_producer", "task_instance_events_enabled"): "True", + ("kafka_event_producer", "topic"): _KAFKA_TOPIC, } ): yield @@ -482,55 +482,55 @@ def _filters_conf(self): ], ) def test_id_is_allowed(self, id_to_check, allowlist, denylist, expected_result): - assert listener._id_is_allowed(id_to_check, allowlist, denylist) is expected_result + assert event_producer._id_is_allowed(id_to_check, allowlist, denylist) is expected_result @conf_vars( { - ("kafka_listener", "task_instance_dag_id_denylist"): "dag1_*", + ("kafka_event_producer", "task_instance_dag_id_denylist"): "dag1_*", } ) def test_dag_run_filter_ignores_ti_lists(self): # The dag_id lists for DR events, are separate from the dag_id lists for TI events. # The dag_id denylist for task instances would match "dag1_task1" but DR filter must ignore it. - assert listener._dag_run_event_allowed("dag1_task1") is True + assert event_producer._dag_run_event_allowed("dag1_task1") is True @conf_vars( { - ("kafka_listener", "dag_run_dag_id_denylist"): "dag1_*", + ("kafka_event_producer", "dag_run_dag_id_denylist"): "dag1_*", } ) def test_task_instance_filter_ignores_dr_lists(self): # The dag_id lists for DR events, are separate from the dag_id lists for TI events. # DR denylist would match "dag1_task1" but TI filter must ignore it. - assert listener._task_instance_event_allowed("dag1_task1", "load") is True + assert event_producer._task_instance_event_allowed("dag1_task1", "load") is True @conf_vars( { - ("kafka_listener", "task_instance_dag_id_allowlist"): "dag1_*", - ("kafka_listener", "task_instance_task_id_denylist"): "*_cleanup", + ("kafka_event_producer", "task_instance_dag_id_allowlist"): "dag1_*", + ("kafka_event_producer", "task_instance_task_id_denylist"): "*_cleanup", } ) def test_task_instance_requires_both_dag_id_and_task_id(self): # dag_id passes (matches allowlist), task_id passes (no deny match). - assert listener._task_instance_event_allowed("dag1_task1", "task2") is True + assert event_producer._task_instance_event_allowed("dag1_task1", "task2") is True # dag_id passes, task_id denied. - assert listener._task_instance_event_allowed("dag1_task1", "task2_cleanup") is False + assert event_producer._task_instance_event_allowed("dag1_task1", "task2_cleanup") is False # dag_id denied (no allowlist match), task_id passes. - assert listener._task_instance_event_allowed("other_dag", "load") is False + assert event_producer._task_instance_event_allowed("other_dag", "load") is False - @conf_vars({("kafka_listener", "dag_run_dag_id_denylist"): _DAG_ID}) + @conf_vars({("kafka_event_producer", "dag_run_dag_id_denylist"): _DAG_ID}) def test_dr_filter_blocks_emission(self, dr_mock, kafka_producer_mock): DagRunListener().on_dag_run_running(dag_run=dr_mock, msg=None) kafka_producer_mock.produce.assert_not_called() - @conf_vars({("kafka_listener", "task_instance_dag_id_denylist"): _DAG_ID}) + @conf_vars({("kafka_event_producer", "task_instance_dag_id_denylist"): _DAG_ID}) def test_ti_filter_blocks_emission_via_dag_id(self, ti_mock, kafka_producer_mock): TaskListener().on_task_instance_running( previous_state="queued", task_instance=ti_mock, **_TI_SESSION_KWARG ) kafka_producer_mock.produce.assert_not_called() - @conf_vars({("kafka_listener", "task_instance_task_id_denylist"): _TASK_ID}) + @conf_vars({("kafka_event_producer", "task_instance_task_id_denylist"): _TASK_ID}) def test_ti_filter_blocks_emission_via_task_id(self, ti_mock, kafka_producer_mock): TaskListener().on_task_instance_running( previous_state="queued", task_instance=ti_mock, **_TI_SESSION_KWARG diff --git a/scripts/ci/docker-compose/integration-kafka.yml b/scripts/ci/docker-compose/integration-kafka.yml index f0ba7fab7d619..da67875dfb791 100644 --- a/scripts/ci/docker-compose/integration-kafka.yml +++ b/scripts/ci/docker-compose/integration-kafka.yml @@ -60,9 +60,9 @@ services: airflow: environment: - INTEGRATION_KAFKA=true - - AIRFLOW__KAFKA_LISTENER__DAG_RUN_EVENTS_ENABLED=True - - AIRFLOW__KAFKA_LISTENER__TASK_INSTANCE_EVENTS_ENABLED=True - - AIRFLOW__KAFKA_LISTENER__TOPIC=airflow.events - - AIRFLOW__KAFKA_LISTENER__SOURCE=dev-breeze + - AIRFLOW__KAFKA_EVENT_PRODUCER__DAG_RUN_EVENTS_ENABLED=True + - AIRFLOW__KAFKA_EVENT_PRODUCER__TASK_INSTANCE_EVENTS_ENABLED=True + - AIRFLOW__KAFKA_EVENT_PRODUCER__TOPIC=airflow.events + - AIRFLOW__KAFKA_EVENT_PRODUCER__SOURCE=dev-breeze depends_on: - broker