This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 74c76cbf02f Document `apply_function` dot notation requirement for
Kafka message queues (#68414)
74c76cbf02f is described below
commit 74c76cbf02f3eaa5cf448d23946969f35531183b
Author: Jorge Rocamora <[email protected]>
AuthorDate: Wed Jul 22 19:04:33 2026 +0200
Document `apply_function` dot notation requirement for Kafka message queues
(#68414)
---
airflow-core/docs/core-concepts/message-queues.rst | 5 +++
.../apache/kafka/docs/message-queues/index.rst | 44 ++++++++++++++++++++++
providers/apache/kafka/docs/sensors.rst | 9 +++++
3 files changed, 58 insertions(+)
diff --git a/airflow-core/docs/core-concepts/message-queues.rst
b/airflow-core/docs/core-concepts/message-queues.rst
index 573189a24f8..5c1daefa70a 100644
--- a/airflow-core/docs/core-concepts/message-queues.rst
+++ b/airflow-core/docs/core-concepts/message-queues.rst
@@ -38,4 +38,9 @@ Airflow constantly monitors the state of an external resource
and updates the as
resource reaches a given state (if it does reach it). To achieve this, we
leverage Airflow Triggers.
Triggers are small, asynchronous pieces of Python code whose job is to poll an
external resource state.
+Each queue provider accepts provider-specific keyword arguments that are
forwarded to the underlying
+trigger. Refer to the provider's message queue documentation for the supported
arguments. Because
+triggers run in the Triggerer, any user code referenced by these arguments
(for example a function
+passed as a string in Python dot notation) must be importable there.
+
The list of supported message queues is available in
:doc:`apache-airflow-providers:core-extensions/message-queues`.
diff --git a/providers/apache/kafka/docs/message-queues/index.rst
b/providers/apache/kafka/docs/message-queues/index.rst
index fd6c0924969..13482f4c484 100644
--- a/providers/apache/kafka/docs/message-queues/index.rst
+++ b/providers/apache/kafka/docs/message-queues/index.rst
@@ -74,3 +74,47 @@ asset.
For how to use the trigger, refer to the documentation of the
:ref:`Messaging Trigger <howto/trigger:MessageQueueTrigger>`
+
+.. _howto/triggers:KafkaApplyFunction:
+
+The ``apply_function``
+----------------------
+
+The ``apply_function`` is applied to every message polled from the Kafka
topic(s). If it returns a truthy
+value, that value is used as the payload of the ``TriggerEvent``. Otherwise,
the trigger keeps polling.
+It is required when using the Kafka queue provider, while the Kafka sensors
also accept ``None``, in which
+case the raw message value (decoded as UTF-8) is used as the event payload.
+
+The function must be passed as a string in Python dot notation, for example
+``"my_package.my_module.my_function"``, not as the function object, because
trigger arguments are
+serialized to the metadata database. The function is imported in the triggerer
at runtime, so the module
+must be importable there, and changes to the function require a triggerer
restart.
+
+Additional arguments can be passed using ``apply_function_args`` and
``apply_function_kwargs``. The message
+is always passed as the last positional argument:
+
+.. code-block:: python
+
+ # my_package/my_module.py
+ import json
+
+ from confluent_kafka import Message
+
+
+ def my_function(prefix: str, message: Message, threshold: int = 0) -> str
| None:
+ val = json.loads(message.value())
+ if val["amount"] > threshold:
+ return f"{prefix}{val}"
+
+.. code-block:: python
+
+ # In your Dag file
+ from airflow.providers.common.messaging.triggers.msg_queue import
MessageQueueTrigger
+
+ trigger = MessageQueueTrigger(
+ scheme="kafka",
+ topics=["my_topic"],
+ apply_function="my_package.my_module.my_function",
+ apply_function_args=["received:"],
+ apply_function_kwargs={"threshold": 100},
+ )
diff --git a/providers/apache/kafka/docs/sensors.rst
b/providers/apache/kafka/docs/sensors.rst
index 7dbf0bc4fa6..c45ccddde25 100644
--- a/providers/apache/kafka/docs/sensors.rst
+++ b/providers/apache/kafka/docs/sensors.rst
@@ -29,6 +29,12 @@ A sensor that defers until a specific message is published
to a Kafka topic.
The sensor will create a consumer reading messages from a Kafka topic until a
message fulfilling criteria defined in the
``apply_function`` parameter is found. If the ``apply_function`` returns any
data, a ``TriggerEvent`` is raised and the ``AwaitMessageSensor`` completes
successfully.
+.. note::
+
+ The ``apply_function`` must be provided as a string in Python dot
notation, not as a function
+ object. The function is imported in the triggerer, so the module must be
importable there. See
+ :ref:`the apply_function documentation
<howto/triggers:KafkaApplyFunction>` for details.
+
For parameter definitions take a look at
:class:`~airflow.providers.apache.kafka.sensors.kafka.AwaitMessageSensor`.
Using the sensor
@@ -57,6 +63,9 @@ Similar to the ``AwaitMessageSensor`` above, this sensor will
defer until it con
of its ``apply_function``. Once a positive event is encountered, the
``AwaitMessageTriggerFunctionSensor`` will trigger a callable provided
to ``event_triggered_function``. Afterwards the sensor will be deferred again,
continuing to consume messages.
+The same :ref:`apply_function requirements
<howto/triggers:KafkaApplyFunction>` as for the ``AwaitMessageSensor``
+apply.
+
For parameter definitions take a look at
:class:`~airflow.providers.apache.kafka.sensors.kafka.AwaitMessageTriggerFunctionSensor`.
Using the sensor
