Move SQS message queue to Amazon provider

The common.messaging abstraction should discover common messaging
queue providers using the same mechanism as we have for other core
extensions. Previously common.messaging had the optional (but not
really) dependencies to other providers, but that was not needed
and introduced unnecessary coupling.

By switching to our built-in discovery mechanism we get immediately
all the niceties of provider discovery mechanisms:

* queue is provided by the actual provider where the service or
  integration already is implemented (sqs -> amazon provider, in
  the future kafka -> kafka provider)
* queues are discovered from installed providers
* there is no coupling or imports between common.messaging and the
  providers that implement messaging, the dependency is in the other
  way - providers that implement messaging depend on common.messaging
* airflow providers queues CLI and providers core extensions
  documentation is automatically generated

Author: potiuk

airflow-core/docs/authoring-and-scheduling/connections.rst

@@ -47,5 +47,5 @@ Airflow allows to define custom connection types. This is what is described in d
 :doc:`apache-airflow-providers:index` - providers give you the capability of defining your own connections.
 The connection customization can be done by any provider, but also
 many of the providers managed by the community define custom connection types.
-The full list of all providers delivered by ``Apache Airflow community managed providers`` can be found in
+The full list of all connections delivered by ``Apache Airflow community managed providers`` can be found in
 :doc:`apache-airflow-providers:core-extensions/connections`.

airflow-core/docs/core-concepts/index.rst

@@ -43,6 +43,7 @@ Here you can find detailed documentation about each one of the core concepts of
     auth-manager/index
     objectstorage
     backfill
+    message-queues
 
 **Communication**
 

airflow-core/docs/core-concepts/message-queues.rst

@@ -0,0 +1,41 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+.. _concepts:message-queues:
+
+Message Queues
+==============
+
+The Message Queues are a way to expose capability of external event-driven scheduling of Dags.
+
+Apache Airflow is primarily designed for time-based and dependency-based scheduling of workflows. However,
+modern data architectures often require near real-time processing and the ability to react to
+events from various sources, such as message queues.
+
+Airflow has native event-driven capability to Airflow, allowing users to create workflows that can be
+triggered by external events, thus enabling more responsive data pipelines.
+
+Airflow supports poll-based event-driven scheduling, where the Triggerer can poll
+external message queues using built-in :class:`airflow.triggers.base.BaseTrigger` classes. This allows users
+to create workflows that can be triggered by external events, such as messages arriving
+in a queue or changes in a database efficiently.
+
+Airflow constantly monitors the state of an external resource and updates the asset whenever the external
+resource reaches a given state (if it does reach it). To achieve this, we leverage Airflow Triggers.
+Triggers are small, asynchronous pieces of Python code whose job is to poll an external resource state.
+
+The list of supported message queues is available in :doc:`apache-airflow-providers:core-extensions/message-queues`.

airflow-core/src/airflow/cli/cli_config.py

@@ -1597,6 +1597,12 @@ class GroupCommand(NamedTuple):
         func=lazy_load_command("airflow.cli.commands.provider_command.executors_list"),
         args=(ARG_OUTPUT, ARG_VERBOSE),
     ),
+    ActionCommand(
+        name="queues",
+        help="Get information about queues provided",
+        func=lazy_load_command("airflow.cli.commands.provider_command.queues_list"),
+        args=(ARG_OUTPUT, ARG_VERBOSE),
+    ),
     ActionCommand(
         name="notifications",
         help="Get information about notifications provided",

airflow-core/src/airflow/cli/commands/provider_command.py

@@ -220,6 +220,19 @@ def executors_list(args):
     )
 
 
+@suppress_logs_and_warning
+@providers_configuration_loaded
+def queues_list(args):
+    """List all queues at the command line."""
+    AirflowConsole().print_as(
+        data=list(ProvidersManager().queue_class_names),
+        output=args.output,
+        mapper=lambda x: {
+            "queue_class_names": x,
+        },
+    )
+
+
 @suppress_logs_and_warning
 @providers_configuration_loaded
 def config_list(args):

airflow-core/src/airflow/provider.yaml.schema.json

@@ -467,6 +467,18 @@
                 }
             }
         },
+        "queues": {
+            "type": "array",
+            "description": "Message Queues exposed by the provider",
+            "items": {
+                "name": {
+                    "type": "string"
+                },
+                "message-queue-class": {
+                    "type": "string"
+                }
+            }
+        },
         "source-date-epoch": {
             "type": "integer",
             "description": "Source date epoch - seconds since epoch (gmtime) when the release documentation was prepared. Used to generate reproducible package builds with flint.",

airflow-core/src/airflow/provider_info.schema.json

@@ -416,6 +416,18 @@
                     "description": "Class to instantiate the plugin"
                 }
             }
