Configure Airflow 3 deferrable operators with custom triggers to free worker slots during long I/O waits

Steps

1. Install the Airflow Triggerer component (a separate process from the scheduler and worker); without it, deferrable operators fall back to blocking execution
2. Write a custom Trigger class inheriting from BaseTrigger, implementing async def run() which yields TriggerEvent when the awaited condition is met; register it in the provider's triggers module
3. In the operator's execute() method, call self.defer(trigger=MyTrigger(...), method_name='execute_complete') to suspend the task and release the worker slot; the task state becomes 'deferred' in the UI
4. Implement execute_complete(self, context, event) on the operator to resume processing once the trigger fires; raise AirflowException on failure events
5. Set worker_concurrency on the Triggerer in your deployment config; a single Triggerer process can handle thousands of concurrent deferred tasks limited only by async I/O capacity, unlike workers which consume one slot per running task