When an external system finishes uploading data, it can publish a message instead of creating a completion file. MessageQueueTrigger monitors that queue and lets Airflow respond when a matching message arrives, supporting near-real-time workflow activation. section 7.3 section 7.4
Configure the trigger with a queue URI and an apply_function. The function receives each message and returns a Boolean, allowing irrelevant messages to be filtered. Attach the trigger to an Asset through AssetWatcher, then use that asset in the DAG’s schedule. section 7.3
Triggering a DAG from Kafka
from airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger from airflow.sdk import Asset, AssetWatcher trigger = MessageQueueTrigger( queue="kafka://kafka:9092/events", apply_function="custom.kafka_util.apply_function", ) asset = Asset("kafka_queue_asset", watchers=[ AssetWatcher(name="kafka_watcher", trigger=trigger) ]) with DAG(dag_id="12_kafka_trigger", schedule=[asset]): copy >> process >> create_metrics Produce a message with Kafka’s console producer; the DAG may start after the consumer’s configurable fetch interval.
