This is an automated email from the ASF dual-hosted git repository.
vincbeck 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 2a835d24768 Enforce the message queue provider contract without
breaking scheme-only providers (#73168)
2a835d24768 is described below
commit 2a835d247687322ee01a0dcdd004dceef04122c2
Author: Keith <[email protected]>
AuthorDate: Tue Sep 15 21:38:55 2026 +0900
Enforce the message queue provider contract without breaking scheme-only
providers (#73168)
BaseMessageQueueProvider declares abstract methods but never inherited
ABC, so an incomplete subclass instantiates silently and its unimplemented
queue_matches returns None, making the provider silently unmatched in
MessageQueueTrigger dispatch instead of failing loudly. Three in-tree
providers (google pubsub, redis, azure service bus) rely on that gap by
implementing only trigger_class for scheme-based dispatch.
Align the contract with reality: inherit ABC, keep trigger_class as the
only abstract method, and give queue_matches / trigger_kwargs concrete
defaults documented as the scheme-only baseline. Missing trigger_class
now fails at instantiation; no provider needs changes.
---
.../common/messaging/providers/base_provider.py | 25 +++++++---
.../messaging/providers/test_base_provider.py | 57 ++++++++++++++++------
2 files changed, 60 insertions(+), 22 deletions(-)
diff --git
a/providers/common/messaging/src/airflow/providers/common/messaging/providers/base_provider.py
b/providers/common/messaging/src/airflow/providers/common/messaging/providers/base_provider.py
index 505e1b647ec..d707bfbcce2 100644
---
a/providers/common/messaging/src/airflow/providers/common/messaging/providers/base_provider.py
+++
b/providers/common/messaging/src/airflow/providers/common/messaging/providers/base_provider.py
@@ -16,19 +16,24 @@
# under the License.
from __future__ import annotations
-from abc import abstractmethod
+from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from airflow.triggers.base import BaseEventTrigger
-class BaseMessageQueueProvider:
+class BaseMessageQueueProvider(ABC):
"""
Base class defining a provider supported by operators/triggers of
common-messaging provider.
To add a new provider supported by the provider, create a new class
extending this base class and add it
to ``MESSAGE_QUEUE_PROVIDERS``.
+
+ Providers can support two dispatch paths: scheme-based matching (set
``scheme``) and
+ queue-URI-based matching (override ``queue_matches``, and
``trigger_kwargs`` when the
+ trigger needs parameters derived from the queue URI). ``trigger_class`` is
required in
+ both cases.
"""
scheme: str | None = None
@@ -44,25 +49,31 @@ class BaseMessageQueueProvider:
"""
return self.scheme == scheme
- @abstractmethod
def queue_matches(self, queue: str) -> bool:
"""
Return whether a given queue (string) matches a specific provider's
pattern.
- This function must be as specific as possible to avoid collision with
other providers.
- Functions in this provider should NOT overlap with each other in their
matching criteria.
+ Providers that only support scheme-based dispatch keep this default,
which matches
+ nothing. Override it to support queue-URI-based dispatch; the
implementation must be
+ as specific as possible to avoid collision with other providers.
Functions in this
+ provider should NOT overlap with each other in their matching criteria.
:param queue: The queue identifier
"""
+ return False
@abstractmethod
def trigger_class(self) -> type[BaseEventTrigger]:
"""Trigger class to use when ``queue_matches`` returns True."""
- @abstractmethod
def trigger_kwargs(self, queue: str, **kwargs) -> dict:
"""
- Parameters passed to the instance of ``trigger_class``.
+ Parameters passed to the instance of ``trigger_class`` on
queue-URI-based dispatch.
+
+ Providers that only support scheme-based dispatch keep this default.
Override it
+ together with ``queue_matches`` when the trigger needs parameters
derived from the
+ queue URI.
:param queue: The queue identifier
"""
+ return {}
diff --git
a/providers/common/messaging/tests/unit/common/messaging/providers/test_base_provider.py
b/providers/common/messaging/tests/unit/common/messaging/providers/test_base_provider.py
index df5741da73a..047e21f783c 100644
---
a/providers/common/messaging/tests/unit/common/messaging/providers/test_base_provider.py
+++
b/providers/common/messaging/tests/unit/common/messaging/providers/test_base_provider.py
@@ -22,7 +22,7 @@ from
airflow.providers.common.messaging.providers.base_provider import BaseMessa
class KafkaLikeProvider(BaseMessageQueueProvider):
- """Minimal complete provider used to exercise the base-class contract."""
+ """Provider overriding the full queue-URI dispatch surface."""
scheme = "kafka"
@@ -33,7 +33,34 @@ class KafkaLikeProvider(BaseMessageQueueProvider):
raise NotImplementedError
def trigger_kwargs(self, queue: str, **kwargs) -> dict:
- return {}
+ return {"topic": queue}
+
+
+class SchemeOnlyProvider(BaseMessageQueueProvider):
+ """Minimal provider shape used by scheme-based dispatch (only
trigger_class implemented)."""
+
+ scheme = "scheme-only"
+
+ def trigger_class(self):
+ raise NotImplementedError
+
+
+class TestContractEnforcement:
+ def test_trigger_class_is_the_only_abstract_method(self):
+ assert BaseMessageQueueProvider.trigger_class.__isabstractmethod__ is
True
+ assert getattr(BaseMessageQueueProvider.queue_matches,
"__isabstractmethod__", False) is False
+ assert getattr(BaseMessageQueueProvider.trigger_kwargs,
"__isabstractmethod__", False) is False
+ assert getattr(BaseMessageQueueProvider.scheme_matches,
"__isabstractmethod__", False) is False
+
+ def test_subclass_without_trigger_class_fails_loudly(self):
+ class IncompleteProvider(BaseMessageQueueProvider):
+ scheme = "incomplete"
+
+ with pytest.raises(TypeError, match="trigger_class"):
+ IncompleteProvider()
+
+ def test_scheme_only_provider_is_instantiable(self):
+ assert SchemeOnlyProvider().scheme == "scheme-only"
class TestSchemeMatches:
@@ -52,20 +79,20 @@ class TestSchemeMatches:
def test_base_class_scheme_defaults_to_none_and_matches_nothing(self):
assert BaseMessageQueueProvider.scheme is None
- assert KafkaLikeProvider.scheme_matches(BaseMessageQueueProvider(),
"kafka") is False
+ assert SchemeOnlyProvider.scheme_matches(SchemeOnlyProvider(),
"kafka") is False
+
+class TestQueueDispatchDefaults:
+ @pytest.mark.parametrize("queue", ["kafka://topic",
"redis+pubsub://channel", ""])
+ def test_default_queue_matches_matches_nothing(self, queue):
+ assert SchemeOnlyProvider().queue_matches(queue) is False
[email protected](
- "method_name",
- [
- "queue_matches",
- "trigger_class",
- "trigger_kwargs",
- ],
-)
-def test_provider_contract_methods_are_marked_abstract(method_name):
- assert getattr(BaseMessageQueueProvider, method_name).__isabstractmethod__
is True
+ def test_default_trigger_kwargs_is_empty(self):
+ assert SchemeOnlyProvider().trigger_kwargs("kafka://topic") == {}
+ def test_overriding_provider_keeps_its_own_dispatch(self):
+ provider = KafkaLikeProvider()
-def test_scheme_matches_is_part_of_the_concrete_surface():
- assert getattr(BaseMessageQueueProvider.scheme_matches,
"__isabstractmethod__", False) is False
+ assert provider.queue_matches("kafka://topic") is True
+ assert provider.queue_matches("sqs://queue") is False
+ assert provider.trigger_kwargs("kafka://topic") == {"topic":
"kafka://topic"}
