### 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())
...
```
pod: k8s.V1Pod,
        tis_to_flush_by_key: dict[TaskInstanceKey, k8s.V1Pod],
    ) -> None:
        """
        Patch existing pod so that the current KubernetesJobWatcher can monitor it via label selectors.
        :param kube_client: kubernetes client for speaking to kube API
        :param pod: V1Pod spec that we will patch with new label
        :param tis_to_flush_by_key: TIs that will be flushed if they aren't adopted
        """
        if TYPE_CHECKING:
            assert self.scheduler_job_id
        self.log.info("attempting to adopt pod %s", pod.metadata.name)
        ti_key = annotations_to_key(pod.metadata.annotations)
        if ti_key not in tis_to_flush_by_key:
            self.log.error("attempting to adopt taskinstance which was not specified by database: %s", ti_key)
            return
        new_worker_id_label = self._make_safe_label_value(self.scheduler_job_id)
        from kubernetes.client.rest import ApiException
        try:
            kube_client.patch_namespaced_pod(
                name=pod.metadata.name,
                namespace=pod.metadata.namespace,
                body={"metadata": {"labels": {"airflow-worker": new_worker_id_label}}},
            )
        except ApiException as e:
            self.log.info("Failed to adopt pod %s. Reason: %s", pod.metadata.name, e)
            return
        del tis_to_flush_by_key[ti_key]
        self.running.add(ti_key)
    def _adopt_completed_pods(self, kube_client: client.CoreV1Api) -> None: