Implement Airflow deferrable operators and triggers to reduce worker slot consumption during long-running waits

Steps

1. Upgrade to Airflow 2.2+ and install the Triggerer component (a separate process that runs async event loops); configure the number of triggerer replicas based on expected concurrent deferred tasks.
2. Replace blocking sensors (e.g., S3KeySensor, BigQueryJobTriggerSensor) with their deferrable equivalents (e.g., S3KeySensorAsync from the common.io provider or provider-specific deferrable operators); set deferrable=True in supported operators.
3. When writing a custom deferrable operator, implement execute() to call self.defer(trigger=MyTrigger(...), method_name='execute_complete') and implement execute_complete(context, event) to handle the trigger result and continue execution.
4. Implement the Trigger class as an async generator that yields TriggerEvent when the awaited condition is met; use asyncio-compatible clients (aiohttp, aiobotocore) rather than blocking SDK calls inside the trigger.
5. Set poke_interval appropriately in the trigger to avoid busy-polling the external system; the trigger runs in the Triggerer process, so excessive polling still consumes resources.
6. Monitor the Triggerer health via the Airflow UI or metrics endpoint; a crashed Triggerer will cause deferred tasks to remain in the deferred state until the Triggerer restarts and reschedules them.