[Improvement] Allow users to override the `kwargs` used when watcher consumer tasks are retried

[tatiana]

With the introduction of the watcher mode execution mode, Cosmos tasks have different memory and CPU requirements depending on the task:

• Producer tasks: will need as much CPU and memory needed as the task will require to run the dbt command. This can be set via the setup_operator_args (documentation
• Consumer tasks by default, they behave as async sensors and require very little memory and CPU.

However, when consumer tasks are retried, they behave as in Cosmos Local Execution mode and require more CPU and memory utilisation.

The Airflow Cluster Policy could help assign consumer task retries to a more powerful node pool, as described in ticket #2148. However, we have not had a chance to try this out yet.

That said, for now, we plan to introduce a new argument, retry_operator_args, that will allow users to define properties to set in the watcher consumer tasks during retry.

We should review this interface while the feature is experimental and improve it once it's released as stable.

[tatiana]

If we are able to support templated fields in the Celery configuration, customers could pass the following to Cosmos via operator_args:

    queue="{{ 'high_mem_cpu' if ti.try_number > 1 else 'default' }}"