its-amann commented on code in PR #77046:
URL: https://github.com/apache/airflow/pull/77046#discussion_r4572205126
##########
providers/common/messaging/src/airflow/providers/common/messaging/triggers/msg_queue.py:
##########
@@ -68,20 +70,42 @@ class MessageQueueTrigger(BaseEventTrigger):
def __init__(
self,
*,
- queue: str | None = None,
scheme: str | None = None,
+ queue: str | None = None,
trigger_queue: str | None = None,
+ queue_uri: 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 trigger_queue is not None:
+ warnings.warn(
+ "The `trigger_queue` parameter is deprecated and will be
removed in a future version. "
+ "Use `queue` instead to assign this trigger to a trigger
queue.",
+ AirflowProviderDeprecationWarning,
+ stacklevel=2,
+ )
- if queue is None and scheme is None:
- raise ValueError("Either `queue` or `scheme` parameter must be
provided.")
+ if queue_uri is not None:
+ warnings.warn(
+ "The `queue_uri` parameter is deprecated and will be removed
in future versions. "
+ "Use the `scheme` parameter instead and pass configuration as
keyword arguments to `MessageQueueTrigger`.",
+ AirflowProviderDeprecationWarning,
+ stacklevel=2,
+ )
- # For backward compatibility, queue takes precedence
- if queue is not None:
+ # Determine whether `queue` represents the trigger queue or the legacy
broker queue URI.
+ # When `scheme` or `queue_uri` is explicitly provided, `queue` is the
trigger queue.
+ # When neither is provided, `queue` is treated as the legacy broker
queue URI for backward compatibility.
+ if scheme is not None or queue_uri is not None:
Review Comment:
this flips behavior for anyone already passing both, e.g.
`MessageQueueTrigger(queue="sqs://...", scheme="kafka")` used to match on the
URI, now the URI silently becomes the trigger queue name and with queues
enabled no triggerer ever picks it up. keeping the old meaning when `queue`
looks like a URI avoids that:
```suggestion
if scheme is not None and queue_uri is None and queue is not None
and "://" in queue:
# Existing callers passing the broker URI as `queue` next to
`scheme` used to get URI
# matching; keep that instead of routing to a trigger queue
named after the URI.
warnings.warn(
"Passing the broker queue URI as `queue` is deprecated, use
`scheme` instead.",
AirflowProviderDeprecationWarning,
stacklevel=2,
)
queue_uri, queue = queue, None
if scheme is not None or queue_uri is not None:
```
fixed this plus a test in this commit, you can just cherry pick it instead
of redoing it:
https://github.com/its-amann/airflow/commit/3c8617366911f59e9b1c4ce848a7b5236109603b
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
