Apache Airflow Provider for RabbitMQ

Trigger Airflow DAGs reactively from RabbitMQ queues — plus hooks, operators, sensors, and deferrable triggers.

Installation • Connection • Components • Examples • Contributing

Powered by Claude Code

Overview

airflow-provider-rmq is a community provider package for Apache Airflow × RabbitMQ. It supports:

• Reactive DAG triggering — RMQ Watcher Plugin starts DAGs automatically when messages arrive in a queue, with no polling and no worker slots consumed
• Publishing messages to exchanges and queues
• Consuming messages with header-based and callable-based filtering
• Waiting for specific messages with sensors (classic poke and deferrable mode)
• Deferrable sensor in pull mode (periodic polling) and push mode (broker-delivered via basic_consume) — choose based on latency requirements
• Full queue and exchange management (declare, delete, purge, bind, unbind)
• SSL/TLS connections
• Dead Letter Queue (DLQ) setup helpers
• QoS configuration (prefetch)

Requirements

Installation

Install from PyPI

pip install airflow-provider-rmq

Building from source

git clone https://github.com/mkozhin/airflow-provider-rmq.git
cd airflow-provider-rmq
pip install build
python -m build
pip install dist/airflow_provider_rmq-*.whl

Connection Setup

Create a new connection in the Airflow UI (Admin > Connections) with:

SSL/TLS Configuration

Add SSL settings in the Extra field as JSON:

{
  "ssl_enabled": true,
  "ca_certs": "/path/to/ca.pem",
  "certfile": "/path/to/client-cert.pem",
  "keyfile": "/path/to/client-key.pem",
  "cert_reqs": "CERT_REQUIRED"
}

The hook also provides custom form widgets for SSL fields (ssl_enabled, ca_certs, certfile, keyfile) visible in the Airflow connection form.

Set "cert_reqs": "CERT_NONE" to disable certificate verification (not recommended for production).

Connection Timing

Three optional Extra keys control AMQP timing. All of them are plain JSON values next to the SSL settings:

{
  "heartbeat": 30,
  "connect_timeout": 15,
  "rpc_timeout": 30
}

heartbeat is the same key the synchronous hook reads; the two paths differ only in default, because RMQHook lives in short sessions inside a task while the watcher holds a connection for days. Setting "heartbeat": 0 is accepted as a deliberate opt-out and logged as a WARNING: with heartbeats off, a link that dies silently (a broker restart behind a proxy that keeps the client half open) stays undetected.

connect_timeout and rpc_timeout have no counterpart in RMQHook — pika is driven by its own retry logic there. A value that is not a positive number falls back to the default with a WARNING.

Components

RMQHook

Import: from airflow_provider_rmq.hooks.rmq import RMQHook

Core hook for all RabbitMQ interactions. Uses pika BlockingConnection with automatic retry logic (tenacity). The connection is closed automatically when the hook object is garbage-collected, so you do not need to call close() manually. Context manager (with) is also supported.

Constructor Parameters

Key Methods

Usage Example

from airflow_provider_rmq.hooks.rmq import RMQHook

hook = RMQHook(rmq_conn_id="rmq_default")
info = hook.queue_info("my_queue")
print(f"Messages in queue: {info['message_count']}")

hook.basic_publish(
    exchange="",
    routing_key="my_queue",
    body='{"key": "value"}',
)
# Connection is closed automatically when hook goes out of scope

RMQPublishOperator

Import: from airflow_provider_rmq.operators.rmq_publish import RMQPublishOperator

Publishes one or more messages to RabbitMQ. Supports strings, dicts (auto-serialized to JSON), and lists.

Parameters

Template fields: exchange, routing_key, message

Usage Example

# Publish a single dict to a queue
RMQPublishOperator(
    task_id="publish",
    queue_name="my_queue",
    message={"event": "order_created", "id": 42},
    delivery_mode=2,
    headers={"x-source": "airflow"},
)

# Publish a batch of messages to an exchange
RMQPublishOperator(
    task_id="publish_batch",
    exchange="events",
    routing_key="orders.new",
    message=[
        {"id": 1, "item": "widget"},
        {"id": 2, "item": "gadget"},
    ],
)

RMQConsumeOperator

Import: from airflow_provider_rmq.operators.rmq_consume import RMQConsumeOperator

Consumes messages from a RabbitMQ queue. Matching messages are ACKed and returned via XCom. Non-matching messages are NACKed with requeue=True.

Parameters

Template fields: queue_name

Returns: list[dict] — list of matched messages, each with keys: body, headers, routing_key, exchange

Usage Example

# Consume with header filter
RMQConsumeOperator(
    task_id="consume_orders",
    queue_name="orders",
    filter_headers={"x-type": "order"},
    max_messages=50,
    qos={"prefetch_count": 10},
)

# Consume with body-path filter
RMQConsumeOperator(
    task_id="consume_active",
    queue_name="events",
    filter_headers={"body.status": "active"},
)

# Consume with custom callable filter
def large_orders(properties, body: str) -> bool:
    import json
    data = json.loads(body)
    return data.get("amount", 0) > 1000

RMQConsumeOperator(
    task_id="consume_large",
    queue_name="orders",
    filter_callable=large_orders,
)

Processing Messages with TaskFlow API

RMQConsumeOperator returns list[dict] via XCom. Use consume.output in a @task function to access and process each message:

from airflow.decorators import dag, task
from airflow_provider_rmq.operators.rmq_consume import RMQConsumeOperator

