**Apache Airflow version**: 2.1.2


**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): 1.19.8

**Environment**:

- **Cloud provider or hardware configuration**:
- **OS** (e.g. from /etc/os-release): Amazon Linux EKS
- **Kernel** (e.g. `uname -a`): 
- **Install tools**: Helm (community chart)
- **Others**:

**What happened**:

With the addition of `__new__` to the `BaseOperatorMeta` class, this breaks our usage of operators that allow configuration through `__init_subclass__` arguments.

Relevant python bug: https://bugs.python.org/issue29581

**What you expected to happen**:

We should be able to use `__init_subclass__` in operators as we used to be able to. We relied on this behavior to customize some of our subclasses. This is working in 2.0.2 (though might exactly be considered a regression).

**How to reproduce it**:

This is a broken example

```py3
import datetime

from airflow import DAG
from airflow.operators.bash import BashOperator


class PrefixedBashOperator(BashOperator):
    def __init_subclass__(cls, command: str = None, **kwargs):
        if command is not None:
            cls._command = command

        super().__init_subclass__(**kwargs)

    def __init__(self, bash_command, **kwargs):
        super().__init__(bash_command=self._command + ' ' + bash_command, **kwargs)


class EchoOperator(PrefixedBashOperator, command='echo'):
    pass


with DAG(dag_id='foo', start_date=datetime.datetime(2021, 7, 1)) as dag:
    EchoOperator(task_id='foo', bash_command='-e "from airflow"', dag=dag)
```

This results in error:

```
TypeError: __new__() got an unexpected keyword argument 'command'
```

**Anything else we need to know**:

This example works. This shows that all that is needed to fix the issue is add `**kwargs` to `__new__`

```py3
import datetime
from abc import ABCMeta

from airflow import DAG
from airflow.models.baseoperator import BaseOperatorMeta
from airflow.operators.bash import BashOperator


class NewBaseOperatorMeta(BaseOperatorMeta):
    def __new__(cls, name, bases, namespace, **kwargs):
        new_cls = super(ABCMeta, cls).__new__(cls, name, bases, namespace, **kwargs)
        new_cls.__init__ = cls._apply_defaults(new_cls.__init__)  # type: ignore
        return new_cls


class PrefixedBashOperator(BashOperator, metaclass=NewBaseOperatorMeta):
    def __init_subclass__(cls, command: str = None, **kwargs):
        if command is not None:
            cls._command = command

        super().__init_subclass__(**kwargs)

    def __init__(self, bash_command, **kwargs):
        super().__init__(bash_command=self._command + ' ' + bash_command, **kwargs)


class EchoOperator(PrefixedBashOperator, command='echo'):
    pass


with DAG(dag_id='foo', start_date=datetime.datetime(2021, 7, 1)) as dag:
    EchoOperator(task_id='foo', bash_command='-e "from airflow"', dag=dag)
```


airflow/models/baseoperator.py
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):
                self.max_retry_delay = max_retry_delay
            else:
                self.log.debug("max_retry_delay isn't a timedelta object, assuming secs")
                self.max_retry_delay = timedelta(seconds=max_retry_delay)
        self.params = params or {}  
        if priority_weight is not None and not isinstance(priority_weight, int):
            raise AirflowException(
                f"`priority_weight` for task '{self.task_id}' only accepts integers, "
                f"received '{type(priority_weight)}'."
            )