

**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,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        upstream: bool = False,
        downstream: bool = False,
        session: Session = None,
    ):
        """
        Clears the state of task instances associated with the task, following
        the parameters specified.
        """
        qry = session.query(TaskInstance).filter(TaskInstance.dag_id == self.dag_id)
        if start_date:
            qry = qry.filter(TaskInstance.execution_date >= start_date)
        if end_date:
            qry = qry.filter(TaskInstance.execution_date <= end_date)
        tasks = [self.task_id]
        if upstream:
            tasks += [t.task_id for t in self.get_flat_relatives(upstream=True)]
        if downstream:
            tasks += [t.task_id for t in self.get_flat_relatives(upstream=False)]
        qry = qry.filter(TaskInstance.task_id.in_(tasks))
        results = qry.all()
        count = len(results)
        clear_task_instances(results, session, dag=self.dag)
        session.commit()
        return count
    @provide_session
    def get_task_instances(