+        },
+        "queues": {
+            "type": "array",
+            "description": "Message Queues exposed by the provider",
+            "items": {
+                "name": {
+                    "type": "string"
+                },
+                "message-queue-class": {
+                    "type": "string"
+                }
+            }
         }
     },
     "definitions": {

airflow-core/src/airflow/providers_manager.py

@@ -416,6 +416,7 @@ def __init__(self):
         self._auth_manager_class_name_set: set[str] = set()
         self._secrets_backend_class_name_set: set[str] = set()
         self._executor_class_name_set: set[str] = set()
+        self._queue_class_name_set: set[str] = set()
         self._provider_configs: dict[str, dict[str, Any]] = {}
         self._trigger_info_set: set[TriggerInfo] = set()
         self._notification_info_set: set[NotificationInfo] = set()
@@ -533,6 +534,12 @@ def initialize_providers_executors(self):
         self.initialize_providers_list()
         self._discover_executors()
 
+    @provider_info_cache("queues")
+    def initialize_providers_queues(self):
+        """Lazy initialization of providers queue information."""
+        self.initialize_providers_list()
+        self._discover_queues()
+
     @provider_info_cache("notifications")
     def initialize_providers_notifications(self):
         """Lazy initialization of providers notifications information."""
@@ -1091,6 +1098,14 @@ def _discover_executors(self) -> None:
                     if _correctness_check(provider_package, executors_class_name, provider):
                         self._executor_class_name_set.add(executors_class_name)
 
+    def _discover_queues(self) -> None:
+        """Retrieve all queues defined in the providers."""
+        for provider_package, provider in self._provider_dict.items():
+            if provider.data.get("queues"):
+                for queue_class_name in provider.data["queues"]:
+                    if _correctness_check(provider_package, queue_class_name, provider):
+                        self._queue_class_name_set.add(queue_class_name)
+
     def _discover_config(self) -> None:
         """Retrieve all configs defined in the providers."""
         for provider_package, provider in self._provider_dict.items():
@@ -1221,6 +1236,11 @@ def executor_class_names(self) -> list[str]:
         self.initialize_providers_executors()
         return sorted(self._executor_class_name_set)
 
+    @property
+    def queue_class_names(self) -> list[str]:
+        self.initialize_providers_queues()
+        return sorted(self._queue_class_name_set)
+
     @property
     def filesystem_module_names(self) -> list[str]:
         self.initialize_providers_filesystems()
@@ -1268,9 +1288,11 @@ def _cleanup(self):
         self._auth_manager_class_name_set.clear()
         self._secrets_backend_class_name_set.clear()
         self._executor_class_name_set.clear()
+        self._queue_class_name_set.clear()
         self._provider_configs.clear()
         self._trigger_info_set.clear()
         self._notification_info_set.clear()
         self._plugins_set.clear()
+
         self._initialized = False
         self._initialization_stack_trace = None

dev/breeze/src/airflow_breeze/global_constants.py

@@ -696,7 +696,8 @@ def generate_provider_dependencies_if_needed():
     {
         "python-version": "3.9",
         "airflow-version": "3.0.0",
-        "remove-providers": "cloudant",
+        # TODO: bring back common-messaging when we bump airflow to 3.0.1
+        "remove-providers": "cloudant common.messaging",
         "run-tests": "true",
     },
 ]

devel-common/src/sphinx_exts/operators_and_hooks_ref.py

@@ -481,6 +481,17 @@ def render_content(
         )
 
 
+class QueuesDirective(BaseJinjaReferenceDirective):
+    """Generate list of queues"""
+
+    def render_content(
+        self, *, tags: set[str] | None, header_separator: str = DEFAULT_HEADER_SEPARATOR
+    ) -> str:
+        return _common_render_list_content(
+            header_separator=header_separator, resource_type="queues", template="queues.rst.jinja2"
+        )
+
+
 class DeferrableOperatorDirective(BaseJinjaReferenceDirective):
     """Generate list of deferrable operators"""
 
@@ -521,6 +532,7 @@ def setup(app):
     app.add_directive("airflow-extra-links", ExtraLinksDirective)
     app.add_directive("airflow-notifications", NotificationsDirective)
     app.add_directive("airflow-executors", ExecutorsDirective)
+    app.add_directive("airflow-queues", QueuesDirective)
     app.add_directive("airflow-deferrable-operators", DeferrableOperatorDirective)
     app.add_directive("airflow-deprecations", DeprecationsDirective)
     app.add_directive("airflow-dataset-schemes", AssetSchemeDirective)