### Overview
Some Executors have their own first class CLI commands (now that’s hardcoding/coupling!) which setup or modify various components related to that Executor.
### Examples
- **5a**) Celery Executor commands: https://github.com/apache/airflow/blob/27e2101f6ee5567b2843cbccf1dca0b0e7c96186/airflow/cli/cli_parser.py#L1689-L1734
- **5b**) Kubernetes Executor commands: https://github.com/apache/airflow/blob/27e2101f6ee5567b2843cbccf1dca0b0e7c96186/airflow/cli/cli_parser.py#L1754-L1771
- **5c**) Default CLI parser has hardcoded logic for Celery and Kubernetes Executors specifically: https://github.com/apache/airflow/blob/27e2101f6ee5567b2843cbccf1dca0b0e7c96186/airflow/cli/cli_parser.py#L63-L99
### Proposal
Update the BaseExecutor interface with a pluggable mechanism to vend CLI `GroupCommands` and parsers. Executor subclasses would then implement these methods, if applicable, which would then be called to fetch commands and parsers from within Airflow Core cli parser code. We would then migrate the existing Executor CLI code from cli_parser to the respective Executor class.

Pseudo-code example for vending `GroupCommand`s:
```python
# Existing code in cli_parser.py
...
airflow_commands: List[CLICommand] = [
GroupCommand(
        name='dags',
        help='Manage DAGs',
        subcommands=DAGS_COMMANDS,
    ),
    ...
]
# New code to add groups vended by executor classes
executor_cls, _ = ExecutorLoader.import_executor_cls(conf.get('core', 'EXECUTOR'))
airflow_commands.append(executor_cls.get_cli_group_commands())
...
```
"""
        Clear tasks that were not yet launched, but were previously queued.
        Tasks can end up in a "Queued" state when a rescheduled/deferred operator
        comes back up for execution (with the same try_number) before the
        pod of its previous incarnation has been fully removed (we think).
        It's also possible when an executor abruptly shuts down (leaving a non-empty
        task_queue on that executor), but that scenario is handled via normal adoption.
        This method checks each of our queued tasks to see if the corresponding pod
        is around, and if not, and there's no matching entry in our own
        task_queue, marks it for re-execution.
        """
        if TYPE_CHECKING:
            assert self.kube_client
        from airflow.models.taskinstance import TaskInstance
        self.log.debug("Clearing tasks that have not been launched")
        query = session.query(TaskInstance).filter(
            TaskInstance.state == TaskInstanceState.QUEUED, TaskInstance.queued_by_job_id == self.job_id
        )
        if self.kubernetes_queue:
            query = query.filter(TaskInstance.queue == self.kubernetes_queue)
        queued_tis: list[TaskInstance] = query.all()
        self.log.info("Found %s queued task instances", len(queued_tis))
        
        allowed_age = self.kube_config.worker_pods_queued_check_interval * 3
        for key, timestamp in list(self.last_handled.items()):
            if time.time() - timestamp > allowed_age:
                del self.last_handled[key]
        for ti in queued_tis: