This is an automated email from the ASF dual-hosted git repository.
hussein-awala 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 e606e1b9527 Add trigger `queue` support for `AsyncCallback` and
`BaseEventTrigger` (#71346)
e606e1b9527 is described below
commit e606e1b9527a2ff368ae7dc8993a870377c2bff0
Author: Zach <[email protected]>
AuthorDate: Thu Aug 20 05:59:15 2026 -0400
Add trigger `queue` support for `AsyncCallback` and `BaseEventTrigger`
(#71346)
* Add trigger queue support for `BaseEventTrigger`
* Enable trigger queue support for async callbacks
* Add newsfragment
* Address classattr bug
* Address feedback
* remove unintentional uv.lock metadata addition
* Adjust queue value to accomodate subclasses not calling super().__init__()
* Fix MessageQueueTrigger queue collision with trigger queue assignment
MessageQueueTrigger's deprecated `queue` constructor param (a broker
URI) was stored under self.queue, silently colliding with the new
triggerer-routing `queue` attribute added to BaseTrigger/BaseEventTrigger.
Any Dag still using that deprecated call style would have its
asset-watcher Trigger row's queue column set to the broker URI, and
since every triggerer started without --queues filters on queue IS
NULL, the trigger would never be picked up again, silently.
* Fix IBM MQ test asserting on renamed MessageQueueTrigger attribute
PR #41 renamed MessageQueueTrigger's deprecated broker-URI storage from
self.queue to self.queue_uri to avoid colliding with the new
BaseEventTrigger.queue triggerer-routing attribute, but missed this
provider-local test asserting on the old attribute name.
* Fix compat-test failures from asserting on unreleased
MessageQueueTrigger.queue
Provider distributions must keep passing tests against previously
released Airflow versions. BaseEventTrigger.queue is a new attribute
added by this branch, so any Airflow release before it ships doesn't
have the attribute at all, and BaseTrigger has no class-level default
either in those older releases. Assertions on trigger.queue therefore
raised AttributeError under the 3.0.6/3.1.8/3.3.0 compat test jobs.
Guard with getattr(trigger, "queue", None), matching the same pattern
already used for this attribute in airflow.serialization.encoders.
* Let MessageQueueTrigger set the triggerer queue via a distinct parameter
The `queue` constructor keyword is already claimed by the deprecated
broker queue URI, so there was no way for a user to route a
MessageQueueTrigger to a specific triggerer queue via
BaseEventTrigger.queue. Add a `triggerer_queue` parameter that is
forwarded to BaseEventTrigger.__init__ instead.
* Skip triggerer-queue assertions on pre-3.4 Airflow-core in compat tests
BaseEventTrigger.queue is unreleased (targets 3.4.0), so the Compat test
matrix against older published airflow-core wheels fails with
AttributeError when asserting on it directly.
* Revert "Skip triggerer-queue assertions on pre-3.4 Airflow-core in compat
tests"
This reverts commit 19935856ce85f97ac31470a0452b31b26be15351.
* Let MessageQueueTrigger's trigger queue work on any Airflow-core version,
and fix trigger-queue terminology
The previous fix relied on BaseEventTrigger.__init__ to set the queue
attribute, which older published airflow-core releases silently drop,
breaking the Compat test matrix. Storing the value under our own name
and exposing it through a `queue` property/setter works regardless of
the installed core version, so no version-gated tests are needed.
Also corrects "triggerer queue" to "trigger queue" throughout, which is
the term used elsewhere for this concept.
* Undo `uv.lock` extras shift from local `uv run` call.
* remove news fragment
---------
Co-authored-by: Xu Han <[email protected]>
---
.../docs/authoring-and-scheduling/deferring.rst | 54 ++++++++++++++--------
airflow-core/docs/howto/deadline-alerts.rst | 15 ++++++
.../src/airflow/dag_processing/collection.py | 1 +
airflow-core/src/airflow/models/callback.py | 10 +++-
airflow-core/src/airflow/models/trigger.py | 2 +-
airflow-core/src/airflow/serialization/decoders.py | 1 +
airflow-core/src/airflow/serialization/encoders.py | 7 ++-
airflow-core/src/airflow/triggers/base.py | 26 +++++++++--
airflow-core/src/airflow/triggers/callback.py | 14 +++++-
.../tests/unit/dag_processing/test_collection.py | 30 ++++++++++++
airflow-core/tests/unit/models/test_callback.py | 18 ++++++++
airflow-core/tests/unit/models/test_trigger.py | 18 ++++++++
.../tests/unit/serialization/test_encoders.py | 39 ++++++++++++++++
.../unit/serialization/test_serialized_objects.py | 30 ++++++++++++
.../tests/unit/triggers/test_base_trigger.py | 10 ++++
airflow-core/tests/unit/triggers/test_callback.py | 19 ++++++++
.../common/messaging/triggers/msg_queue.py | 41 ++++++++++++----
.../common/messaging/triggers/test_msg_queue.py | 29 +++++++++++-
.../ibm/mq/tests/unit/ibm/mq/queues/test_mq.py | 3 +-
task-sdk/src/airflow/sdk/definitions/callback.py | 18 +++++++-
.../src/airflow/sdk/execution_time/task_runner.py | 6 +--
.../tests/task_sdk/definitions/test_callback.py | 16 +++++++
22 files changed, 364 insertions(+), 43 deletions(-)
diff --git a/airflow-core/docs/authoring-and-scheduling/deferring.rst
b/airflow-core/docs/authoring-and-scheduling/deferring.rst
index 8a074e4d66e..e3577c437aa 100644
--- a/airflow-core/docs/authoring-and-scheduling/deferring.rst
+++ b/airflow-core/docs/authoring-and-scheduling/deferring.rst
@@ -493,6 +493,8 @@ According to `benchmarks
<https://github.com/apache/airflow/pull/58803#pullreque
You can determine a suitable value for your deployment by creating a large
number of triggers (for example, by triggering a Dag with many deferrable
tasks) and observing both how the load is distributed across Triggerers in your
environment and how long it takes for all Triggerers to pick up the triggers.
+.. _deferring/triggerer_queue_assignment:
+
Controlling Triggerer Host Assignment Per Trigger
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -508,7 +510,7 @@ Under some circumstances, it may be desirable to assign a
Trigger to a specific
If you are using :doc:`Multi-Team mode</core-concepts/multi-team>`, the
``--team-name`` option provides
native team-scoped triggerer assignment for all trigger types
(task-created, event-driven, and callback).
See :ref:`Team-scoped Triggerer <multi-team-triggerer>` in the Multi-Team
documentation.
- The ``--queues`` option described below is an older, queue-based mechanism
that can be combined with
+ The ``--queues`` option described below is a distinct queue-based
mechanism which may be combined with
``--team-name`` if needed.
To enable queue assignment for triggers, do the following:
@@ -532,27 +534,41 @@ Trigger Queue Assignment Caveats
''''''''''''''''''''''''''''''''
This feature is only compatible with executors which utilize the task
``queue`` concept
-(such as the
:ref:`CeleryExecutor<apache-airflow-providers-celery:celery_executor:queue>`).
-
-Additionally, queue assignment is currently only compatible with the subset of
triggers originating from a task's defer ``method``.
-
-+------------------------------------------------------------------------------------------------------+----------------------+----------------------------------------------------------------------------------+
-| Trigger Type
| Supports queues? | Triggerer assignment when
:ref:`config:triggerer__queues_enabled` is ``True`` |
-+======================================================================================================+======================+==================================================================================+
-| Task-created Trigger instances
| Yes | Any triggerer with the task
queue present in its ``--queues`` option |
-+------------------------------------------------------------------------------------------------------+----------------------+----------------------------------------------------------------------------------+
-| :doc:`Event-Driven Triggers<../authoring-and-scheduling/event-scheduling>`
| No | Any triggerer running without
the ``--queues`` option |
-+------------------------------------------------------------------------------------------------------+----------------------+----------------------------------------------------------------------------------+
-| Triggers from async
:doc:`Callbacks<../administration-and-deployment/logging-monitoring/callbacks>`
| No | Any triggerer running without the ``--queues``
option |
-+------------------------------------------------------------------------------------------------------+----------------------+----------------------------------------------------------------------------------+
-
-If you use queues for task-based triggers, while **also** using event-based
triggers and/or callback triggers,
-you must run one or more triggerer hosts **without** the ``--queues`` option,
so the latter 2 types of triggers are still run.
+(such as the
:ref:`CeleryExecutor<apache-airflow-providers-celery:celery_executor:queue>`)
for
+task-created triggers. :doc:`Event-Driven
Triggers<../authoring-and-scheduling/event-scheduling>`
+are not tied to a task queue, but a
:class:`~airflow.triggers.base.BaseEventTrigger` subclass can
+be assigned to a queue explicitly by passing ``queue=`` to
``super().__init__()``, independently
+of any executor. Similarly, triggers from async
+:doc:`Callbacks<../administration-and-deployment/logging-monitoring/callbacks>`
can be assigned to a
+queue by passing ``queue=`` to :class:`~airflow.sdk.AsyncCallback`.
+
++------------------------------------------------------------------------------------------------------+-----------------------------------------------------------------------------------------+
+| Trigger Type
| Triggerer assignment source when
:ref:`config:triggerer__queues_enabled` is ``True`` |
++======================================================================================================+=========================================================================================+
+| Task-created Trigger instances
| Any triggerer with the task queue present in its
``--queues`` option |
++------------------------------------------------------------------------------------------------------+-----------------------------------------------------------------------------------------+
+| :doc:`Event-Driven Triggers<../authoring-and-scheduling/event-scheduling>`
| Any triggerer whose ``--queues`` includes the
trigger's ``queue``, or any |
+|
| triggerer without ``--queues`` if no ``queue`` was
set |
++------------------------------------------------------------------------------------------------------+-----------------------------------------------------------------------------------------+
+| Triggers from async
:doc:`Callbacks<../administration-and-deployment/logging-monitoring/callbacks>`
| Any triggerer whose ``--queues`` includes the callback's ``queue``, or any
|
+|
| triggerer without ``--queues`` if no ``queue`` was
set |
++------------------------------------------------------------------------------------------------------+-----------------------------------------------------------------------------------------+
+
+If you run a mix of triggers with queues **and** without queues assigned, (of
any trigger type), you must run at least one triggerer host
+**without** the ``--queues`` option, so the latter are still run.
.. note::
To enable trigger queues, you must set the ``--queues`` option on one or
more triggerers' startup command (these values may differ between the various
triggerers).
- If you set the ``--queue`` value of a triggerer to some value which no
task queues exist for, that triggerer will never run any triggers.
- Similarly, all ``triggerer`` instances running without the ``--queues``
option will only consume event-driven and callback-based triggers.
+ If you set the ``--queue`` value of a triggerer to some value which no
triggers are assigned (either from their task queue, or from their explicit
``queue`` field), then that triggerer
+ will never run any triggers. Similarly, all triggerer instances running
without the ``--queues`` option will only consume callback-based triggers and
event-driven triggers that were not
+ assigned an explicit ``queue``.
+
+An event-driven trigger's ``queue`` is set once, when the trigger is
registered for an
+:class:`~airflow.sdk.AssetWatcher` during Dag processing. A callback trigger's
``queue`` is set
+from the ``queue`` passed to :class:`~airflow.sdk.AsyncCallback` when the
deadline is missed and the
+callback is queued on the Triggerer. Both are independent of
+:doc:`Multi-Team mode</core-concepts/multi-team>`'s ``team_name`` scoping —
the two can be
+combined.
The following example shows how to run HA triggerers so that all trigger types
are run (assuming all tasks'
diff --git a/airflow-core/docs/howto/deadline-alerts.rst
b/airflow-core/docs/howto/deadline-alerts.rst
index 124b7d87040..76b18a99e45 100644
--- a/airflow-core/docs/howto/deadline-alerts.rst
+++ b/airflow-core/docs/howto/deadline-alerts.rst
@@ -362,6 +362,21 @@ A **custom asynchronous callback** might look like this:
):
EmptyOperator(task_id="example_task")
+.. tip::
+ ``AsyncCallback`` accepts an optional ``queue`` parameter to assign the
resulting trigger to a
+ specific :ref:`trigger queue <config:triggerer__queues_enabled>`. If not
specified, the
+ callback trigger runs on any triggerer started without the ``--queues``
option. See
+ :ref:`Controlling Triggerer Host Assignment Per Trigger
<deferring/triggerer_queue_assignment>`
+ for details.
+
+ .. code-block:: python
+
+ AsyncCallback(
+ my_callback,
+ kwargs={"msg": "deadline missed"},
+ queue="alerts",
+ )
+
Templating and Context
^^^^^^^^^^^^^^^^^^^^^^
diff --git a/airflow-core/src/airflow/dag_processing/collection.py
b/airflow-core/src/airflow/dag_processing/collection.py
index 8361cc43b0c..11e1099de7b 100644
--- a/airflow-core/src/airflow/dag_processing/collection.py
+++ b/airflow-core/src/airflow/dag_processing/collection.py
@@ -1174,6 +1174,7 @@ class AssetModelOperation(NamedTuple):
classpath=triggers[trigger_hash]["classpath"],
kwargs=triggers[trigger_hash]["kwargs"],
team_name=team_name,
+ queue=triggers[trigger_hash].get("queue"),
)
for trigger_hash in all_trigger_hashes
if trigger_hash not in orm_triggers
diff --git a/airflow-core/src/airflow/models/callback.py
b/airflow-core/src/airflow/models/callback.py
index 1e8e758450d..352885a420e 100644
--- a/airflow-core/src/airflow/models/callback.py
+++ b/airflow-core/src/airflow/models/callback.py
@@ -102,6 +102,13 @@ class
ImportPathExecutorCallbackDefProtocol(ImportPathCallbackDefProtocol, Proto
executor: str | None
+@runtime_checkable
+class ImportPathAsyncCallbackDefProtocol(ImportPathCallbackDefProtocol,
Protocol):
+ """Protocol for callbacks that use the import path fetch method and
support trigger queue assignment."""
+
+ queue: str | None
+
+
class Callback(Base, BaseWorkload):
"""Base class for callbacks."""
@@ -201,7 +208,7 @@ class Callback(Base, BaseWorkload):
match type(callback_def).__name__:
case "AsyncCallback":
if TYPE_CHECKING:
- assert isinstance(callback_def,
ImportPathCallbackDefProtocol)
+ assert isinstance(callback_def,
ImportPathAsyncCallbackDefProtocol)
return TriggererCallback(callback_def, **kwargs)
case "SyncCallback":
@@ -244,6 +251,7 @@ class TriggererCallback(Callback):
CallbackTrigger(
callback_path=self.data["path"],
callback_kwargs=self.data["kwargs"],
+ queue=self.data.get("queue"),
)
)
self.trigger.team_name = team_name
diff --git a/airflow-core/src/airflow/models/trigger.py
b/airflow-core/src/airflow/models/trigger.py
index 4915637c88d..75578974db8 100644
--- a/airflow-core/src/airflow/models/trigger.py
+++ b/airflow-core/src/airflow/models/trigger.py
@@ -201,7 +201,7 @@ class Trigger(Base):
def from_object(cls, trigger: BaseTrigger) -> Trigger:
"""Alternative constructor that creates a trigger row based directly
off of a Trigger object."""
classpath, kwargs = trigger.serialize()
- return cls(classpath=classpath, kwargs=kwargs)
+ return cls(classpath=classpath, kwargs=kwargs, queue=trigger.queue)
@classmethod
@provide_session
diff --git a/airflow-core/src/airflow/serialization/decoders.py
b/airflow-core/src/airflow/serialization/decoders.py
index 0a8f703de30..49f88480b57 100644
--- a/airflow-core/src/airflow/serialization/decoders.py
+++ b/airflow-core/src/airflow/serialization/decoders.py
@@ -127,6 +127,7 @@ def _decode_asset(var: dict[str, Any]):
trigger={
"classpath": watcher["trigger"]["classpath"],
"kwargs":
smart_decode_trigger_kwargs(watcher["trigger"]["kwargs"]),
+ **({"queue": watcher["trigger"]["queue"]} if "queue" in
watcher["trigger"] else {}),
},
)
for watcher in watchers
diff --git a/airflow-core/src/airflow/serialization/encoders.py
b/airflow-core/src/airflow/serialization/encoders.py
index b3aac16363d..5f64ea0c76c 100644
--- a/airflow-core/src/airflow/serialization/encoders.py
+++ b/airflow-core/src/airflow/serialization/encoders.py
@@ -183,6 +183,7 @@ def encode_trigger(trigger: BaseEventTrigger | dict):
if isinstance(trigger, dict):
classpath = trigger["classpath"]
kwargs = trigger["kwargs"]
+ queue = trigger.get("queue")
# unwrap any kwargs that are themselves serialized objects, to avoid
double-serialization in the trigger's own serialize() method.
unwrapped = {}
for k, v in kwargs.items():
@@ -193,10 +194,14 @@ def encode_trigger(trigger: BaseEventTrigger | dict):
kwargs = unwrapped
else:
classpath, kwargs = trigger.serialize()
- return {
+ queue = getattr(trigger, "queue", None)
+ encoded = {
"classpath": classpath,
"kwargs": {k: _ensure_serialized(v) for k, v in kwargs.items()},
}
+ if queue is not None:
+ encoded["queue"] = queue
+ return encoded
def encode_asset_like(a: BaseAsset | SerializedAssetBase) -> dict[str, Any]:
diff --git a/airflow-core/src/airflow/triggers/base.py
b/airflow-core/src/airflow/triggers/base.py
index ee57b536d5e..01b0e04522f 100644
--- a/airflow-core/src/airflow/triggers/base.py
+++ b/airflow-core/src/airflow/triggers/base.py
@@ -78,7 +78,16 @@ class BaseTrigger(abc.ABC, Templater, LoggingMixin):
let them be re-instantiated elsewhere.
"""
- supports_triggerer_queue: bool = True
+ # Whether a deferred task's queue should be inherited by this trigger when
+ # ``triggerer.queues_enabled`` is set. Subclasses that assign their own
``queue``
+ # directly (e.g. ``BaseEventTrigger``, ``CallbackTrigger``) must set this
to False,
+ # or ``_defer_task`` will overwrite that queue with the deferring task's
queue.
+ trigger_queue_inherited_from_task: bool = True
+
+ # Trigger queue assignment. None means no explicit assignment;
``BaseEventTrigger`` and
+ # ``CallbackTrigger`` set this in their ``__init__``, see
`trigger_queue_inherited_from_task`.
+ # Declared as a class attribute since many provider triggers don't call
``super().__init__()``.
+ queue: str | None = None
def __init__(self, **kwargs):
super().__init__()
@@ -293,15 +302,26 @@ class BaseEventTrigger(BaseTrigger):
See :mod:`airflow.triggers.shared_stream` for the full ack-mode design,
including snapshot-at-fan-out semantics, per-event timeout behavior, and
triggerer-restart redeliver notes.
+
+ **Trigger queue assignment**
+
+ Pass ``queue`` to assign this trigger to a specific trigger queue (see
+ :ref:`config:triggerer__queues_enabled` and the ``--queues`` option of
+ ``airflow triggerer``). When used with team-based triggerer node
assignment,
+ the team and queue function as a logical 'AND'.
"""
- supports_triggerer_queue: bool = False
+ # BaseEventTrigger sets its own `queue` directly (see `__init__` below), so
+ # `_defer_task` must not overwrite it with the deferring task's queue.
+ trigger_queue_inherited_from_task: bool = False
- def __init__(self, **kwargs):
+ def __init__(self, *, queue: str | None = None, **kwargs):
super().__init__(**kwargs)
# Injected by the triggerer before run() is called
self.asset_state_store: AssetStateStoreAccessors | None = None
+ # Read by Dag processing when registering the trigger; unused once
running.
+ self.queue = queue
@staticmethod
def hash(classpath: str, kwargs: dict[str, Any]) -> int:
diff --git a/airflow-core/src/airflow/triggers/callback.py
b/airflow-core/src/airflow/triggers/callback.py
index b27920fed86..7645c42e980 100644
--- a/airflow-core/src/airflow/triggers/callback.py
+++ b/airflow-core/src/airflow/triggers/callback.py
@@ -35,12 +35,22 @@ PAYLOAD_BODY_KEY = "body"
class CallbackTrigger(BaseTrigger):
"""Trigger that executes a callback function asynchronously."""
- supports_triggerer_queue: bool = False
+ # CallbackTrigger sets its own `queue` directly (see `__init__` below), so
+ # `_defer_task` must not overwrite it with the deferring task's queue.
+ trigger_queue_inherited_from_task: bool = False
- def __init__(self, callback_path: str, callback_kwargs: dict[str, Any] |
None = None):
+ def __init__(
+ self,
+ callback_path: str,
+ callback_kwargs: dict[str, Any] | None = None,
+ *,
+ queue: str | None = None,
+ ):
super().__init__()
self.callback_path = callback_path
self.callback_kwargs = callback_kwargs or {}
+ # Read by Trigger.from_object() when persisting the row; unused once
running.
+ self.queue = queue
def serialize(self) -> tuple[str, dict[str, Any]]:
return (
diff --git a/airflow-core/tests/unit/dag_processing/test_collection.py
b/airflow-core/tests/unit/dag_processing/test_collection.py
index ad21350be6b..8c21119305a 100644
--- a/airflow-core/tests/unit/dag_processing/test_collection.py
+++ b/airflow-core/tests/unit/dag_processing/test_collection.py
@@ -395,6 +395,36 @@ class TestAssetModelOperation:
assert len(triggers) == 1
assert triggers[0].team_name == expected
+ @pytest.mark.usefixtures("testing_dag_bundle")
+ @pytest.mark.parametrize(
+ ("queue", "expected"),
+ [pytest.param("my_q", "my_q", id="has-queue"), pytest.param(None,
None, id="no-queue")],
+ )
+ def test_add_asset_trigger_references_populates_queue(self, dag_maker,
session, queue, expected):
+ """Ensure the dag processor tracks the queue value of a
`BaseEventTrigger`-type trigger."""
+ trigger = FileDeleteTrigger(filepath="/tmp/test.txt",
poke_interval=5.0)
+ trigger.queue = queue
+ asset = Asset("trigger_q_asset",
watchers=[AssetWatcher(name="watcher", trigger=trigger)])
+ with dag_maker(dag_id="test_trigger_q_dag", schedule=[asset]) as dag:
+ EmptyOperator(task_id="mytask")
+
+ dags = {dag.dag_id: LazyDeserializedDAG.from_dag(dag)}
+ orm_dags = DagModelOperation(dags, "testing",
None).add_dags(session=session)
+ orm_dags[dag.dag_id].is_paused = False
+
+ asset_op = AssetModelOperation.collect(dags)
+ orm_assets = asset_op.sync_assets(session=session)
+ session.flush()
+
+ asset_op.add_dag_asset_references(orm_dags, orm_assets,
session=session)
+ asset_op.activate_assets_if_possible(orm_assets.values(),
session=session)
+ asset_op.add_asset_trigger_references(orm_assets, session=session)
+ session.flush()
+
+ triggers = session.scalars(select(Trigger)).all()
+ assert len(triggers) == 1
+ assert triggers[0].queue == expected
+
@pytest.mark.usefixtures("testing_dag_bundle")
def test_add_asset_trigger_references_hash_consistency(self, dag_maker,
session):
"""Trigger hash from the DAG-parsed path must equal the hash computed
diff --git a/airflow-core/tests/unit/models/test_callback.py
b/airflow-core/tests/unit/models/test_callback.py
index b2b13582710..27639f8f88f 100644
--- a/airflow-core/tests/unit/models/test_callback.py
+++ b/airflow-core/tests/unit/models/test_callback.py
@@ -117,6 +117,15 @@ class TestCallback:
"dag_id": TEST_DAG_ID,
}
+ def test_get_metric_info_includes_queue(self):
+ """``queue`` is stored in ``self.data`` for a queued AsyncCallback, so
it flows into metric tags."""
+ queued_callback = AsyncCallback(async_callback,
kwargs=TEST_CALLBACK_KWARGS, queue="custom-queue")
+ callback = TriggererCallback(queued_callback,
prefix="deadline_alerts", dag_id=TEST_DAG_ID)
+
+ metric_info = callback.get_metric_info(CallbackState.SUCCESS, "0")
+
+ assert metric_info["tags"]["queue"] == "custom-queue"
+
def test_get_metric_info_dict_values_are_stringified(self):
"""
Regression for ``TypeError: unhashable type: 'dict'`` raised by
OpenTelemetry's
@@ -185,8 +194,17 @@ class TestTriggererCallback:
assert isinstance(callback.trigger, Trigger)
assert callback.trigger.kwargs["callback_path"] ==
TEST_ASYNC_CALLBACK.path
assert callback.trigger.kwargs["callback_kwargs"] ==
TEST_ASYNC_CALLBACK.kwargs
+ assert callback.trigger.queue is None
assert callback.state == CallbackState.QUEUED
+ def test_queue_populates_trigger_queue(self, session):
+ queued_callback = AsyncCallback(async_callback,
kwargs=TEST_CALLBACK_KWARGS, queue="custom-queue")
+ callback = TriggererCallback(queued_callback)
+
+ callback.queue(session=session)
+
+ assert callback.trigger.queue == "custom-queue"
+
@staticmethod
def _queue_callback(session, *, has_bundle, has_team):
from airflow.models.dagbundle import DagBundleModel
diff --git a/airflow-core/tests/unit/models/test_trigger.py
b/airflow-core/tests/unit/models/test_trigger.py
index b48f00622fe..0ab5a89e7ca 100644
--- a/airflow-core/tests/unit/models/test_trigger.py
+++ b/airflow-core/tests/unit/models/test_trigger.py
@@ -1107,6 +1107,24 @@ def test_serialize_sensitive_kwargs():
assert "value2" not in trigger_row.encrypted_kwargs
+def test_from_object_reads_queue_from_trigger():
+ """A ``queue`` attribute on the trigger object is carried onto the
persisted row."""
+ trigger_instance = SensitiveKwargsTrigger(param1="value1", param2="value2")
+ trigger_instance.queue = "custom-queue"
+
+ trigger_row: Trigger = Trigger.from_object(trigger_instance)
+
+ assert trigger_row.queue == "custom-queue"
+
+
+def test_from_object_defaults_queue_to_none_when_not_set_on_trigger():
+ trigger_instance = SensitiveKwargsTrigger(param1="value1", param2="value2")
+
+ trigger_row: Trigger = Trigger.from_object(trigger_instance)
+
+ assert trigger_row.queue is None
+
+
def test_kwargs_not_encrypted():
"""
Tests that we don't decrypt kwargs if they aren't encrypted.
diff --git a/airflow-core/tests/unit/serialization/test_encoders.py
b/airflow-core/tests/unit/serialization/test_encoders.py
index a5fa0f0ddb6..f26d6f46ff4 100644
--- a/airflow-core/tests/unit/serialization/test_encoders.py
+++ b/airflow-core/tests/unit/serialization/test_encoders.py
@@ -250,6 +250,45 @@ class TestEncodeTrigger:
assert result == encode_trigger(trigger)
_assert_fully_serialized(result["kwargs"])
+ def test_encode_omits_queue_when_unset(self):
+ """No ``queue`` key is added when the trigger was not assigned one."""
+ trigger = FileDeleteTrigger(filepath="/tmp/test.txt",
poke_interval=10.0)
+ result = encode_trigger(trigger)
+
+ assert "queue" not in result
+
+ def test_encode_from_trigger_object_with_queue(self):
+ """A trigger's ``queue`` attribute is carried into the encoded dict."""
+ trigger = _CallableKwargsTrigger(topics=())
+ trigger.queue = "team_a"
+ result = encode_trigger(trigger)
+
+ assert result["queue"] == "team_a"
+ assert "queue" not in result["kwargs"]
+
+ def test_encode_from_dict_with_queue(self):
+ """``queue`` on an already-encoded dict is preserved, not dropped as a
kwarg."""
+ already_encoded = {
+ "classpath":
"airflow.providers.standard.triggers.file.FileDeleteTrigger",
+ "kwargs": {"filepath": "/tmp/test.txt", "poke_interval": 10.0},
+ "queue": "team_a",
+ }
+ result = encode_trigger(already_encoded)
+
+ assert result["queue"] == "team_a"
+ assert "queue" not in result["kwargs"]
+
+ def test_re_encode_preserves_queue(self):
+ """Encoding the output of encode_trigger again does not lose
``queue``."""
+ trigger = _CallableKwargsTrigger(topics=())
+ trigger.queue = "team_a"
+
+ first = encode_trigger(trigger)
+ second = encode_trigger(first)
+
+ assert first == second
+ assert second["queue"] == "team_a"
+
def test_assert_fully_serialized_rejects_non_json_values():
"""The guard rejects non-JSON-safe values left in encode_trigger output."""
diff --git a/airflow-core/tests/unit/serialization/test_serialized_objects.py
b/airflow-core/tests/unit/serialization/test_serialized_objects.py
index 8495100cc47..da9a8316585 100644
--- a/airflow-core/tests/unit/serialization/test_serialized_objects.py
+++ b/airflow-core/tests/unit/serialization/test_serialized_objects.py
@@ -835,6 +835,36 @@ def test_decode_asset_with_consumer_teams():
}
[email protected]("q_val", [None, "my_q"])
+def test_decode_asset_preserves_watcher_trigger_queue(q_val: str | None):
+ from airflow.serialization.decoders import decode_asset_like
+
+ decoded = decode_asset_like(
+ {
+ "__type": "asset",
+ "name": "test",
+ "uri": "s3://bucket/key",
+ "group": "asset",
+ "extra": {},
+ "watchers": [
+ {
+ "name": "watcher",
+ "trigger": {
+ "classpath":
"airflow.providers.standard.triggers.file.FileDeleteTrigger",
+ "kwargs": {"filepath": "/tmp"},
+ **({"queue": q_val} if q_val else {}),
+ },
+ }
+ ],
+ }
+ )
+ assert isinstance(decoded, SerializedAsset)
+ if q_val:
+ assert decoded.watchers[0].trigger["queue"] == q_val
+ else:
+ assert "queue" not in decoded.watchers[0].trigger
+
+
def test_decode_asset_defaults_access_control_to_empty_dict():
from airflow.serialization.decoders import decode_asset_like
diff --git a/airflow-core/tests/unit/triggers/test_base_trigger.py
b/airflow-core/tests/unit/triggers/test_base_trigger.py
index 06cb06c94ce..f65d56517b9 100644
--- a/airflow-core/tests/unit/triggers/test_base_trigger.py
+++ b/airflow-core/tests/unit/triggers/test_base_trigger.py
@@ -287,3 +287,13 @@ def test_create_shared_stream_producer_raises_by_default():
"""
with pytest.raises(NotImplementedError,
match="create_shared_stream_producer"):
_PlainEventTrigger.create_shared_stream_producer({})
+
+
+def test_base_event_trigger_queue_not_inherited_from_task():
+ """False so `_defer_task` doesn't overwrite the trigger's own `queue` with
the task's."""
+ assert BaseEventTrigger.trigger_queue_inherited_from_task is False
+
+
+def test_base_event_trigger_queue_defaults_to_none():
+ trigger = _PlainEventTrigger()
+ assert trigger.queue is None
diff --git a/airflow-core/tests/unit/triggers/test_callback.py
b/airflow-core/tests/unit/triggers/test_callback.py
index 99eca603323..6707b9ada2a 100644
--- a/airflow-core/tests/unit/triggers/test_callback.py
+++ b/airflow-core/tests/unit/triggers/test_callback.py
@@ -78,6 +78,25 @@ class TestCallbackTrigger:
"callback_kwargs": expected_serialized_kwargs,
}
+ def test_queue_not_inherited_from_task(self):
+ """False so `_defer_task` doesn't overwrite the trigger's own `queue`
with the task's."""
+ assert CallbackTrigger.trigger_queue_inherited_from_task is False
+
+ def test_queue_attribute_is_not_part_of_serialized_kwargs(self):
+ """``queue`` is read directly off the trigger by
``Trigger.from_object``, not via serialize()."""
+ trigger = CallbackTrigger(
+ callback_path=TEST_CALLBACK_PATH,
+ callback_kwargs=None,
+ queue="custom-queue",
+ )
+
+ assert trigger.queue == "custom-queue"
+ _, kwargs = trigger.serialize()
+ assert "queue" not in kwargs
+
+ def test_queue_defaults_to_none(self, trigger):
+ assert trigger.queue is None
+
@pytest.mark.asyncio
async def test_run_success_with_async_function(self, trigger,
mock_import_string):
"""Test trigger handles async functions correctly."""
diff --git
a/providers/common/messaging/src/airflow/providers/common/messaging/triggers/msg_queue.py
b/providers/common/messaging/src/airflow/providers/common/messaging/triggers/msg_queue.py
index 5dd56333d25..d69407e888c 100644
---
a/providers/common/messaging/src/airflow/providers/common/messaging/triggers/msg_queue.py
+++
b/providers/common/messaging/src/airflow/providers/common/messaging/triggers/msg_queue.py
@@ -52,16 +52,31 @@ class MessageQueueTrigger(BaseEventTrigger):
:param scheme: The queue scheme (e.g., 'kafka', 'redis+pubsub', 'sqs').
Used for provider matching.
:param queue: **Deprecated** The queue identifier (URI format). If
provided, this takes precedence over scheme parameter.
This parameter is deprecated and will be removed in future versions.
Use the 'scheme' parameter instead.
+ :param trigger_queue: Assign this trigger to a specific trigger queue (see
+ :ref:`config:triggerer__queues_enabled` and the ``--queues`` option of
``airflow triggerer``).
+ Named differently from the ``queue`` parameter above, which is the
deprecated broker queue URI
+ and cannot be repurposed for triggerer routing without breaking
existing callers.
.. seealso::
For more information on how to use this trigger, take a look at the
guide:
:ref:`howto/trigger:MessageQueueTrigger`
"""
- queue: str | None = None
+ queue_uri: str | None = None
scheme: str | None = None
- def __init__(self, *, queue: str | None = None, scheme: str | None = None,
**kwargs: Any) -> None:
+ def __init__(
+ self,
+ *,
+ queue: str | None = None,
+ scheme: str | None = None,
+ trigger_queue: str | None = None,
+ **kwargs: Any,
+ ) -> None:
+ # Stored under our own name rather than through
`BaseEventTrigger.__init__(queue=...)` so this
+ # works regardless of the installed airflow-core version, see `queue`
property below.
+ self._trigger_queue = trigger_queue
+
if queue is None and scheme is None:
raise ValueError("Either `queue` or `scheme` parameter must be
provided.")
@@ -73,14 +88,22 @@ class MessageQueueTrigger(BaseEventTrigger):
AirflowProviderDeprecationWarning,
stacklevel=2,
)
- self.queue = queue
+ self.queue_uri = queue
self.scheme = None
else:
- self.queue = None
+ self.queue_uri = None
self.scheme = scheme
self.kwargs = kwargs
+ @property
+ def queue(self) -> str | None:
+ return self._trigger_queue
+
+ @queue.setter
+ def queue(self, value: str | None) -> None:
+ self._trigger_queue = value
+
@cached_property
def trigger(self) -> BaseEventTrigger:
if len(MESSAGE_QUEUE_PROVIDERS) == 0:
@@ -91,12 +114,12 @@ class MessageQueueTrigger(BaseEventTrigger):
raise ValueError("No message queue providers are available. ")
# Find matching providers based on queue URI or scheme
- if self.queue is not None:
+ if self.queue_uri is not None:
# Use existing queue-based matching for backward compatibility
providers = [
- provider for provider in MESSAGE_QUEUE_PROVIDERS if
provider.queue_matches(self.queue)
+ provider for provider in MESSAGE_QUEUE_PROVIDERS if
provider.queue_matches(self.queue_uri)
]
- identifier = self.queue
+ identifier = self.queue_uri
match_by = "queue"
elif self.scheme is not None:
# Use new scheme-based matching
@@ -131,9 +154,9 @@ class MessageQueueTrigger(BaseEventTrigger):
# Create trigger instance
selected_provider = providers[0]
- if self.queue is not None:
+ if self.queue_uri is not None:
# Pass queue to trigger_kwargs for backward compatibility
- trigger_kwargs = selected_provider.trigger_kwargs(self.queue,
**self.kwargs)
+ trigger_kwargs = selected_provider.trigger_kwargs(self.queue_uri,
**self.kwargs)
return selected_provider.trigger_class()(**trigger_kwargs,
**self.kwargs)
# For scheme-based matching, we need to pass all current kwargs to the
trigger
return selected_provider.trigger_class()(**self.kwargs)
diff --git
a/providers/common/messaging/tests/unit/common/messaging/triggers/test_msg_queue.py
b/providers/common/messaging/tests/unit/common/messaging/triggers/test_msg_queue.py
index b81bd6469d4..7a84be4a94c 100644
---
a/providers/common/messaging/tests/unit/common/messaging/triggers/test_msg_queue.py
+++
b/providers/common/messaging/tests/unit/common/messaging/triggers/test_msg_queue.py
@@ -216,15 +216,40 @@ class TestMessageQueueTriggerScheme:
def test_queue_takes_precedence_over_scheme(self):
"""Test that queue parameter takes precedence when both are
provided."""
trigger = MessageQueueTrigger(queue=PROVIDER_1_QUEUE,
scheme=PROVIDER_2_SCHEME)
- assert trigger.queue == PROVIDER_1_QUEUE
+ assert trigger.queue_uri == PROVIDER_1_QUEUE
assert trigger.scheme is None
def test_scheme_only_initialization(self):
"""Test initialization with scheme parameter only."""
trigger = MessageQueueTrigger(scheme=PROVIDER_2_SCHEME)
- assert trigger.queue is None
+ assert trigger.queue_uri is None
assert trigger.scheme == PROVIDER_2_SCHEME
+ @pytest.mark.usefixtures("collect_queue_param_deprecation_warning")
+ def test_deprecated_queue_param_does_not_set_trigger_queue(self):
+ """Regression test: the deprecated `queue` (broker URI) must not leak
into the unrelated
+ `queue` property used for trigger queue assignment (see #71346), or
the resulting Trigger
+ row would never be picked up by any triggerer."""
+ trigger = MessageQueueTrigger(queue=PROVIDER_1_QUEUE)
+ assert trigger.queue_uri == PROVIDER_1_QUEUE
+ assert trigger.queue is None
+
+ def test_scheme_param_does_not_set_trigger_queue(self):
+ trigger = MessageQueueTrigger(scheme=PROVIDER_2_SCHEME)
+ assert trigger.queue is None
+
+ @pytest.mark.usefixtures("collect_queue_param_deprecation_warning")
+ def
test_trigger_queue_param_sets_trigger_queue_alongside_deprecated_queue(self):
+ """The deprecated `queue` (broker URI) parameter claims the `queue`
keyword, so trigger
+ queue routing must go through the distinct `trigger_queue` parameter
instead."""
+ trigger = MessageQueueTrigger(queue=PROVIDER_1_QUEUE,
trigger_queue="my-trigger-queue")
+ assert trigger.queue_uri == PROVIDER_1_QUEUE
+ assert trigger.queue == "my-trigger-queue"
+
+ def test_trigger_queue_param_sets_trigger_queue_alongside_scheme(self):
+ trigger = MessageQueueTrigger(scheme=PROVIDER_2_SCHEME,
trigger_queue="my-trigger-queue")
+ assert trigger.queue == "my-trigger-queue"
+
def test_scheme_provider_matching(self):
"""Test that scheme matching works correctly."""
provider1 = MockProvider(PROVIDER_1_NAME, PROVIDER_1_PATTERN,
scheme=PROVIDER_1_SCHEME)
diff --git a/providers/ibm/mq/tests/unit/ibm/mq/queues/test_mq.py
b/providers/ibm/mq/tests/unit/ibm/mq/queues/test_mq.py
index 34082ead06d..85a938bbe57 100644
--- a/providers/ibm/mq/tests/unit/ibm/mq/queues/test_mq.py
+++ b/providers/ibm/mq/tests/unit/ibm/mq/queues/test_mq.py
@@ -241,7 +241,8 @@ class TestIBMMQMessageQueueProvider:
trigger =
MessageQueueTrigger(queue="ibmmq://mq_default/MY.QUEUE.NAME", open_options=32)
assert trigger.scheme is None
- assert trigger.queue == "ibmmq://mq_default/MY.QUEUE.NAME"
+ assert trigger.queue_uri == "ibmmq://mq_default/MY.QUEUE.NAME"
+ assert trigger.queue is None
assert isinstance(trigger.trigger, AwaitMessageTrigger)
assert trigger.trigger.mq_conn_id == "mq_default"
assert trigger.trigger.queue_name == "MY.QUEUE.NAME"
diff --git a/task-sdk/src/airflow/sdk/definitions/callback.py
b/task-sdk/src/airflow/sdk/definitions/callback.py
index 355294303ab..f12e6343d81 100644
--- a/task-sdk/src/airflow/sdk/definitions/callback.py
+++ b/task-sdk/src/airflow/sdk/definitions/callback.py
@@ -151,16 +151,32 @@ class AsyncCallback(Callback):
triggerer.
It will be called with Airflow context and specified kwargs when a
deadline is missed.
+
+ Pass ``queue`` to assign the resulting trigger to a specific trigger queue
(see
+ :ref:`config:triggerer__queues_enabled` and the ``--queues`` option of
``airflow triggerer``).
"""
- def __init__(self, callback_callable: Callable | str, kwargs: dict | None
= None):
+ queue: str | None
+
+ def __init__(
+ self,
+ callback_callable: Callable | str,
+ kwargs: dict | None = None,
+ *,
+ queue: str | None = None,
+ ):
super().__init__(callback_callable=callback_callable, kwargs=kwargs)
+ self.queue = queue
@classmethod
def verify_callable(cls, callback: Callable):
if not (inspect.iscoroutinefunction(callback) or hasattr(callback,
"__await__")):
raise AttributeError(f"Provided callback {callback} is not
awaitable.")
+ @classmethod
+ def serialized_fields(cls) -> tuple[str, ...]:
+ return super().serialized_fields() + ("queue",)
+
class SyncCallback(Callback):
"""
diff --git a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
index 1ce848128d6..eeb27880622 100644
--- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
+++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
@@ -1478,10 +1478,10 @@ def _defer_task(
log.info("Pausing task as DEFERRED. ", dag_id=ti.dag_id,
task_id=ti.task_id, run_id=ti.run_id)
classpath, trigger_kwargs = defer.trigger.serialize()
queue: str | None = None
- # Currently, only task-associated BaseTrigger instances may have a
non-None queue,
- # and only when triggerer.queues_enabled conf is True.
+ # Only inherit the deferring task's queue when triggerer.queues_enabled
conf is True
+ # and the trigger doesn't already manage its own queue (e.g.
BaseEventTrigger, CallbackTrigger).
if conf.getboolean("triggerer", "queues_enabled", fallback=False) and
getattr(
- defer.trigger, "supports_triggerer_queue", True
+ defer.trigger, "trigger_queue_inherited_from_task", True
):
queue = ti.task.queue
diff --git a/task-sdk/tests/task_sdk/definitions/test_callback.py
b/task-sdk/tests/task_sdk/definitions/test_callback.py
index cab8f3b28ad..a900732090e 100644
--- a/task-sdk/tests/task_sdk/definitions/test_callback.py
+++ b/task-sdk/tests/task_sdk/definitions/test_callback.py
@@ -240,12 +240,28 @@ class TestAsyncCallback:
with pytest.raises(AttributeError, match="is not awaitable."):
AsyncCallback(empty_sync_callback_for_deadline_tests)
+ @pytest.mark.parametrize(
+ "queue",
+ [pytest.param("custom-queue", id="with_queue"), pytest.param(None,
id="without_queue")],
+ )
+ def test_init_queue(self, queue):
+ callback = AsyncCallback(TEST_CALLBACK_PATH,
kwargs=TEST_CALLBACK_KWARGS, queue=queue)
+ assert callback.queue == queue
+
def test_serialize_deserialize(self):
callback = AsyncCallback(TEST_CALLBACK_PATH,
kwargs=TEST_CALLBACK_KWARGS)
serialized = serialize(callback)
deserialized = cast("Callback", deserialize(serialized.copy()))
assert callback == deserialized
+ def test_serialize_deserialize_round_trip_keeps_queue(self):
+ callback = AsyncCallback(TEST_CALLBACK_PATH,
kwargs=TEST_CALLBACK_KWARGS, queue="custom-queue")
+
+ deserialized = cast("AsyncCallback", deserialize(serialize(callback)))
+
+ assert deserialized == callback
+ assert deserialized.queue == "custom-queue"
+
class TestSyncCallback:
@pytest.mark.parametrize(
