airflow.providers.apache.kafka.queues.kafka¶

Attributes¶

QUEUE_REGEXP

Classes¶

KafkaMessageQueueProvider

Configuration for Apache Kafka integration with common-messaging.

Module Contents¶

airflow.providers.apache.kafka.queues.kafka.QUEUE_REGEXP = '^kafka://'[source]¶

class airflow.providers.apache.kafka.queues.kafka.KafkaMessageQueueProvider[source]¶

Bases: airflow.providers.common.messaging.providers.base_provider.BaseMessageQueueProvider

Configuration for Apache Kafka integration with common-messaging.

[START kafka_message_queue_provider_description]

It uses kafka as scheme for identifying Kafka queues.
For parameter definitions take a look at AwaitMessageTrigger.

from airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger
from airflow.sdk import Asset, AssetWatcher

trigger = MessageQueueTrigger(
    scheme="kafka",
    # Additional Kafka AwaitMessageTrigger parameters as needed
    topics=["my_topic"],
    apply_function="module.apply_function",
    bootstrap_servers="localhost:9092",
)

asset = Asset("kafka_queue_asset", watchers=[AssetWatcher(name="kafka_watcher", trigger=trigger)])

For a complete example, see: tests.system.common.messaging.kafka_message_queue_trigger

[END kafka_message_queue_provider_description]

scheme = 'kafka'[source]¶

queue_matches(queue)[source]¶

Return whether a given queue (string) matches a specific provider’s pattern.

This function must be as specific as possible to avoid collision with other providers. Functions in this provider should NOT overlap with each other in their matching criteria.

Parameters:: queue (str) – The queue identifier

trigger_class()[source]¶

Trigger class to use when queue_matches returns True.

trigger_kwargs(queue, **kwargs)[source]¶

Parameters passed to the instance of trigger_class.

Parameters:: queue (str) – The queue identifier