Add trigger queue support for AsyncCallback and BaseEventTrigger - #71346
Conversation
boring-cyborg
Bot
added
area:DAG-processing
area:deadline-alerts
1ca1494 to
bb21aa1
Compare
zach-overflow
changed the title
queue support for AsyncCallback and BaseEventTriggerqueue support for AsyncCallback and BaseEventTrigger
d7f6479 to
f297fc0
Compare
b169e5a to
24bde08
Compare
549f90a to
a52b817
Compare
|
@uranusjr Thanks for the review! I believe I've addressed everything, but please let me know if you have any other feedback. |
a52b817 to
3a3c2a7
Compare
uranusjr
left a comment
uranusjr
left a comment
There was a problem hiding this comment.
What do we do with the argument collision? As this stands, the queue cannot actually be set by the user, can it?
@uranusjr Fair point, I think I was a bit overzealous when I saw that the Users can then assign specifically the |
|
cc @uranusjr is this good to merge? Should I get any other reviewers eyes on this? Please let me know if there's anyone in particular I might want to tag here if so, and I'll be happy to do so 👍 |
hussein-awala
left a comment
hussein-awala
left a comment
There was a problem hiding this comment.
LGTM once the newsfragment is removed
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.
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.
…gger.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.
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.
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.
…mpat tests" This reverts commit 1993585.
…ion, 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.
f8e6b94 to
122a949
Compare
4 participants
What?
BaseEventTriggerand theAsyncCallbackcategories of triggers.Why?
3.2.0, but only for task-based triggers. This was because at the time, theBaseEventTriggerandAsyncCallbackpatterns were in the middle of development.Was generative AI tooling used to co-author this PR?