@dag(...)
def my_pipeline():
    consume = RMQConsumeOperator(
        task_id="consume",
        queue_name="orders",
        max_messages=50,
    )

    @task
    def process_messages(messages: list[dict]) -> list[dict]:
        results = []
        for msg in messages:
            body = msg["body"]          # message body (str)
            headers = msg["headers"]    # AMQP headers (dict)
            rk = msg["routing_key"]     # routing key
            exchange = msg["exchange"]  # source exchange
            log.info("Message: body=%s, headers=%s", body, headers)

            data = json.loads(body)
            results.append(data)
        return results

    processed = process_messages(consume.output)
    processed >> next_task  # pass results downstream

RMQQueueManagementOperator

Import: from airflow_provider_rmq.operators.rmq_management import RMQQueueManagementOperator

Performs queue and exchange management operations on RabbitMQ.

Parameters

Template fields: queue_name, exchange_name, routing_key, arguments

Supported Actions

Usage Example

# Create a durable queue
RMQQueueManagementOperator(
    task_id="create_queue",
    action="declare_queue",
    queue_name="my_queue",
    durable=True,
)

# Create a topic exchange and bind a queue
RMQQueueManagementOperator(
    task_id="create_exchange",
    action="declare_exchange",
    exchange_name="events",
    exchange_type="topic",
    durable=True,
)

RMQQueueManagementOperator(
    task_id="bind",
    action="bind_queue",
    queue_name="my_queue",
    exchange_name="events",
    routing_key="orders.*",
)

RMQSensor

Import: from airflow_provider_rmq.sensors.rmq import RMQSensor

Waits for a message in a RabbitMQ queue that matches optional filter conditions. Supports classic poke mode and deferrable mode.

Parameters

Template fields: queue_name, message_wait_timeout

Returns: dict | None — matched message with keys: body, headers, routing_key, exchange

Deferrable Mode

When deferrable=True, the sensor defers execution to the Airflow triggerer process using RMQTrigger. This frees the worker slot while waiting for a message, which is more resource-efficient for long waits.

Limitation: filter_callable cannot be used with deferrable=True because Python callables cannot be serialized to the triggerer process. Use filter_headers instead.

Pull vs Push Mode

The mode parameter (only relevant with deferrable=True) controls how the trigger receives messages:

Timeout behaviour: when message_wait_timeout expires, the sensor raises AirflowSkipException — the task is marked SKIPPED (not FAILED) and downstream tasks are skipped. No on_failure_callback is triggered. This makes it safe to use message_wait_timeout for planned stops (e.g., end of business hours) without generating false alerts.

RabbitMQ 4.0+ quorum queue note: non-matching messages are NACKed with requeue=True. Quorum queues enforce a default redelivery limit of 20 — after 20 redeliveries the message is dead-lettered or dropped. Applies to both pull and push modes.

Usage Example

# Classic poke mode with callable filter
RMQSensor(
    task_id="wait_for_order",
    queue_name="orders",
    filter_callable=lambda props, body: "urgent" in body,
    poke_interval=10,
    timeout=300,
    mode="reschedule",
)

# Deferrable pull mode (default)
RMQSensor(
    task_id="wait_for_event",
    queue_name="events",
    filter_headers={"x-type": "payment"},
    deferrable=True,
    timeout=600,
)

# Deferrable push mode — broker delivers instantly, give up after 60 s
RMQSensor(
    task_id="wait_for_event_push",
    queue_name="events",
    filter_headers={"x-type": "payment"},
    deferrable=True,
    mode="push",
    message_wait_timeout=60,
    timeout=120,
)

# Dynamic timeout via XCom — e.g. compute remaining seconds until end of business hours
RMQSensor(
    task_id="wait_for_message",
    queue_name="events",
    deferrable=True,
    mode="push",
    message_wait_timeout="{{ ti.xcom_pull(task_ids='compute_timeout') }}",
)

Processing Sensor Result with TaskFlow API

RMQSensor returns dict | None via XCom. Use sensor.output in a @task function to access the matched message:

from airflow.decorators import dag, task
from airflow_provider_rmq.sensors.rmq import RMQSensor

@dag(...)
def my_pipeline():
    wait = RMQSensor(
        task_id="wait_for_event",
        queue_name="events",
        filter_headers={"x-type": "payment"},
        deferrable=True,
    )

    @task
    def handle_event(message: dict):
        log.info("Received: %s", message)
        return message

    handle_event(wait.output)

RMQTrigger

Import: from airflow_provider_rmq.triggers.rmq import RMQTrigger

Async trigger for deferrable sensor mode. Uses aio_pika for non-blocking AMQP access. Typically not used directly — RMQSensor with deferrable=True creates it automatically.

Parameters

MessageFilter (Utility)

Import: from airflow_provider_rmq.utils.filters import MessageFilter

Evaluates whether a RabbitMQ message matches given filter conditions. Used internally by operators and sensors.

Filter Modes

1. Header filtering (filter_headers): dict of key-value pairs that message headers must match.

   • Regular keys check properties.headers dict
   • Keys starting with body. traverse the JSON-parsed message body (e.g., {"body.data.status": "active"})

2. Callable filtering (filter_callable): fn(properties, body_str) -> bool

Both can be combined (AND logic: both must pass).

RMQ Watcher Plugin

The RMQ Watcher Plugin inverts the usual sensor pattern: instead of a DAG waiting for a message, a RabbitMQ message causes the DAG to start automatically — without polling, without deferred task slots, without worker resources.

How it works

The Scheduler process runs a background asyncio loop (via Airflow Listener API) that subscribes to queues via AMQP basic_consume. When a matching message arrives, trigger_dag() is called directly inside the process. One RobustConnection per conn_id is shared across all subscriptions to that cluster, with cooldown publishing on a second connection of its own (see Connection Resilience).

Every 60 seconds (configurable via Airflow Variable rmq_watcher_reconcile_interval) a reconciliation loop re-scans DAG files for @rmq_trigger decorators (mtime-based — only changed files are re-parsed) and syncs subscriptions to the database.

dag_id in @dag(...) (positional or keyword) must be a string literal, or a simple module-level string constant defined earlier in the file than the decorated function, for the AST scan of dag_file subscriptions to find it. If it isn't, the subscription simply does not appear on the Subscriptions page at all. This is the one case the ⚠ dag not found badge (see Subscription Management) cannot catch: there's no row to highlight, since the subscription never registers in the first place. The only signal is a WARNING in the Scheduler log, and only after a restart or when the file's mtime changes (not on every reconcile cycle — only when the file is actually re-scanned).

Quick Start

Step 1 — annotate your DAG:

from airflow.decorators import dag, task
from airflow_provider_rmq.watcher.decorators import rmq_trigger

@rmq_trigger(queue="orders", conn_id="rmq_default")
@dag(schedule=None)
def orders_dag():
    @task
    def process(**context):
        conf = context["dag_run"].conf
        print(f"Body: {conf['body']}, Headers: {conf['headers']}")
    process()

orders_dag()

@rmq_trigger must be the outermost decorator, directly above @dag(...) (as shown) — putting it below raises a TypeError at import time instead of silently doing nothing.

Step 2 — restart the Scheduler. The plugin activates automatically; no extra configuration is needed.

Step 3 — publish a message to orders — the DAG starts within seconds.

Multi-Queue and Cooldown

Subscribe one DAG to several queues and throttle repeated triggers with cooldown:

from airflow.decorators import dag, task
from airflow_provider_rmq.watcher.decorators import rmq_trigger

@rmq_trigger(
    queues=["orders", "payments"],  # message from any queue starts the DAG
    cooldown=300,                    # 300 s cooldown — DAG runs once per window
    conn_id="rmq_default",
)
@dag(dag_id="my_dag", schedule=None)
def my_dag():
    @task
    def process(**context):
        conf = context["dag_run"].conf
        # conf["source"] == "cooldown" when triggered via cooldown mechanism
        # conf["body"] and conf["headers"] are empty — original data is not
        # preserved through the DLX chain
        print(conf["source"])
    process()

my_dag()

How cooldown works:

• When the first matching message arrives, the plugin publishes a TTL marker to rmq_watcher.pending.{dag_id} (a no-consumer queue with x-max-length=1 and DLX to rmq_watcher.fire).
• After N seconds the marker expires and is routed to rmq_watcher.fire; the fire consumer calls trigger_dag() with an idempotent run_id.
• Additional messages arriving during the cooldown window are ACKed silently — the pending queue rejects the duplicate publish (x-overflow=reject-publish).
• All RMQ infrastructure (rmq_watcher.fire exchange, queue, and per-DAG rmq_watcher.pending.* queues) is created automatically by the plugin on startup.

Limitations:

• All cooldown subscriptions for a DAG share one pending queue and one timer.
• All cooldown DAGs must use the same conn_id / vhost.
• conf["body"] and conf["headers"] in the DAG run conf are empty when triggered via cooldown — original message data is lost in the DLX chain.
• Changing cooldown in a DAG file takes effect on the next reconcile cycle (default 60 s); already-running timers in RMQ are not affected.
• A cooldown DAG Airflow never serializes — removed from the folder, broken by an import error, present on another instance only — leaves its fire event circling rmq_watcher.fire for as long as the process runs, while every subscription row of that DAG keeps saying listening. The fire consumer holds no row of its own, so the scheduler log is the only place it shows: an INFO line per event and a WARNING on every tenth.

RabbitMQ Permissions (cooldown only):

When cooldown > 0 is used, the Airflow RMQ user needs configure/write/read permissions on the rmq_watcher.* resource pattern in addition to permissions on your application queues:

rabbitmqctl set_permissions -p <vhost> <user> "^(rmq_watcher\\..*|your-queue.*)$" "^(rmq_watcher\\..*|your-queue.*)$" "^(rmq_watcher\\..*|your-queue.*)$"

This covers the rmq_watcher.fire exchange, rmq_watcher.fire queue and rmq_watcher.pending.<dag_id> queues that the cooldown mechanism creates automatically.

Exchange-mode triggers

Instead of subscribing to a pre-existing, manually-bound queue (queue=/queues=), a DAG can subscribe directly to a topic exchange. When exchange= is given, the provider owns the RMQ infrastructure end-to-end — no manual queue creation, no external YAML route table:

from airflow.decorators import dag, task
from airflow_provider_rmq.watcher.decorators import rmq_trigger

# Jetstat-shaped routing keys: id × status cross-product
@rmq_trigger(
    exchange="jetstat.airflow",
    routing_key_ids=["670f877702775c2de8325b1f"],
    routing_key_status="succeeded",   # defaults to "*" = any status
)
@dag(dag_id="jetstat_succeeded", schedule=None)
def jetstat_succeeded_dag():
    @task
    def process(**context):
        conf = context["dag_run"].conf
        print(conf["routing_key"])  # "670f877702775c2de8325b1f.succeeded"
    process()

jetstat_succeeded_dag()

# Literal routing keys of any shape (not tied to the id/status form)
@rmq_trigger(exchange="some.other.exchange", routing_keys=["region.eu.alert"])
@dag(dag_id="region_alerts", schedule=None)
def region_alerts_dag():
    ...

Both forms can be combined on the same call — the final routing key set is the union of routing_keys and the routing_key_ids × routing_key_status cross-product.

What the provider provisions automatically on every reconcile cycle:

• The exchange itself (topic, durable, with an alternate-exchange for unroutable messages)
• A dedicated queue rmq_watcher.sub.{dag_id} — one shared queue per DAG, consumed exactly like any queue= subscription
• Bindings between that queue and the exchange, kept in sync with the routing keys currently declared in the decorator (diffed against RabbitMQ's actual binding state via the Management HTTP API — not against anything stored in the Airflow DB)
• Safety nets: unroutable messages land in {exchange}.unrouted (TTL 8h); every routed message is mirrored into {exchange}.log (catch-all # binding, TTL 8h) for downstream logging/auditing

Connection extra — management_url: the broker's Management HTTP API endpoint. Exchange-mode needs it to read current bindings (AMQP 0-9-1 has no "show my bindings" operation), and the liveness watchdog uses it for subscriptions of every type. Add it to the same Airflow Connection used for AMQP:

{
  "management_url": "https://rabbitmq.example.com"
}

The account needs the management tag on top of its AMQP permissions — the Management API answers a tagless user with 401/403 whatever its set_permissions say:

rabbitmqctl set_user_tags <user> management

What the Management API replies is shaped by the rights of the account that asked: a user tagged management is shown only the channels of its own connections, while monitoring and administrator see the whole vhost. On the minimum-privilege account above, the watcher therefore sees its own consumers and no one else's.

The same login/password from the connection are reused for the Management API call. If management_url is not set, bind-diff is skipped on every cycle (logged as ERROR) — the queue is still declared and consumed normally, but bindings never get created/updated.

management_url also selects the liveness probe for every subscription, queue= and exchange= alike. With it, the check reads GET /api/consumers/{vhost}, can name individual consumer tags and reports the Broker consumers number. Without it, the check falls back to a passive queue declare: that vouches for — or condemns — all consumers of the conn_id at once and returns no count, so the Broker consumers column stays — even when the verdict is positive. After two consecutive Management API failures on one conn_id (wrong URL, credentials without the management tag, 404, timeout) the check switches to that same passive declare rather than staying blind. The same two-cycle bound covers a conn_id whose Airflow connection cannot be read at all — renamed, deleted while subscriptions still name it, or a secrets backend that keeps refusing: the check then asks the pooled connection itself, which needs no metadata row.

No stacking — one DAG, one exchange. Multiple @rmq_trigger(exchange=...) decorators on the same DAG raise ValueError — they would all resolve to the same rmq_watcher.sub.{dag_id} queue, and the last one parsed would silently win. The error fires as soon as the DAG object actually exists: immediately at decoration time for a direct DAG instance, or when the @dag(...) factory is called (e.g. jetstat_succeeded_dag() at the end of the file, as in the example above) for TaskFlow style — either way, always during import, before the DAG is registered. Use a single decorator call with the union of routing keys, or subscribe to multiple exchanges across multiple DAGs. To consume from several exchanges on the same DAG, fall back to queue=/queues= with manually created and bound queues.

RabbitMQ permissions (exchange-mode only): in addition to the rmq_watcher\..* pattern already required for cooldown, the Airflow RMQ user needs:

# configure: declare the exchange / alternate-exchange / its queues
rabbitmqctl set_permissions -p <vhost> <user> "^(rmq_watcher\\..*|jetstat\\.airflow(\\.unrouted|\\.log)?|...)$" \
  "^(rmq_watcher\\..*|jetstat\\.airflow(\\.unrouted|\\.log)?|...)$" \
  "^(rmq_watcher\\..*|jetstat\\.airflow(\\.unrouted)?|...)$"

configure/write are needed on {exchange}(.unrouted|.log)?; read is additionally needed on {exchange}(.unrouted)? because binding a queue from an exchange requires read access on the source exchange, not just configure on the destination queue. Replace jetstat.airflow with whatever name is actually passed to exchange=.

Migrating from queue= to exchange=: switching an existing subscription does not clean up after itself — the old, manually-created queue is not deleted automatically and is left without a consumer once the DAG file is redeployed with exchange=. Remove it manually once the migration is confirmed working.

Renaming the DAG: changing dag_id provisions a new rmq_watcher.sub.{new_dag_id} queue/bindings on the next reconcile cycle. The old rmq_watcher.sub.{old_dag_id} becomes orphaned (its subscription metadata no longer exists in any parsed DAG file — see ADR-0005) and is not removed automatically. Delete it manually.

Monitoring:

• RabbitMQ Management UI — rmq_watcher.sub.{dag_id} should show consumer count > 0 when the DAG's subscription is active
• Airflow logs — WARNING for orphaned rmq_watcher.sub.* queues/bindings (with a rabbitmqadmin delete queue ... hint); ERROR for a skipped bind-diff (Management API unreachable) or for an exchange property conflict (PRECONDITION_FAILED — the exchange name is already used by something else with different properties)

Rollback: remove exchange=/routing_keys=/routing_key_ids=/routing_key_status= from the decorator and redeploy. rmq_watcher.sub.{dag_id} becomes orphaned — a WARNING appears in the logs, the TTL (8h) caps unbounded growth, and manual cleanup follows the hint in the WARNING text. The exchange itself and its .unrouted/.log queues are not touched by rollback (other DAGs may still be using them).

Payload passed to the DAG

conf = context["dag_run"].conf
# Immediate trigger (cooldown=0 or no cooldown):
# {
#     "source":          "immediate",
#     "body":            "<UTF-8 decoded message body>",
#     "headers":         {"key": "value", ...},
#     "routing_key":     "orders.created",
#     "queue":           "orders",
#     "subscription_id": 42,
# }
# Exchange-mode triggers (exchange=) reuse this exact "immediate" shape — "queue" is
# always rmq_watcher.sub.{dag_id} (not the exchange name), and "routing_key" is the
# actual matched routing key (e.g. "<id>.<status>" for the routing_key_ids form).
#
# Cooldown trigger (fired after TTL expires in rmq_watcher.fire):
# {
#     "source":          "cooldown",
#     "dag_id":          "<dag_id>",
#     "body":            "",        # empty — original message body not preserved
#     "headers":         {},        # empty — original headers not preserved
#     "routing_key":     "<dag_id>",
#     "queue":           "rmq_watcher.fire",
#     "subscription_id": None,
# }

Subscription Management

dag_file subscriptions are read-only in the UI — reconciliation overwrites DB from code every 60 s. Only the enabled toggle can be changed via UI for code-managed subscriptions.

Exchange-mode subscriptions show up in the UI like any other dag_file subscription — by their queue name (rmq_watcher.sub.{dag_id}) only. The exchange/routing_keys metadata is not displayed there; the DAG source file is the single source of truth for that.

Any subscription whose dag_id doesn't correspond to an active Airflow DAG is marked on the page with a ⚠ dag not found badge. This is a one-sided signal: its presence means a real problem — a matching message will be ACKed without triggering anything — but its absence guarantees nothing. Three reasons the badge can miss a broken subscription: is_paused is not checked, so a paused DAG never gets the badge even though a matching message is still ACKed without triggering a run, exactly like a missing DAG (see ADR 0006); a just-added DAG may be badged for a short while until the Scheduler parses it; and a just-deleted or renamed DAG conversely stays un-badged for a while until DagModel.is_active catches up. The last two are expected lag, not bugs.

Subscriptions Page Access

The page and its menu entry belong to two roles: Admin and Op. Viewer and User see neither the entry nor the RabbitMQ category it hangs under, and a direct visit to /rmq-watcher/subscriptions is refused.

Admin gets the page from Airflow itself, which hands that role every non-DAG permission on every role synchronisation. Op gets it from the provider: at each webserver start the plugin grants the role can_read, can_create, can_edit and can_delete on the resource RMQ Subscriptions, plus menu_access on Subscriptions and on RabbitMQ. The grant runs on every start, so upgrading the provider opens the page to Op by itself, with nothing to do by hand.

Two starts are needed on a virgin metadata database: Airflow creates the Op role after the plugin's callback has run, so the first start finds no role, logs a warning and leaves the grant to the start after it. The grant also needs the FAB auth manager, which is the default one — under another auth_manager there is no role to grant to and the callback logs a warning instead.

Warning. Those permissions include creating and deleting subscriptions, so access to the page is the right to make an arbitrary DAG run on a message from a queue. An installation where Op should not hold that right puts grant_op_access = false in the [rmq_watcher] section of airflow.cfg — or sets AIRFLOW__RMQ_WATCHER__GRANT_OP_ACCESS=false in the webserver's environment — and restarts the webserver.

The webserver is the only process that acts on the switch. A plugin blueprint is registered by every application that builds an appbuilder — the FAB commands behind airflow users, airflow roles and airflow sync-perm build one too — and the permissions of a page belong to the application that serves the page, so the grant asks for Airflow's own web UI and stands aside where there is none. A command run in a shell that does not carry the switch therefore leaves the role alone instead of answering to a value the operator never set there.

The switch governs the access, not merely the grant: a false value removes the same six permissions from the Op role at the next webserver start. false, off, no, n and 0 all say it, and so does a value in no spelling the switch knows: an unreadable answer closes the page rather than holding it open. Granting or revoking them by hand while the switch is true or unset achieves nothing — the next start puts the provider's own answer back.

It is an Airflow configuration option and not an Airflow Variable because Airflow gives the Op role create, edit and delete on Variables. A switch kept there is one the very role it restricts can set back to true for itself, taking the page back at the next webserver restart; airflow.cfg and the process environment are outside the reach of every role the web UI has, which is what a switch that withdraws a privilege has to be.

menu_access is granted on the displayed names of the entry and of its category, and the FAB menu is flat. Another installed plugin that names its own entry Subscriptions or its own category RabbitMQ therefore shares those two permissions, and a revocation closes its menu to Op along with this one.

Any other role opens the page through Airflow's CLI:

airflow roles add-perms MyRole -a can_read -r "RMQ Subscriptions"
airflow roles add-perms MyRole -a menu_access -r "Subscriptions"
airflow roles add-perms MyRole -a menu_access -r "RabbitMQ"

Without menu_access on the category the entry stays out of the menu even with the other two granted. Without can_create, can_edit and can_delete the page still draws all eight of its controls — the template checks no permission — and each of them answers with a refusal.

Connection Resilience

The watcher holds long-lived AMQP connections inside the Scheduler, so it treats a broker restart, a silently dropped TCP link, a resource alarm or an unavailable database as normal weather: once the external problem is gone, subscriptions come back on their own. The design is written up in ADR 0007; the short version is four layers.

1 — Heartbeats and per-call timeouts. Every async connection carries a heartbeat (30 s by default), and every AMQP call — connect, channel(), declare, bind, publish — runs under connect_timeout/rpc_timeout (see Connection Timing). A link that dies turns into an exception instead of a call that waits forever. The connect carries the watcher's own bound rather than the library's: connect_robust(timeout=…) covers the TCP connect and the AMQP handshake and stops there, while the wait that follows it — for the broker to declare the connection unblocked — is outside it, and under a resource alarm that wait lasts as long as the alarm does. A connect that runs out of time is not cancelled but left in flight, and its connection stays in the pool: cancelling it would poison the connection object for good, and dropping it would leave an open connection to the broker that nothing can close. Each conn_id builds its connections under a lock of its own, and that lock covers starting a connect rather than waiting for one: however many callers want the connection, the broker is asked for a single attempt, and they wait for it side by side rather than one behind another. A broker that stopped answering therefore costs each caller one connect_timeout, and holds up the conn_ids that use it and no others. A pooled connection is handed out only while it is usable — not merely unclosed: a reconnect that never finished leaves an object with no transport under it, which reports itself open and fails every call made through it, so it is replaced rather than handed to the next caller.

2 — Liveness watchdog. Every reconcile cycle the watcher asks the broker whether its consumers are still registered. Each subscription attaches under its own consumer tag rmq_watcher.{hostname}.{pid}.{sub_id}.{nonce} (the cooldown fire consumer uses fire in place of {sub_id}), and the check looks for that exact tag — a foreign consumer on the same queue, a second DAG subscribed to it, or another scheduler replica in HA cannot vouch for ours, and the {nonce}, fresh for every attach, keeps a ghost consumer left on the broker by a close() that never returned from vouching for the task that replaced it. The tag is claimed — and the subscription reported as listening — only once the broker has confirmed the registration, which itself runs under rpc_timeout: a basic.consume that never comes back would otherwise leave the check looking for a tag the broker never held. With management_url configured the tag is looked up in GET /api/consumers/{vhost}; without it, the watcher falls back to a passive queue_declare probe on a separate channel, whose verdict covers every consumer of that conn_id at once. Either way the tag is also checked against the channel the client holds it on: RabbitMQ cancels a consumer of ours — a deleted queue, a quorum-queue or stream leader change, the node hosting a classic queue restarting under a client connected to another node — with a basic.cancel that closes neither the connection nor the channel, so every probe of that connection keeps succeeding while nothing consumes; the tag missing from our own channel is what condemns it. Two negative checks in a row (never less than two reconcile intervals) cancel exactly those subscriptions whose own tag the broker did not confirm and drop the connection they share — restarting a task alone would hand it the same zombie connection out of the pool. The status row does not wait for the second check: a conn_id whose consumers the broker denies reads error with the reason and the check number straight away. The confirmed subscriptions of that conn_id are neither cancelled nor restarted: the shared connection is dropped underneath them, they reconnect through their own retry loop and surface the drop as a transient consumer error (a WARNING names them). The fire consumer goes down with the connection it runs on, because it holds the connection object it was handed at startup for its whole life; the reconcile cycle starts it again — on the connection the pool holds now — when that object was replaced under it, and moves it when the cooldown conn_id moves. A conn_id whose only attached consumer is the fire consumer is recreated on that consumer's verdict alone: nothing else on it can be asked about — its subscriptions are in their own retry loop — and a restart would otherwise be handed the same pooled object every cycle. A conn_id whose tasks are alive but never reach listening for two cycles offers the check no candidate at all, so the connection itself is asked whether it still answers an RPC: silence recreates it, an answer leaves it in place and the row says the fault is downstream of the connection — a subscription stuck in error because its trigger keeps failing is attached to a healthy broker connection. A Management API failure is "no data", not a verdict: it logs a WARNING, leaves the counters alone and keeps the stored status — until the second failure in a row, on which the passive declare takes over; an unreadable Airflow connection is treated the same way. A hung AMQP probe, on the other hand, counts as dead. The same conn_id is never recreated more often than once every 5 cycles; a verdict held back by that limit is logged and stored as degraded.

3 — Cycle timeout. The whole reconcile iteration runs under a budget of max(reconcile_interval × 3, 300) seconds. Exceeding it stops the manager and recreates the event loop after 30 s, with an ERROR naming the phase (migrate / scan / sync / read subs / reconcile) and how long it took. The budget is deliberately generous — hitting it pauses consumption on every conn_id — while the per-call timeouts of layer 1 catch a stuck network operation much earlier and only for the subscription that owns it. Each blocking step of the cycle (the DAG-file scan, the sync to the database, the subscription read) has a 60-second bound of its own and only ever one attempt in flight, so an unresponsive database ends a single cycle rather than costing the loop, and cannot fill the cycle's four-worker pool with stuck copies of the same call.

4 — Publishing on its own connection. Cooldown publishing uses a separate, lazily opened connection per conn_id. Under a memory or disk alarm RabbitMQ blocks publishing connections by stopping to read from the socket, which would otherwise take the acknowledgements of every consumer on the same connection down with it — and a connection the broker has flagged as blocked stops writing on the client side too (aiormq holds its own writer until connection.unblocked arrives), so the split is what keeps basic.ack moving: the consuming connection publishes nothing and is never flagged. What the split does not cover is a connection opened during an alarm, which does not become usable until the alarm clears — hence the client-side bound on connect in layer 1. A placeholder the broker returns as unroutable — the pending queue is missing, so nothing would ever fire the DAG after the window — puts the delivery back on the queue instead of acknowledging it. Two publish timeouts in a row recreate the publishing connection alone and leave consumption untouched, and a publish that keeps failing pauses that subscription for a growing interval (1 s, doubling up to 60 s) before reattaching, which stretches how long an alarm can last before it costs the message: every failed attempt requeues the delivery and spends one of the quorum queue's 20 redeliveries, and the pauses spread those 20 over roughly a quarter of an hour instead of seconds. An alarm outlasting that still ends with the broker dead-lettering or dropping the delivery. The cooldown fire consumer runs on the lowest conn_id among the cooldown subscriptions, which keeps it from being moved by the order rows happen to come back in; cooldown subscriptions spread over several conn_ids are logged as an ERROR, because the pending queues live on one broker only.

Airflow Variables:

Both are re-read at the start of every cycle in a thread pool under a short timeout of their own, so a changed Variable takes effect on the next cycle; a read still stuck in a worker blocks the next one from starting, and the loop gives up on it at its own timeout and keeps the values it already has instead of stalling; a database that refuses the read outright is not told apart from an unset Variable — it reads as no value at all, and the built-in defaults take effect until a later cycle reads a real one.

Airflow configuration:

It is read once per webserver start, when the page's permissions are set, and it is read from airflow.cfg and the webserver's environment and from nowhere else: the _cmd and _secret indirections that would reach a shell command or a secrets backend are honoured only for the options Airflow lists as sensitive, and this one is not among them. The read therefore touches neither the network nor the database, and nothing in it can hang while the webserver is still building its application. It happens before anything is touched, and the answer is read towards withholding. 1, t, true, y, yes and on hold the page open; 0, f, false, n, no and off close it — case, surrounding whitespace and a trailing # comment do not matter. A value in none of those spellings closes the page as well, with the value named in a warning in the log: the switch has one job, and it is taking a privilege away, so what nobody can read is answered in the direction that withholds the privilege rather than the one that hands it out. Only an option nobody set at all is read as the default true. The provider declares the option in its configuration metadata, so Airflow's Configuration page, airflow config list and airflow config get-value rmq_watcher grant_op_access show it with its default on an installation where nobody has set it.

What the Subscriptions page shows. The Connections block at /rmq-watcher/subscriptions lists one row per conn_id:

A row is written for every conn_id that appears in the subscription list, including one whose tasks all died. The counts and last_reconcile_at move on every cycle, which is what proves the loop is alive; status keeps its last stored value whenever the check produced no data.

Rows are never deleted. A conn_id no enabled subscription mentions any more stops being stamped and keeps the age it had at that moment, so its row grows permanently stale — including after the documented way to pause consumption, turning the last subscription of that conn_id off in the UI. Turning one back on resumes the stamps on the next cycle.

The Status column of the Subscriptions table further down the page is a different one: it reports a single subscription's own task — connecting, listening, error or disconnected — while the Connections block above reports the conn_id those tasks share (a grouped row shows how many of the group's subscriptions are listening).

Schema migration. The status table carries broker_consumer_count and last_reconcile_at alongside its other columns, and an idempotent migration adds whatever a given database lacks: it inspects the live table and issues ALTER TABLE … ADD COLUMN only for what is missing. The plugin runs it once at load, and the watcher loop retries it at the start of each cycle (one attempt in flight, with a growing backoff between failures) until it succeeds, so a database that was unreachable at load time does not leave the process running against an outdated table. While the table is behind, the Connections block shows a short notice instead of failing to render the page.

Delivery Guarantee (immediate mode)

For subscriptions with cooldown=0 the watcher acknowledges a delivery only after trigger_dag has created the DAG run:

• A failed trigger is NACKed with requeue, logged as a WARNING and retried with a growing backoff (1 s, doubling up to 60 s); the subscription status becomes error with the reason, so a subscription that stopped starting DAG runs is visible on the page. One answer out of trigger_dag is exempt, and it is the next bullet.
• One trigger failure the status keeps quiet about: a DAG Airflow holds no serialized version of yet — DagNotFound out of trigger_dag, which is the state a freshly started scheduler is in until its DAG processor has parsed the files. The connection, the channel and the consumer registration are all intact, and what the delivery waits for is the rest of Airflow, so the row is held at listening — which is what the consumer is doing, and what keeps the subscription a candidate of the liveness check. An error an earlier, unrelated failure left there is cleared by the first not-yet-serialized delivery rather than carried through the whole warmup, so a conn_id whose subscriptions are all warming up keeps its own row green as well. The delivery itself is requeued and paused for the same growing interval. The quiet is bounded: 25 such deliveries in a row are long enough to be a failure of their own, and the row then reads error naming the missing serialized version, with Airflow's own Dag id ... not found kept behind it. The streak counts one uninterrupted warmup — a reconnect starts it over, and so does any delivery that goes through or fails for another reason — and every tenth not-yet-serialized delivery is logged as a WARNING, which on a quorum queue that spends its delivery-limit before the streak runs out is the only trace the warmup leaves. The cooldown fire consumer treats the same answer the same way and holds its reported status at listening: it has no row of its own, but that status is what keeps it a candidate of the liveness check, so a fire consumer that stops consuming during a warmup is noticed all the same. One such consumer serves every cooldown DAG, so its events are counted per dag_id to pick the log level and for nothing else — every tenth is a WARNING. A cooldown DAG that never gets serialized therefore shows up in the scheduler log alone.
• Deduplication rests on the producer. When the message carries an AMQP message_id, the run id is deterministic (rmq__{queue}__{message_id}, sanitized and truncated with a digest when it does not fit), so a redelivery lands on the run it already produced and is acknowledged as a duplicate. Two different messages sharing one message_id therefore collapse into a single DAG run. Without message_id the run id carries a timestamp instead and a redelivery starts the DAG again.
• Boundary of the guarantee: a DAG that is paused, inactive or missing ends in a terminal ACK — the event is dropped on purpose, with a WARNING in the log. NACKing it would turn a paused DAG into an accumulator of redeliveries.
• Boundary of the silence: each not-yet-serialized delivery spends one unit of the queue's delivery-limit — 20 by default on a quorum queue — so an event waiting out a long warmup can be dead-lettered or dropped by the broker before the streak reaches 25. The row goes on saying listening, nothing records that the event existed, and the WARNING on every tenth delivery is the only trace left (docs/backlog/warmup-loses-the-message-invisibly.md).
• Messages that do not match the filter are NACKed with requeue (see ADR 0002) and go back to the head of the queue.

prefetch is deliberately not set. Because filter misses are requeued and return to the head of the queue, any finite prefetch window would eventually fill up with them and consumption would stop for good — while the status still read connected. The cost is accepted explicitly: the number of unacknowledged deliveries is unbounded, so a queue with a large backlog is read into scheduler memory, and the oldest delivery can hit the broker's consumer_timeout (30 minutes by default), after which RabbitMQ closes the channel with PRECONDITION_FAILED and returns the whole window to the queue. If large backlogs are expected on a watched queue, raise consumer_timeout in the RabbitMQ configuration.

Best Practices

• Use a dedicated queue per DAG trigger (e.g. orders.airflow-trigger separate from orders). Avoids NACK hot-loops on quorum queues and interference with other consumers.
• To pause message consumption without stopping the DAG: toggle the subscription off in the UI rather than pausing the DAG. Pausing the DAG acks messages silently.
• In multi-scheduler HA deployments each active scheduler runs its own consumer, which may cause duplicate runs. The protection is a deterministic run_id: subscriptions with cooldown > 0 always have one (rmq_cooldown__{dag_id}__{message_id}), and immediate ones have it whenever the producer sets an AMQP message_id (rmq__{queue}__{message_id}) — in both cases the second replica's trigger lands on the DAG run the first one already created. Without a message_id an immediate run id carries a timestamp instead and both replicas start the DAG; set max_active_runs=1 as a lightweight mitigation there.

Example DAGs

The package includes several example DAGs in docs/example_dags/. All examples use the TaskFlow API (@dag / @task decorators) and demonstrate how to process consumed messages in downstream tasks via XCom.

Repository Structure

airflow-provider-rmq/
├── airflow_provider_rmq/
│   ├── __init__.py                  # Provider metadata & get_provider_info()
│   ├── hooks/
│   │   └── rmq.py                   # RMQHook
│   ├── operators/
│   │   ├── rmq_publish.py           # RMQPublishOperator
│   │   ├── rmq_consume.py           # RMQConsumeOperator
│   │   └── rmq_management.py        # RMQQueueManagementOperator
│   ├── sensors/
│   │   └── rmq.py                   # RMQSensor
│   ├── triggers/
│   │   └── rmq.py                   # RMQTrigger
│   ├── utils/
│   │   ├── amqp.py                  # build_amqp_connection(), get_amqp_timeouts(), call_with_timeout()
│   │   ├── executor.py              # BoundedExecutor (thread pool for blocking calls)
│   │   ├── filters.py               # MessageFilter
│   │   ├── management.py            # Management HTTP API client (bindings, queue consumers)
│   │   ├── metrics.py               # Stats counters, no-op without statsd
│   │   └── ssl.py                   # build_ssl_context()
│   └── watcher/
│       ├── decorators.py            # @rmq_trigger
│       ├── subscription_builder.py  # build_subscriptions(), has_exchange_conflict()
│       ├── subscription_form.py     # parse_cooldown(), parse_filter_data() (UI form parsing)
│       ├── models.py                # RMQSubscription, RMQConnStatus, WatcherSession
│       ├── tunables.py              # settings tuning the watcher, with their defaults
│       ├── consumer.py              # RMQConsumerManager
│       ├── orphan_tracker.py        # OrphanTracker (cooldown + exchange-mode orphan detection)
│       ├── listener.py              # RMQWatcherListener (Scheduler Listener)
│       ├── views.py                 # RMQWatcherView (Flask-AppBuilder UI)
│       └── plugin.py                # RMQWatcherPlugin (AirflowPlugin)
├── docs/
│   ├── adr/                         # Architecture decision records
│   └── example_dags/                # Example DAGs
├── tests/                           # Unit tests
├── CHANGELOG.md
├── pyproject.toml
└── readme.md

Running Tests

# Install dev dependencies
pip install -e ".[dev]"

# Run all tests
pytest tests/

# Run specific test module
pytest tests/test_trigger.py -v

License

Apache License 2.0. See LICENSE for details.