zach-overflow commented on code in PR #77046:
URL: https://github.com/apache/airflow/pull/77046#discussion_r4573573701
##########
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:
+ if queue is not None and trigger_queue is not None:
+ raise ValueError("Cannot provide both `queue` and deprecated
`trigger_queue`.")
Review Comment:
I'd recommend adding a link in this exception message to help clarify why
this is the case to those unfamiliar with the newer trigger queues feature.
```suggestion
raise ValueError("Cannot provide both `queue` and deprecated
`trigger_queue`. For more details on trigger queues, see:
https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/deferring.html#controlling-triggerer-host-assignment-per-trigger";)
```
##########
providers/common/messaging/src/airflow/providers/common/messaging/triggers/msg_queue.py:
##########
@@ -90,9 +114,9 @@ def __init__(
)
self.queue_uri = queue
self.scheme = None
+ self._trigger_queue = trigger_queue
else:
- self.queue_uri = None
- self.scheme = scheme
+ raise ValueError("Either `queue` or `scheme` parameter must be
provided.")
Review Comment:
Same suggestion here: might be good to clarify this message with an added
link to the docs
```suggestion
raise ValueError("Either `queue` or `scheme` parameter must be
provided. For more details on trigger queues, see:
https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/deferring.html#controlling-triggerer-host-assignment-per-trigger";)
```
--
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]
