You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "xBis7 (via GitHub)" <gi...@apache.org> on 2026/06/05 14:41:00 UTC
[PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
xBis7 opened a new pull request, #68082:
URL: https://github.com/apache/airflow/pull/68082
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
This PR is adding a listener under the Kafka provider which provides an implementation for the DagRun and TaskInstance state change event hooks.
The listener publishes a message to a pre-existing Kafka topic for every event. Each message has some extra metadata for the dag or task that the event belongs to.
DagRun events and TaskInstance events are separated and guarded behind different config flags. Both flags are disabled by default because each listener adds load to Airflow that users might not want.
The listener expects that the user has already created the topic and then defines it in the listener's config section. In case the topic doesn't exist, the listener doesn't fail. Instead it logs a warning and checks for the topic existence again after a configured interval.
Users can also filter events for dag runs based on dag_id and for tasks based on dag_id + task_id. For example, someone could choose to get only dag run state events for a particular dag and nothing else.
By adding a listener to the provider we can have dags consume from the Kafka topic and react to messages. When a particular message arrives, a deferred task could be triggered. This could also work for dags across multiple Airflow installations, all sharing the common topic. For that reason, I also added a `source` config key, so that users can distinguish where the messages came from.
## Testing
I've added unit tests and also a simple integration test that uses an actual broker. I've also tested the changes manually.
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [X] Yes (please specify the tool below)
Claude Code, Opus 4.7
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3364164916
##########
airflow-core/tests/integration/otel/test_otel.py:
##########
Review Comment:
Yes - else would be good to split the core changes/test utils in one PR (and backport them) and separate the kafka changes - because kafka has movved a long time since we split the release branch. And kafka is always released from main.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3567066652
##########
providers/apache/kafka/src/airflow/providers/apache/kafka/hooks/base.py:
##########
@@ -58,18 +64,36 @@ class KafkaBaseHook(BaseHook):
"""
A base hook for interacting with Apache Kafka.
+ The Airflow connection's extra JSON is merged with the ``config_dict``.
+ Each ``config_dict`` key overrides the same key from the connection.
+ When the connection is not available and ``config_dict`` contains the key
+ ``bootstrap.servers``, the client is built from ``config_dict`` alone
+ without requiring a connection.
+
:param kafka_config_id: The connection object to use, defaults to "kafka_default"
+ :param config_dict: Optional confluent-kafka client configuration
+ (e.g. ``{"bootstrap.servers": "localhost:9092"}``). Each key overrides the same
+ key from the connection's config. Callback options (``error_cb``, ``throttle_cb``,
+ ``stats_cb``, ``log_cb``, ``oauth_cb``, ``on_commit``) may be callables or
+ dotted-path strings, which are resolved before the client is built.
"""
conn_name_attr = "kafka_config_id"
default_conn_name = "kafka_default"
conn_type = "kafka"
hook_name = "Apache Kafka"
- def __init__(self, kafka_config_id=default_conn_name, *args, **kwargs):
+ def __init__(
+ self,
+ kafka_config_id=default_conn_name,
+ config_dict: dict[str, Any] | None = None,
Review Comment:
I added the param just for convenience and no other serious reason. I can see why you are concerned. I removed it and updated the patch and the tests.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #68082:
URL: https://github.com/apache/airflow/pull/68082#issuecomment-4994637353
I went through the PR and here is what I think. This is not a replacement or alternative of asset watcher, they serve 2 different purposes:
- Asset watcher. Schedule/trigger dags based on external event such as a message in Kafka topic
- This PR. Emit events to kafka topic on dag run and task instance state change
The name listener is definitely confusing so we should change that. Here are some idea on how we could call it:
- Event publisher
- Event exporter
- Event forwarder
- Event reporter
Not related but side comment on this one. If we know this is a business case needed by users we should make it a Airflow feature (with providers being able to hook their implementation such as Kafka) as opposed to a Kafka feature.
@vikramkoka
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "vikramkoka (via GitHub)" <gi...@apache.org>.
vikramkoka commented on PR #68082:
URL: https://github.com/apache/airflow/pull/68082#issuecomment-4634949660
I am confused by the use case here.
Shouldn't this be an Asset Watcher?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3403774224
##########
providers/apache/kafka/src/airflow/providers/apache/kafka/plugins/listener.py:
##########
@@ -0,0 +1,440 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import atexit
+import json
+import logging
+import os
+import time
+from datetime import datetime, timezone
+from fnmatch import fnmatch
+from functools import lru_cache
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.apache.kafka.version_compat import AIRFLOW_V_3_0_PLUS
+from airflow.providers.common.compat.sdk import AirflowPlugin, conf, hookimpl
+from airflow.utils.net import get_hostname
+
+if TYPE_CHECKING:
+ from confluent_kafka import Producer
+ from sqlalchemy.orm import Session
+
+ from airflow.models import DagRun, TaskInstance
+ from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
+ from airflow.utils.state import TaskInstanceState
+
+log = logging.getLogger(__name__)
+
+CONFIG_SECTION = "kafka_listener"
+SCHEMA_VERSION = 1
+
+
+@lru_cache(maxsize=1)
+def _dag_run_events_enabled() -> bool:
+ return conf.getboolean(CONFIG_SECTION, "dag_run_events_enabled", fallback="False")
+
+
+@lru_cache(maxsize=1)
+def _task_instance_events_enabled() -> bool:
+ return conf.getboolean(CONFIG_SECTION, "task_instance_events_enabled", fallback="False")
+
+
+@lru_cache(maxsize=1)
+def _get_topic() -> str:
+ return conf.get(CONFIG_SECTION, "topic", fallback="airflow.events")
+
+
+@lru_cache(maxsize=1)
+def _get_bootstrap_servers() -> str:
+ return conf.get(CONFIG_SECTION, "bootstrap_servers", fallback="").strip()
+
+
+@lru_cache(maxsize=1)
+def _get_source() -> str:
+ source_from_conf = conf.get(CONFIG_SECTION, "source", fallback="")
+ # Default to the current hostname, if un-set.
+ return source_from_conf or get_hostname()
+
+
+def _parse_filter_patterns_to_tuple(raw: str) -> tuple[str, ...]:
+ return tuple(p.strip() for p in raw.split(",") if p.strip())
+
+
+@lru_cache(maxsize=1)
+def _get_dag_run_dag_id_allowlist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(conf.get(CONFIG_SECTION, "dag_run_dag_id_allowlist", fallback=""))
+
+
+@lru_cache(maxsize=1)
+def _get_dag_run_dag_id_denylist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(conf.get(CONFIG_SECTION, "dag_run_dag_id_denylist", fallback=""))
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_dag_id_allowlist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_dag_id_allowlist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_dag_id_denylist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_dag_id_denylist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_task_id_allowlist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_task_id_allowlist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_task_id_denylist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_task_id_denylist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_topic_check_timeout() -> int:
+ return conf.getint(CONFIG_SECTION, "topic_check_timeout", fallback="10")
+
+
+@lru_cache(maxsize=1)
+def _get_topic_check_retry_interval() -> int:
+ return conf.getint(CONFIG_SECTION, "topic_check_retry_interval", fallback="60")
+
+
+def _id_is_allowed(id_to_check: str, allowlist: tuple[str, ...], denylist: tuple[str, ...]) -> bool:
+ """Deny takes precedence; empty allowlist means 'allow all'."""
+ if denylist and any(fnmatch(id_to_check, id_pattern) for id_pattern in denylist):
+ return False
+ if allowlist and not any(fnmatch(id_to_check, id_pattern) for id_pattern in allowlist):
+ return False
+ return True
+
+
+def _dag_run_event_allowed(dag_id: str) -> bool:
+ return _id_is_allowed(
+ dag_id,
+ _get_dag_run_dag_id_allowlist(),
+ _get_dag_run_dag_id_denylist(),
+ )
+
+
+def _task_instance_event_allowed(dag_id: str, task_id: str) -> bool:
+ return _id_is_allowed(
+ dag_id,
+ _get_task_instance_dag_id_allowlist(),
+ _get_task_instance_dag_id_denylist(),
+ ) and _id_is_allowed(
+ task_id,
+ _get_task_instance_task_id_allowlist(),
+ _get_task_instance_task_id_denylist(),
+ )
+
+
+# confluent_kafka.Producer is not fork-safe — its background threads and
+# broker sockets do not survive ``os.fork``. Keep at most one Producer per
+# process and reset the cached state in the child so the next call re-inits.
+# If the initialization is successful, then the value is cached,
+# otherwise init is retried on an interval.
+_producer: Producer | None = None
+_producer_retry_after: float = 0.0
+
+
+def _reset_producer_after_fork() -> None:
+ """Drop any inherited producer in the child; the child re-inits on next call."""
+ global _producer, _producer_retry_after
+ _producer = None
+ _producer_retry_after = 0.0
+
+
+os.register_at_fork(after_in_child=_reset_producer_after_fork)
+
+
+def _get_producer() -> Producer | None:
+ global _producer, _producer_retry_after
+ # If there is a cached producer, return it, don't re-initialize.
+ if _producer is not None or time.monotonic() < _producer_retry_after:
+ return _producer
+
+ result: Producer | None = None
+ try:
+ # The plugin is loaded by every Airflow process. The Producer is a heavy dependency, and
+ # by lazily importing here, we avoid the unneeded startup cost.
+ from confluent_kafka import Producer
+
+ # No need to check if bootstrap_servers are set, because it's done when registering the listeners.
+ # If not set, then there won't be any listeners set and this part of the code will never execute.
+ bootstrap_servers = _get_bootstrap_servers()
+ producer = Producer({"bootstrap.servers": bootstrap_servers})
Review Comment:
I've reworked the approach and now it uses the producer hook directly.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3567114241
##########
providers/apache/kafka/src/airflow/providers/apache/kafka/plugins/listener.py:
##########
@@ -0,0 +1,475 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import atexit
+import json
+import logging
+import os
+import time
+from datetime import datetime, timezone
+from fnmatch import fnmatch
+from functools import lru_cache
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.apache.kafka.version_compat import AIRFLOW_V_3_0_PLUS
+from airflow.providers.common.compat.sdk import AirflowPlugin, conf, hookimpl
+from airflow.utils.net import get_hostname
+
+if TYPE_CHECKING:
+ from confluent_kafka import Producer
+ from sqlalchemy.orm import Session
+
+ from airflow.models import DagRun, TaskInstance
+ from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
+ from airflow.utils.state import TaskInstanceState
+
+log = logging.getLogger(__name__)
+
+CONFIG_SECTION = "kafka_listener"
+SCHEMA_VERSION = 1
+
+
+@lru_cache(maxsize=1)
+def _dag_run_events_enabled() -> bool:
+ return conf.getboolean(CONFIG_SECTION, "dag_run_events_enabled", fallback="False")
+
+
+@lru_cache(maxsize=1)
+def _task_instance_events_enabled() -> bool:
+ return conf.getboolean(CONFIG_SECTION, "task_instance_events_enabled", fallback="False")
+
+
+@lru_cache(maxsize=1)
+def _get_kafka_config_id() -> str:
+ return conf.get(CONFIG_SECTION, "kafka_config_id", fallback="").strip()
+
+
+@lru_cache(maxsize=1)
+def _get_config_dict() -> dict[str, Any] | None:
+ raw_dict_str = conf.get(CONFIG_SECTION, "config_dict", fallback="").strip()
+ if not raw_dict_str:
+ return None
+ config_dict = json.loads(raw_dict_str)
+ if not isinstance(config_dict, dict):
+ raise ValueError(
+ f"The 'config_dict' option of the [{CONFIG_SECTION}] config section "
+ f"must be a JSON object, got: {type(config_dict).__name__}"
+ )
+ return config_dict
+
+
+@lru_cache(maxsize=1)
+def _get_topic() -> str:
+ return conf.get(CONFIG_SECTION, "topic", fallback="airflow.events")
+
+
+@lru_cache(maxsize=1)
+def _get_source() -> str:
+ source_from_conf = conf.get(CONFIG_SECTION, "source", fallback="")
+ # Default to the current hostname, if un-set.
+ return source_from_conf or get_hostname()
+
+
+def _parse_filter_patterns_to_tuple(raw: str) -> tuple[str, ...]:
+ return tuple(p.strip() for p in raw.split(",") if p.strip())
+
+
+@lru_cache(maxsize=1)
+def _get_dag_run_dag_id_allowlist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(conf.get(CONFIG_SECTION, "dag_run_dag_id_allowlist", fallback=""))
+
+
+@lru_cache(maxsize=1)
+def _get_dag_run_dag_id_denylist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(conf.get(CONFIG_SECTION, "dag_run_dag_id_denylist", fallback=""))
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_dag_id_allowlist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_dag_id_allowlist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_dag_id_denylist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_dag_id_denylist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_task_id_allowlist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_task_id_allowlist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_task_id_denylist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_task_id_denylist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_topic_check_timeout() -> int:
+ return conf.getint(CONFIG_SECTION, "topic_check_timeout", fallback="10")
+
+
+@lru_cache(maxsize=1)
+def _get_topic_check_retry_interval() -> int:
+ return conf.getint(CONFIG_SECTION, "topic_check_retry_interval", fallback="60")
+
+
+def _id_is_allowed(id_to_check: str, allowlist: tuple[str, ...], denylist: tuple[str, ...]) -> bool:
+ """Deny takes precedence; empty allowlist means 'allow all'."""
+ if denylist and any(fnmatch(id_to_check, id_pattern) for id_pattern in denylist):
+ return False
+ if allowlist and not any(fnmatch(id_to_check, id_pattern) for id_pattern in allowlist):
+ return False
+ return True
+
+
+def _dag_run_event_allowed(dag_id: str) -> bool:
+ return _id_is_allowed(
+ dag_id,
+ _get_dag_run_dag_id_allowlist(),
+ _get_dag_run_dag_id_denylist(),
+ )
+
+
+def _task_instance_event_allowed(dag_id: str, task_id: str) -> bool:
+ return _id_is_allowed(
+ dag_id,
+ _get_task_instance_dag_id_allowlist(),
+ _get_task_instance_dag_id_denylist(),
+ ) and _id_is_allowed(
+ task_id,
+ _get_task_instance_task_id_allowlist(),
+ _get_task_instance_task_id_denylist(),
+ )
+
+
+# Listener-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.
+_producer: Producer | None = None
+_topic_exists: bool = False
Review Comment:
I updated the code to handle the scenario and I also added a new unit test.
`_on_delivery` was just logging a warning in case there was an error. I extended it to check the error code and if it matches an unknown topic, then reset the flags. If we have reached the `_on_delivery` then the topic was already there and we tried to produce a message. So it doesn't make sense to check again if `_topic_exists` is true.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3403770354
##########
providers/apache/kafka/docs/configurations-ref.rst:
##########
@@ -0,0 +1,91 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _configuration:kafka:
+
+.. include:: /../../../../devel-common/src/sphinx_exts/includes/providers-configurations-ref.rst
+.. include:: /../../../../devel-common/src/sphinx_exts/includes/sections-and-options.rst
+
+
+Highlighted configurations
+===========================
+
+The ``[kafka_listener]`` section configures the ``KafkaListenerPlugin``,
+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`` and the plugin
+is a no-op if ``bootstrap_servers`` isn't also set.
+
+.. _configuration_kafka_listener_activation:kafka:
+
+Activating the listener
+-----------------------
+
+To enable event publishing you need to
+
+ * enable at least one event-type flag
+ * configure a broker address
+ * have a pre-existing kafka topic
+
+.. code-block:: ini
+
+ [kafka_listener]
+ dag_run_events_enabled = True
+ task_instance_events_enabled = True
+ bootstrap_servers = broker1:9092,broker2:9092
Review Comment:
I removed `bootstrap_servers` and added `kafka_config_id` instead. Also, added a new config dict key.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3566041501
##########
providers/apache/kafka/src/airflow/providers/apache/kafka/hooks/base.py:
##########
@@ -85,10 +109,38 @@ def get_ui_field_behaviour(cls) -> dict[str, Any]:
def _get_client(self, config) -> Any:
return AdminClient(config)
+ def _get_config(self) -> dict[str, Any]:
+ """Resolve the client config: the connection extras, with ``config_dict`` keys taking precedence."""
+ config_dict = self.config_dict or {}
+ try:
+ config = dict(self.get_connection(self.kafka_config_id).extra_dejson)
+ except Exception:
+ # The connection lookup can fail for different reasons depending on the
+ # component the hook runs in (missing connection, no DB/API access). When
+ # ``config_dict`` carries the brokers the hook can run without a connection;
+ # otherwise re-raise.
+ if not config_dict.get("bootstrap.servers"):
+ raise
+ self.log.debug(
+ "Kafka connection %r is not available; building the client from 'config_dict' only.",
+ self.kafka_config_id,
+ )
+ config = {}
+ config.update(config_dict)
+ self._resolve_callbacks(config)
Review Comment:
From my perspective, we could still resolve the importable callback from the connection extra instead of having a new `config_dict` for all the Kafka Hooks.
WDYT?
##########
providers/apache/kafka/src/airflow/providers/apache/kafka/plugins/listener.py:
##########
@@ -0,0 +1,475 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import atexit
+import json
+import logging
+import os
+import time
+from datetime import datetime, timezone
+from fnmatch import fnmatch
+from functools import lru_cache
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.apache.kafka.version_compat import AIRFLOW_V_3_0_PLUS
+from airflow.providers.common.compat.sdk import AirflowPlugin, conf, hookimpl
+from airflow.utils.net import get_hostname
+
+if TYPE_CHECKING:
+ from confluent_kafka import Producer
+ from sqlalchemy.orm import Session
+
+ from airflow.models import DagRun, TaskInstance
+ from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
+ from airflow.utils.state import TaskInstanceState
+
+log = logging.getLogger(__name__)
+
+CONFIG_SECTION = "kafka_listener"
+SCHEMA_VERSION = 1
+
+
+@lru_cache(maxsize=1)
+def _dag_run_events_enabled() -> bool:
+ return conf.getboolean(CONFIG_SECTION, "dag_run_events_enabled", fallback="False")
+
+
+@lru_cache(maxsize=1)
+def _task_instance_events_enabled() -> bool:
+ return conf.getboolean(CONFIG_SECTION, "task_instance_events_enabled", fallback="False")
+
+
+@lru_cache(maxsize=1)
+def _get_kafka_config_id() -> str:
+ return conf.get(CONFIG_SECTION, "kafka_config_id", fallback="").strip()
+
+
+@lru_cache(maxsize=1)
+def _get_config_dict() -> dict[str, Any] | None:
+ raw_dict_str = conf.get(CONFIG_SECTION, "config_dict", fallback="").strip()
+ if not raw_dict_str:
+ return None
+ config_dict = json.loads(raw_dict_str)
+ if not isinstance(config_dict, dict):
+ raise ValueError(
+ f"The 'config_dict' option of the [{CONFIG_SECTION}] config section "
+ f"must be a JSON object, got: {type(config_dict).__name__}"
+ )
+ return config_dict
+
+
+@lru_cache(maxsize=1)
+def _get_topic() -> str:
+ return conf.get(CONFIG_SECTION, "topic", fallback="airflow.events")
+
+
+@lru_cache(maxsize=1)
+def _get_source() -> str:
+ source_from_conf = conf.get(CONFIG_SECTION, "source", fallback="")
+ # Default to the current hostname, if un-set.
+ return source_from_conf or get_hostname()
+
+
+def _parse_filter_patterns_to_tuple(raw: str) -> tuple[str, ...]:
+ return tuple(p.strip() for p in raw.split(",") if p.strip())
+
+
+@lru_cache(maxsize=1)
+def _get_dag_run_dag_id_allowlist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(conf.get(CONFIG_SECTION, "dag_run_dag_id_allowlist", fallback=""))
+
+
+@lru_cache(maxsize=1)
+def _get_dag_run_dag_id_denylist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(conf.get(CONFIG_SECTION, "dag_run_dag_id_denylist", fallback=""))
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_dag_id_allowlist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_dag_id_allowlist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_dag_id_denylist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_dag_id_denylist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_task_id_allowlist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_task_id_allowlist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_task_id_denylist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_task_id_denylist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_topic_check_timeout() -> int:
+ return conf.getint(CONFIG_SECTION, "topic_check_timeout", fallback="10")
+
+
+@lru_cache(maxsize=1)
+def _get_topic_check_retry_interval() -> int:
+ return conf.getint(CONFIG_SECTION, "topic_check_retry_interval", fallback="60")
+
+
+def _id_is_allowed(id_to_check: str, allowlist: tuple[str, ...], denylist: tuple[str, ...]) -> bool:
+ """Deny takes precedence; empty allowlist means 'allow all'."""
+ if denylist and any(fnmatch(id_to_check, id_pattern) for id_pattern in denylist):
+ return False
+ if allowlist and not any(fnmatch(id_to_check, id_pattern) for id_pattern in allowlist):
+ return False
+ return True
+
+
+def _dag_run_event_allowed(dag_id: str) -> bool:
+ return _id_is_allowed(
+ dag_id,
+ _get_dag_run_dag_id_allowlist(),
+ _get_dag_run_dag_id_denylist(),
+ )
+
+
+def _task_instance_event_allowed(dag_id: str, task_id: str) -> bool:
+ return _id_is_allowed(
+ dag_id,
+ _get_task_instance_dag_id_allowlist(),
+ _get_task_instance_dag_id_denylist(),
+ ) and _id_is_allowed(
+ task_id,
+ _get_task_instance_task_id_allowlist(),
+ _get_task_instance_task_id_denylist(),
+ )
+
+
+# Listener-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.
+_producer: Producer | None = None
+_topic_exists: bool = False
Review Comment:
Do we need to reset the `_check_topic_exists` flag if we really encounter the topic not found (e.g. someone accidental delete the topic)?
Perhaps as follow-up, non-blocking atm, since user should be able to tell from the error loggings.
##########
providers/apache/kafka/src/airflow/providers/apache/kafka/hooks/base.py:
##########
@@ -58,18 +64,36 @@ class KafkaBaseHook(BaseHook):
"""
A base hook for interacting with Apache Kafka.
+ The Airflow connection's extra JSON is merged with the ``config_dict``.
+ Each ``config_dict`` key overrides the same key from the connection.
+ When the connection is not available and ``config_dict`` contains the key
+ ``bootstrap.servers``, the client is built from ``config_dict`` alone
+ without requiring a connection.
+
:param kafka_config_id: The connection object to use, defaults to "kafka_default"
+ :param config_dict: Optional confluent-kafka client configuration
+ (e.g. ``{"bootstrap.servers": "localhost:9092"}``). Each key overrides the same
+ key from the connection's config. Callback options (``error_cb``, ``throttle_cb``,
+ ``stats_cb``, ``log_cb``, ``oauth_cb``, ``on_commit``) may be callables or
+ dotted-path strings, which are resolved before the client is built.
"""
conn_name_attr = "kafka_config_id"
default_conn_name = "kafka_default"
conn_type = "kafka"
hook_name = "Apache Kafka"
- def __init__(self, kafka_config_id=default_conn_name, *args, **kwargs):
+ def __init__(
+ self,
+ kafka_config_id=default_conn_name,
+ config_dict: dict[str, Any] | None = None,
Review Comment:
I'm still a bit concern about exposing the `config_dict` to bypass the connection.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3567064800
##########
providers/apache/kafka/src/airflow/providers/apache/kafka/hooks/base.py:
##########
@@ -85,10 +109,38 @@ def get_ui_field_behaviour(cls) -> dict[str, Any]:
def _get_client(self, config) -> Any:
return AdminClient(config)
+ def _get_config(self) -> dict[str, Any]:
+ """Resolve the client config: the connection extras, with ``config_dict`` keys taking precedence."""
+ config_dict = self.config_dict or {}
+ try:
+ config = dict(self.get_connection(self.kafka_config_id).extra_dejson)
+ except Exception:
+ # The connection lookup can fail for different reasons depending on the
+ # component the hook runs in (missing connection, no DB/API access). When
+ # ``config_dict`` carries the brokers the hook can run without a connection;
+ # otherwise re-raise.
+ if not config_dict.get("bootstrap.servers"):
+ raise
+ self.log.debug(
+ "Kafka connection %r is not available; building the client from 'config_dict' only.",
+ self.kafka_config_id,
+ )
+ config = {}
+ config.update(config_dict)
+ self._resolve_callbacks(config)
Review Comment:
The patch was resolving the callbacks from both the connection and the `config_dict` param. I just removed the `config_dict` so now it just resolves them from the connection. Diffs are smaller and cleaner.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3367598608
##########
providers/apache/kafka/docs/configurations-ref.rst:
##########
@@ -0,0 +1,91 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _configuration:kafka:
+
+.. include:: /../../../../devel-common/src/sphinx_exts/includes/providers-configurations-ref.rst
+.. include:: /../../../../devel-common/src/sphinx_exts/includes/sections-and-options.rst
+
+
+Highlighted configurations
+===========================
+
+The ``[kafka_listener]`` section configures the ``KafkaListenerPlugin``,
+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`` and the plugin
+is a no-op if ``bootstrap_servers`` isn't also set.
+
+.. _configuration_kafka_listener_activation:kafka:
+
+Activating the listener
+-----------------------
+
+To enable event publishing you need to
+
+ * enable at least one event-type flag
+ * configure a broker address
+ * have a pre-existing kafka topic
+
+.. code-block:: ini
+
+ [kafka_listener]
+ dag_run_events_enabled = True
+ task_instance_events_enabled = True
+ bootstrap_servers = broker1:9092,broker2:9092
Review Comment:
Should use specify `kafka_config_id` instead of `bootstrap_servers`? The `bootstrap_servers` seems ambiguous for what Airflow Kafka Connection.
##########
providers/apache/kafka/src/airflow/providers/apache/kafka/plugins/listener.py:
##########
@@ -0,0 +1,440 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import atexit
+import json
+import logging
+import os
+import time
+from datetime import datetime, timezone
+from fnmatch import fnmatch
+from functools import lru_cache
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.apache.kafka.version_compat import AIRFLOW_V_3_0_PLUS
+from airflow.providers.common.compat.sdk import AirflowPlugin, conf, hookimpl
+from airflow.utils.net import get_hostname
+
+if TYPE_CHECKING:
+ from confluent_kafka import Producer
+ from sqlalchemy.orm import Session
+
+ from airflow.models import DagRun, TaskInstance
+ from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
+ from airflow.utils.state import TaskInstanceState
+
+log = logging.getLogger(__name__)
+
+CONFIG_SECTION = "kafka_listener"
+SCHEMA_VERSION = 1
+
+
+@lru_cache(maxsize=1)
+def _dag_run_events_enabled() -> bool:
+ return conf.getboolean(CONFIG_SECTION, "dag_run_events_enabled", fallback="False")
+
+
+@lru_cache(maxsize=1)
+def _task_instance_events_enabled() -> bool:
+ return conf.getboolean(CONFIG_SECTION, "task_instance_events_enabled", fallback="False")
+
+
+@lru_cache(maxsize=1)
+def _get_topic() -> str:
+ return conf.get(CONFIG_SECTION, "topic", fallback="airflow.events")
+
+
+@lru_cache(maxsize=1)
+def _get_bootstrap_servers() -> str:
+ return conf.get(CONFIG_SECTION, "bootstrap_servers", fallback="").strip()
+
+
+@lru_cache(maxsize=1)
+def _get_source() -> str:
+ source_from_conf = conf.get(CONFIG_SECTION, "source", fallback="")
+ # Default to the current hostname, if un-set.
+ return source_from_conf or get_hostname()
+
+
+def _parse_filter_patterns_to_tuple(raw: str) -> tuple[str, ...]:
+ return tuple(p.strip() for p in raw.split(",") if p.strip())
+
+
+@lru_cache(maxsize=1)
+def _get_dag_run_dag_id_allowlist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(conf.get(CONFIG_SECTION, "dag_run_dag_id_allowlist", fallback=""))
+
+
+@lru_cache(maxsize=1)
+def _get_dag_run_dag_id_denylist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(conf.get(CONFIG_SECTION, "dag_run_dag_id_denylist", fallback=""))
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_dag_id_allowlist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_dag_id_allowlist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_dag_id_denylist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_dag_id_denylist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_task_id_allowlist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_task_id_allowlist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_task_id_denylist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_task_id_denylist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_topic_check_timeout() -> int:
+ return conf.getint(CONFIG_SECTION, "topic_check_timeout", fallback="10")
+
+
+@lru_cache(maxsize=1)
+def _get_topic_check_retry_interval() -> int:
+ return conf.getint(CONFIG_SECTION, "topic_check_retry_interval", fallback="60")
+
+
+def _id_is_allowed(id_to_check: str, allowlist: tuple[str, ...], denylist: tuple[str, ...]) -> bool:
+ """Deny takes precedence; empty allowlist means 'allow all'."""
+ if denylist and any(fnmatch(id_to_check, id_pattern) for id_pattern in denylist):
+ return False
+ if allowlist and not any(fnmatch(id_to_check, id_pattern) for id_pattern in allowlist):
+ return False
+ return True
+
+
+def _dag_run_event_allowed(dag_id: str) -> bool:
+ return _id_is_allowed(
+ dag_id,
+ _get_dag_run_dag_id_allowlist(),
+ _get_dag_run_dag_id_denylist(),
+ )
+
+
+def _task_instance_event_allowed(dag_id: str, task_id: str) -> bool:
+ return _id_is_allowed(
+ dag_id,
+ _get_task_instance_dag_id_allowlist(),
+ _get_task_instance_dag_id_denylist(),
+ ) and _id_is_allowed(
+ task_id,
+ _get_task_instance_task_id_allowlist(),
+ _get_task_instance_task_id_denylist(),
+ )
+
+
+# confluent_kafka.Producer is not fork-safe — its background threads and
+# broker sockets do not survive ``os.fork``. Keep at most one Producer per
+# process and reset the cached state in the child so the next call re-inits.
+# If the initialization is successful, then the value is cached,
+# otherwise init is retried on an interval.
+_producer: Producer | None = None
+_producer_retry_after: float = 0.0
+
+
+def _reset_producer_after_fork() -> None:
+ """Drop any inherited producer in the child; the child re-inits on next call."""
+ global _producer, _producer_retry_after
+ _producer = None
+ _producer_retry_after = 0.0
+
+
+os.register_at_fork(after_in_child=_reset_producer_after_fork)
+
+
+def _get_producer() -> Producer | None:
+ global _producer, _producer_retry_after
+ # If there is a cached producer, return it, don't re-initialize.
+ if _producer is not None or time.monotonic() < _producer_retry_after:
+ return _producer
+
+ result: Producer | None = None
+ try:
+ # The plugin is loaded by every Airflow process. The Producer is a heavy dependency, and
+ # by lazily importing here, we avoid the unneeded startup cost.
+ from confluent_kafka import Producer
+
+ # No need to check if bootstrap_servers are set, because it's done when registering the listeners.
+ # If not set, then there won't be any listeners set and this part of the code will never execute.
+ bootstrap_servers = _get_bootstrap_servers()
+ producer = Producer({"bootstrap.servers": bootstrap_servers})
Review Comment:
Would it be better to leverage `KafkaProducerHook` instead of constructing ourself (which duplicates exact what hook feature provide).
https://github.com/apache/airflow/blob/18caee1497217ef2a0dd0b42bd8a05cb690459aa/providers/apache/kafka/src/airflow/providers/apache/kafka/hooks/produce.py#L24-L42
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3364062269
##########
airflow-core/tests/integration/otel/test_otel.py:
##########
Review Comment:
I refactored these setup functions
```
serialize_and_get_dags,
start_scheduler,
terminate_process,
unpause_trigger_dag_and_get_run_id,
wait_for_dag_run,
```
and moved them under `tests_common` so that they can be reused. They are very useful in cases where we want an integration test that runs an actual dag with an actual scheduler.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3364037914
##########
airflow-core/tests/integration/otel/test_otel.py:
##########
Review Comment:
Are these changes in airflow-core added by accident or why do pytests need change for the provider extension?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 merged PR #68082:
URL: https://github.com/apache/airflow/pull/68082
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3364149189
##########
airflow-core/tests/integration/otel/test_otel.py:
##########
Review Comment:
Tests aren't included in releases, right? And the rest of the changes are related to the kafka provider which will be released separately with the providers.
There aren't any changes to core that would require this PR to be backported to the v3-2-test branch.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3364100457
##########
airflow-core/tests/integration/otel/test_otel.py:
##########
Review Comment:
Okay, but then actually we would not need to backport these to v3-2-test branch, could stay there as they are?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3403770354
##########
providers/apache/kafka/docs/configurations-ref.rst:
##########
@@ -0,0 +1,91 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _configuration:kafka:
+
+.. include:: /../../../../devel-common/src/sphinx_exts/includes/providers-configurations-ref.rst
+.. include:: /../../../../devel-common/src/sphinx_exts/includes/sections-and-options.rst
+
+
+Highlighted configurations
+===========================
+
+The ``[kafka_listener]`` section configures the ``KafkaListenerPlugin``,
+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`` and the plugin
+is a no-op if ``bootstrap_servers`` isn't also set.
+
+.. _configuration_kafka_listener_activation:kafka:
+
+Activating the listener
+-----------------------
+
+To enable event publishing you need to
+
+ * enable at least one event-type flag
+ * configure a broker address
+ * have a pre-existing kafka topic
+
+.. code-block:: ini
+
+ [kafka_listener]
+ dag_run_events_enabled = True
+ task_instance_events_enabled = True
+ bootstrap_servers = broker1:9092,broker2:9092
Review Comment:
I removed `bootstrap_servers` and added `kafka_config_id` instead.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on PR #68082:
URL: https://github.com/apache/airflow/pull/68082#issuecomment-4635976090
> Shouldn't this be an Asset Watcher?
@vikramkoka No but it could be used together with an Asset Watcher. An Asset Watcher is essentially on the consumer side of events while this PR is on the producer side.
The new listener produces a message to a Kafka topic for every event state change. We can consume the messages from the topic either using the existing consumer hook from the Kafka provider or using an Asset Watcher.
For example, team A with Airflow installation A triggers dag1 and has enabled the listener which publishes events for every state change to a Kafka topic
1. dag_run.dag1.running
2. task1.running
3. task1.success
4. task2.running
5. task2.success
6. dag_run.dag1.success
Team B with Airflow installation B, is monitoring the same kafka topic and has a deferred task that waits to run after it consumes the message `task1.success`. The message consumption can happen using an Asset Watcher.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on PR #68082:
URL: https://github.com/apache/airflow/pull/68082#issuecomment-4956838030
The CI is green. It should be good to get merged.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on PR #68082:
URL: https://github.com/apache/airflow/pull/68082#issuecomment-4994857602
@vincbeck Thank you for the comment! I've started working on renaming the listener to `Event Producer` as a suggestion from @shahar1. `Event Publisher` is also accurate but producer is the term generally used by Kafka, and so it made sense to me. If you don't mind, I'll move forward with that.
OpenLineage and Informatica also have a similar listener plugin
https://github.com/apache/airflow/blob/main/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
https://github.com/apache/airflow/blob/main/providers/informatica/src/airflow/providers/informatica/plugins/listener.py
Also see
https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/listeners/listener.py
https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/listeners/spec/dagrun.py
https://github.com/apache/airflow/blob/main/shared/listeners/src/airflow_shared/listeners/spec/taskinstance.py
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #68082:
URL: https://github.com/apache/airflow/pull/68082#issuecomment-4995100132
Oh I see, interesting. It is kind of weird to me to expect deployment manager to have to install a plugin to set that up but that's another discussion we can leave for another time :)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3373053448
##########
providers/apache/kafka/src/airflow/providers/apache/kafka/plugins/listener.py:
##########
@@ -0,0 +1,440 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import atexit
+import json
+import logging
+import os
+import time
+from datetime import datetime, timezone
+from fnmatch import fnmatch
+from functools import lru_cache
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.apache.kafka.version_compat import AIRFLOW_V_3_0_PLUS
+from airflow.providers.common.compat.sdk import AirflowPlugin, conf, hookimpl
+from airflow.utils.net import get_hostname
+
+if TYPE_CHECKING:
+ from confluent_kafka import Producer
+ from sqlalchemy.orm import Session
+
+ from airflow.models import DagRun, TaskInstance
+ from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
+ from airflow.utils.state import TaskInstanceState
+
+log = logging.getLogger(__name__)
+
+CONFIG_SECTION = "kafka_listener"
+SCHEMA_VERSION = 1
+
+
+@lru_cache(maxsize=1)
+def _dag_run_events_enabled() -> bool:
+ return conf.getboolean(CONFIG_SECTION, "dag_run_events_enabled", fallback="False")
+
+
+@lru_cache(maxsize=1)
+def _task_instance_events_enabled() -> bool:
+ return conf.getboolean(CONFIG_SECTION, "task_instance_events_enabled", fallback="False")
+
+
+@lru_cache(maxsize=1)
+def _get_topic() -> str:
+ return conf.get(CONFIG_SECTION, "topic", fallback="airflow.events")
+
+
+@lru_cache(maxsize=1)
+def _get_bootstrap_servers() -> str:
+ return conf.get(CONFIG_SECTION, "bootstrap_servers", fallback="").strip()
+
+
+@lru_cache(maxsize=1)
+def _get_source() -> str:
+ source_from_conf = conf.get(CONFIG_SECTION, "source", fallback="")
+ # Default to the current hostname, if un-set.
+ return source_from_conf or get_hostname()
+
+
+def _parse_filter_patterns_to_tuple(raw: str) -> tuple[str, ...]:
+ return tuple(p.strip() for p in raw.split(",") if p.strip())
+
+
+@lru_cache(maxsize=1)
+def _get_dag_run_dag_id_allowlist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(conf.get(CONFIG_SECTION, "dag_run_dag_id_allowlist", fallback=""))
+
+
+@lru_cache(maxsize=1)
+def _get_dag_run_dag_id_denylist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(conf.get(CONFIG_SECTION, "dag_run_dag_id_denylist", fallback=""))
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_dag_id_allowlist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_dag_id_allowlist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_dag_id_denylist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_dag_id_denylist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_task_id_allowlist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_task_id_allowlist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_task_instance_task_id_denylist() -> tuple[str, ...]:
+ return _parse_filter_patterns_to_tuple(
+ conf.get(CONFIG_SECTION, "task_instance_task_id_denylist", fallback="")
+ )
+
+
+@lru_cache(maxsize=1)
+def _get_topic_check_timeout() -> int:
+ return conf.getint(CONFIG_SECTION, "topic_check_timeout", fallback="10")
+
+
+@lru_cache(maxsize=1)
+def _get_topic_check_retry_interval() -> int:
+ return conf.getint(CONFIG_SECTION, "topic_check_retry_interval", fallback="60")
+
+
+def _id_is_allowed(id_to_check: str, allowlist: tuple[str, ...], denylist: tuple[str, ...]) -> bool:
+ """Deny takes precedence; empty allowlist means 'allow all'."""
+ if denylist and any(fnmatch(id_to_check, id_pattern) for id_pattern in denylist):
+ return False
+ if allowlist and not any(fnmatch(id_to_check, id_pattern) for id_pattern in allowlist):
+ return False
+ return True
+
+
+def _dag_run_event_allowed(dag_id: str) -> bool:
+ return _id_is_allowed(
+ dag_id,
+ _get_dag_run_dag_id_allowlist(),
+ _get_dag_run_dag_id_denylist(),
+ )
+
+
+def _task_instance_event_allowed(dag_id: str, task_id: str) -> bool:
+ return _id_is_allowed(
+ dag_id,
+ _get_task_instance_dag_id_allowlist(),
+ _get_task_instance_dag_id_denylist(),
+ ) and _id_is_allowed(
+ task_id,
+ _get_task_instance_task_id_allowlist(),
+ _get_task_instance_task_id_denylist(),
+ )
+
+
+# confluent_kafka.Producer is not fork-safe — its background threads and
+# broker sockets do not survive ``os.fork``. Keep at most one Producer per
+# process and reset the cached state in the child so the next call re-inits.
+# If the initialization is successful, then the value is cached,
+# otherwise init is retried on an interval.
+_producer: Producer | None = None
+_producer_retry_after: float = 0.0
+
+
+def _reset_producer_after_fork() -> None:
+ """Drop any inherited producer in the child; the child re-inits on next call."""
+ global _producer, _producer_retry_after
+ _producer = None
+ _producer_retry_after = 0.0
+
+
+os.register_at_fork(after_in_child=_reset_producer_after_fork)
+
+
+def _get_producer() -> Producer | None:
+ global _producer, _producer_retry_after
+ # If there is a cached producer, return it, don't re-initialize.
+ if _producer is not None or time.monotonic() < _producer_retry_after:
+ return _producer
+
+ result: Producer | None = None
+ try:
+ # The plugin is loaded by every Airflow process. The Producer is a heavy dependency, and
+ # by lazily importing here, we avoid the unneeded startup cost.
+ from confluent_kafka import Producer
+
+ # No need to check if bootstrap_servers are set, because it's done when registering the listeners.
+ # If not set, then there won't be any listeners set and this part of the code will never execute.
+ bootstrap_servers = _get_bootstrap_servers()
+ producer = Producer({"bootstrap.servers": bootstrap_servers})
Review Comment:
Currently, users create an airflow connection and pass `bootstrap_servers` and any other properties to the connection. Then they initialize the Producer or Consumer hook in a dag and pass the connection name -- `kafka_config_id` -- as a param to the constructor. The connection carries all the necessary config.
In my initial implementation, I reused the existing hook and added a `kafka_config_id` property in the configuration. But, it felt tiring to always have to create a connection in advance. That's why I didn't use the hook and instead added a `bootstrap_servers` key in the config.
Then I noticed that the initialized producer isn't fork safe and we need to handle that as well. So, I reset the producer at process fork, and register an atexit hook that flushes the producer at process exit.
Because of your comments I'm revisiting the approach. I'm thinking about moving some of the config options and the forking changes under the hook itself.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on PR #68082:
URL: https://github.com/apache/airflow/pull/68082#issuecomment-4957407791
Thank you @shahar1 @jason810496 for the reviews and help!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3563601793
##########
devel-common/src/tests_common/test_utils/integration_setup.py:
##########
@@ -0,0 +1,203 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+import logging
+import os
+import subprocess
+import time
+
+from sqlalchemy import func, select
+
+from airflow.dag_processing.bundles.manager import DagBundlesManager
+from airflow.dag_processing.dagbag import DagBag
+from airflow.models import DagRun
+from airflow.models.serialized_dag import SerializedDagModel
+from airflow.serialization.definitions.dag import SerializedDAG
+from airflow.utils.session import create_session
+from airflow.utils.state import State
+
+from tests_common.test_utils.dag import create_scheduler_dag
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS, AIRFLOW_V_3_1_PLUS
+
+try:
+ from airflow.sdk import timezone
+except ImportError:
+ from airflow.utils import timezone # type: ignore[no-redef,attr-defined]
+
+log = logging.getLogger(__name__)
+
+
+def unpause_trigger_dag_and_get_run_id(dag_id: str, conf: dict | None = None) -> str:
+ unpause_command = ["airflow", "dags", "unpause", dag_id]
+
+ # Unpause the dag using the cli.
+ subprocess.run(unpause_command, check=True, env=os.environ.copy())
+
+ execution_date = timezone.utcnow()
+ run_id = f"manual__{execution_date.isoformat()}"
+
+ trigger_command = [
+ "airflow",
+ "dags",
+ "trigger",
+ dag_id,
+ "--run-id",
+ run_id,
+ "--logical-date",
+ execution_date.isoformat(),
+ ]
+
+ if conf:
+ import json
Review Comment:
Please extract the import outside of the function
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on code in PR #68082:
URL: https://github.com/apache/airflow/pull/68082#discussion_r3563673989
##########
devel-common/src/tests_common/test_utils/integration_setup.py:
##########
@@ -0,0 +1,203 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+import logging
+import os
+import subprocess
+import time
+
+from sqlalchemy import func, select
+
+from airflow.dag_processing.bundles.manager import DagBundlesManager
+from airflow.dag_processing.dagbag import DagBag
+from airflow.models import DagRun
+from airflow.models.serialized_dag import SerializedDagModel
+from airflow.serialization.definitions.dag import SerializedDAG
+from airflow.utils.session import create_session
+from airflow.utils.state import State
+
+from tests_common.test_utils.dag import create_scheduler_dag
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS, AIRFLOW_V_3_1_PLUS
+
+try:
+ from airflow.sdk import timezone
+except ImportError:
+ from airflow.utils import timezone # type: ignore[no-redef,attr-defined]
+
+log = logging.getLogger(__name__)
+
+
+def unpause_trigger_dag_and_get_run_id(dag_id: str, conf: dict | None = None) -> str:
+ unpause_command = ["airflow", "dags", "unpause", dag_id]
+
+ # Unpause the dag using the cli.
+ subprocess.run(unpause_command, check=True, env=os.environ.copy())
+
+ execution_date = timezone.utcnow()
+ run_id = f"manual__{execution_date.isoformat()}"
+
+ trigger_command = [
+ "airflow",
+ "dags",
+ "trigger",
+ dag_id,
+ "--run-id",
+ run_id,
+ "--logical-date",
+ execution_date.isoformat(),
+ ]
+
+ if conf:
+ import json
Review Comment:
There is already an import on top. Good catch! Removed.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add a DagRun and TaskInstance event listener under the Kafka provider [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on PR #68082:
URL: https://github.com/apache/airflow/pull/68082#issuecomment-4691808034
Hey @jason810496, I've updated the PR. It took me more time than originally planned but I need to try a few different approaches and test things thoroughly. I think that this last revision is better.
I added two new options in the listener config
1. set the kafka connection name
2. set a json dictionary with the kafka config
The producer, consumer and admin client hooks, all require an airflow connection to be initialized. The airflow connection carries an extra param JSON that holds all the necessary configuration. This configuration can also carry security info.
Initially I tried to workaround it by defining `bootstrap.servers` in the listener config, so that users can initialize a producer without creating a connection.
Confluent kafka producer accepts the config as a dictionary. But the airflow connection json is a string, which has limitation like not being able to pass standard callables like `on_delivery` or `oauth_cb`, etc. without having to hardcode them in the hook or the caller.
I added an optional parameter to the base class, so that hook callers can pass the config dictionary directly to the hook without using a connection. The param is optional for back-compat reasons. Also, I added a new method that interprets the config callables whether they are part of the connection or the dictionary. It accepts dot-path strings pointing to locally installed packages.
If a user has a connection but has also set a config key in the dict param, the dict value overrides the value from the connection.
The changes are more extensive than originally planned but I think that they improve the way the hooks are configured greatly.
Please take a look and let me know how it looks. I also updated all the tests.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org