Fix MessageQueueTrigger queue collision with trigger queue assignment - #41
Merged
zach-overflow merged 1 commit intoAug 10, 2026
Conversation
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.
zach-overflow
marked this pull request as ready for review
August 10, 2026 16:25
zach-overflow
added a commit
that referenced
this pull request
Aug 10, 2026
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.
zach-overflow
added a commit
that referenced
this pull request
Aug 11, 2026
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.
zach-overflow
added a commit
that referenced
this pull request
Aug 12, 2026
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.
zach-overflow
added a commit
that referenced
this pull request
Aug 19, 2026
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.
sortega
pushed a commit
that referenced
this pull request
Aug 24, 2026
apache#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 1993585. * 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 <xu.han@datadoghq.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What?
queue(broker URI) constructor param from
self.queuetoself.queue_uri, soit no longer collides with the new
BaseEventTrigger.queueattribute thisbranch introduces for triggerer queue assignment.
queue=/scheme=constructor params never populate the new triggerer-routing
.queueattribute.
Why?
MessageQueueTrigger.__init__unconditionally doesself.queue = <broker URI or None>, directly overwriting thequeueattribute this branch addsto
BaseTriggerfor triggerer-host routing.MessageQueueTriggerbacksessentially every real-world AssetWatcher integration (SQS, Kafka, Redis,
Azure Service Bus, IBM MQ, PubSub via common.messaging) — any Dag still
using the documented-but-deprecated
MessageQueueTrigger(queue="https:// sqs.../my-queue")call style would have that URI persisted intoTrigger.queue. Sinceids_for_triggerer/get_sorted_triggersfilter onqueue IS NULLfor any triggerer started without--queues(i.e. everytriggerer today, since
--queuesrequiresqueues_enabled=True), thattrigger would never be picked up by any triggerer again — silently, with
no error, regardless of whether
queues_enabledis even turned on.Testing
queue_uriattribute;2 new regression tests confirming
.queuestaysNonefor both thedeprecated
queue=and the newscheme=construction styles.breeze run pytest providers/common/messaging/tests/unit/common/messaging/triggers/test_msg_queue.py— 20 passed.Additional Notes
Scoped to MessageQueueTrigger only — confirmed via
git grepit's the onlytrigger class in the tree that assigns
self.queuedirectly.Was generative AI tooling used to co-author this PR?
queuesupport forAsyncCallbackandBaseEventTriggerapache/airflow#71346's diff and grepping the provider tree for otherself.queueassignments.