

**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
self.max_retry_delay = max_retry_delay
            else:
                self.log.debug("Max_retry_delay isn't timedelta object, assuming secs")
                self.max_retry_delay = timedelta(seconds=max_retry_delay)  
        self.params = params or {}  
        self.priority_weight = priority_weight
        if not WeightRule.is_valid(weight_rule):
            raise AirflowException(
                "The weight_rule must be one of {all_weight_rules},"
                "'{d}.{t}'; received '{tr}'.".format(
                    all_weight_rules=WeightRule.all_weight_rules,
                    d=dag.dag_id if dag else "",
                    t=task_id,
                    tr=weight_rule,
                )
            )
        self.weight_rule = weight_rule
        self.resources: Optional[Resources] = Resources(**resources) if resources else None
        self.run_as_user = run_as_user
        self.task_concurrency = task_concurrency
        self.executor_config = executor_config or {}
        self.do_xcom_push = do_xcom_push
        
        self._upstream_task_ids: Set[str] = set()
        self._downstream_task_ids: Set[str] = set()
        self._dag = None
        self.dag = dag or DagContext.get_current_dag()
        
        
        from airflow.models.dag import DAG