Verify watcher sensor retries honor setup_operator_args-style queue overrides (cluster policy)

[pankajkoti]

Summary

In watcher mode, we currently rely on ExecutionConfig.setup_operator_args (PR #2133) to send the producer task to a high-resource queue while sensors stay on a lightweight queue. When sensors retry (and run _fallback_to_local_run), the retry still executes on the sensor queue. We aim to validate and document a supported approach for routing sensor retries to the producer queue using an Airflow cluster policy.

What should happen

• A concrete recipe lives in docs (and potentially tests) showing how to use task_instance_mutation_hook to flip ti.queue to the producer queue on retries (e.g., when watcher sensors run _fallback_to_local_run).
• We confirm Cosmos works with this configuration (no regressions from the hook) and that retries succeed when the producer queue is used.
• Optional: consider adding a helper in Cosmos (or a sample snippet) so users don’t have to hand-roll the hook.

[pankajkoti]

I guess potentially adding something like the below snippet in the airflow_local_settings.py could help achieve this

from airflow.models import TaskInstance

  def task_instance_mutation_hook(ti: TaskInstance) -> None:
      if <some_identifier_for_sensor_tasks> and ti.try_number > 1:
          ti.queue = "dbt_producer_task_queue"

We might need to add an identifier for the sensor tasks that can be used in the task_instance_mutation_hook