Enable triggerer queues (#59239)

* Add trigger_queues to triggerer

* Update config additions, add doc section and unit tests

* Fix default consume_trigger_queues split on commas

* Add migration script, test updates, and news fragment

* Address feedback, bugfixes

* core unit test typing fix

* Fix trigger callback trigger_queue assignment

* Fix mypy issues in google provider triggers

* Fix 0015_2_9_0_update_trigger_kwargs_type migration

* fix deprecated sqlalchemy language in test

* refactor trigger_queue in kwargs

* Add examples to doc section, clarify newsfragment

* Initial refactor to derive trigger queues from task queues

* Move trigger queue assignment to task deferral API calls, update docs + tests

* Address mypy failures in test_ecs.py

* Remove redundant changes from refactor

* Make trigger.queue column max length consistent with taskinstance.queue column max len

* Remove no longer required typing-only changes

* manual refactors to unit test conflict resolutions

Author: zach-overflow

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

@@ -468,6 +468,71 @@ According to `benchmarks <https://github.com/apache/airflow/pull/58803#pullreque
 
 You can determine a suitable value for your deployment by creating a large number of triggers (for example, by triggering a Dag with many deferrable tasks) and observing both how the load is distributed across Triggerers in your environment and how long it takes for all Triggerers to pick up the triggers.
 
+Controlling Triggerer Host Assignment Per Trigger
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+.. versionadded:: 3.2.0
+
+Under some circumstances, it may be desirable to assign a Trigger to a specific subset of ``triggerer`` hosts. Some examples of when this might be desirable are:
+
+* In a multi-tenant Airflow system where you run a distinct set of ``triggerers`` per team.
+* Running distinct sets of ``triggerers`` hosts, where each set of hosts are configured for different trigger operations (e.g. each set of triggerers may have different cloud permissions).
+
+To enable queue assignment for triggers, do the following:
+
+1. Set the :ref:`config:triggerer__queues_enabled` config value to ``true``. This will ensure when tasks defer, they pass their assigned task queue to the newly registered trigger instance.
+2. For a given ``triggerer`` host(s), add ``--queues=<comma-separated string of task queue names to consume from>`` to the Triggerers' startup command. This option ensures the triggerer will only pick up ``trigger`` instances deferred by tasks from the specified task queue(s).
+
+For example, let's say you are running two triggerer hosts (labeled "X", and "Y") with the following commands:
+
+.. code-block:: bash
+
+      # triggerer "X" startup command
+      airflow triggerer --queues=alice,bob
+      # triggerer "Y" startup command
+      airflow triggerer --queues=test_q
+
+In this scenario, triggerer "X" will exclusively run triggers deferred from tasks originating from task queues ``"alice"`` or ``"bob"``.
+Similarly, triggerer "Y" will exclusively run triggers deferred from tasks originating from task queue ``"test_q"``.
+
+Trigger Queue Assignment Caveats
+''''''''''''''''''''''''''''''''
+
+This feature is only compatible with executors which utilize the task ``queue`` concept
+(such as the :ref:`CeleryExecutor<apache-airflow-providers-celery:celery_executor:queue>`).
+
+Additionally, queue assignment is currently only compatible with the subset of triggers originating from a task's defer ``method``.
+
++------------------------------------------------------------------------------------------------------+----------------------+----------------------------------------------------------------------------------+
+|          Trigger Type                                                                                |   Supports queues?   |  Triggerer assignment when :ref:`config:triggerer__queues_enabled` is ``True``   |
++======================================================================================================+======================+==================================================================================+
+| Task-created Trigger instances                                                                       |   Yes                |  Any triggerer with the the task queue present in its ``--queues`` option        |
++------------------------------------------------------------------------------------------------------+----------------------+----------------------------------------------------------------------------------+
+| :doc:`Event-Driven Triggers<../authoring-and-scheduling/event-scheduling>`                           |   No                 |  Any triggerer running without the ``--queues`` option                           |
++------------------------------------------------------------------------------------------------------+----------------------+----------------------------------------------------------------------------------+
+| Triggers from async :doc:`Callbacks<../administration-and-deployment/logging-monitoring/callbacks>`  |   No                 |  Any triggerer running without the ``--queues`` option                           |
++------------------------------------------------------------------------------------------------------+----------------------+----------------------------------------------------------------------------------+
+
+If you use queues for task-based triggers, while **also** using event-based triggers and/or callback triggers,
+you must run one or more triggerer hosts **without** the ``--queues`` option, so the latter 2 types of triggers are still run.
+
+.. note::
+    To enable trigger queues, you must set the ``--queues`` option on one or more triggerers' startup command (these values may differ between the various triggerers).
+    If you set the ``--queue`` value of a triggerer to some value which no task queues exist for, that triggerer will never run any triggers.
+    Similarly, all ``triggerer`` instances running without the ``--queues`` option will only consume event-driven and callback-based triggers.
+
+
+The following example shows how to run HA triggerers so that all trigger types are run (assuming all tasks'
+queue values are set to either ``"team_A"`` or ``"team_B"``):
+
+.. code-block:: bash
+
+      # triggerer "A" startup command, only consumes triggers registered by tasks in queue "team_A"
+      airflow triggerer --queues=team_A
+      # triggerer "B" startup command, only consumes triggers registered by tasks in queue "team_B"
+      airflow triggerer --queues=team_B
+      # triggerer "C" startup command, consumes only event-based triggers and callback-based triggers.
+      airflow triggerer
 
 
 Difference between Mode='reschedule' and Deferrable=True in Sensors

airflow-core/docs/img/airflow_erd.sha256

@@ -1 +1 @@
-1c79db933fe961f2d23605e1dcf73125923ea39bbd719c800e8adc1aa5bedb52
+26217aaa3d4efa7c2b9cc4687e713024649729e0d23ce36ddc4ad84754ae4f1d