

**Description**

We are using a 3rd party executor which has the ability to use multiple queues, however the `default_queue` is defined under the `celery` heading and is used regardless of the executor that you are using.

See: https://github.com/apache/airflow/blob/2.0.1/airflow/models/baseoperator.py#L366

It would be nice to decouple the default_queue configuration argument away from the celery section to allow other executors to utilise this functionality with less confusion.

**Are you willing to submit a PR?**

Yep! Will open a pull request shortly


**Related Issues**

Not that I'm aware of.

airflow/models/baseoperator.py
t=task_id,
                    tr=trigger_rule,
                )
            )
        self.trigger_rule = trigger_rule
        self.depends_on_past = depends_on_past
        self.wait_for_downstream = wait_for_downstream
        if wait_for_downstream:
            self.depends_on_past = True
        self.retries = retries
        self.queue = queue
        self.pool = Pool.DEFAULT_POOL_NAME if pool is None else pool
        self.pool_slots = pool_slots
        if self.pool_slots < 1:
            raise AirflowException(f"pool slots for {self.task_id} in dag {dag.dag_id} cannot be less than 1")
        self.sla = sla
        self.execution_timeout = execution_timeout
        self.on_execute_callback = on_execute_callback
        self.on_failure_callback = on_failure_callback
        self.on_success_callback = on_success_callback
        self.on_retry_callback = on_retry_callback
        if isinstance(retry_delay, timedelta):
            self.retry_delay = retry_delay
        else:
            self.log.debug("Retry_delay isn't timedelta object, assuming secs")
            self.retry_delay = timedelta(seconds=retry_delay)  
        self.retry_exponential_backoff = retry_exponential_backoff
        self.max_retry_delay = max_retry_delay
        if max_retry_delay:
            if isinstance(max_retry_delay, timedelta):