**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
t2.set_downstream(x3)
    :param tasks: List of tasks or XComArgs to set dependencies
    :type tasks: List[airflow.models.BaseOperator], airflow.models.BaseOperator, List[airflow.models.XComArg],
        or XComArg
    """
    from airflow.models.xcom_arg import XComArg
    for index, up_task in enumerate(tasks[:-1]):
        down_task = tasks[index + 1]
        if isinstance(up_task, (BaseOperator, XComArg)):
            up_task.set_downstream(down_task)
            continue
        if isinstance(down_task, (BaseOperator, XComArg)):
            down_task.set_upstream(up_task)
            continue
        if not isinstance(up_task, Sequence) or not isinstance(down_task, Sequence):
            raise TypeError(
                'Chain not supported between instances of {up_type} and {down_type}'.format(
                    up_type=type(up_task), down_type=type(down_task)
                )
            )
        up_task_list = up_task
        down_task_list = down_task
        if len(up_task_list) != len(down_task_list):
            raise AirflowException(
                f'Chain not supported different length Iterable '
                f'but get {len(up_task_list)} and {len(down_task_list)}'
            )
        for up_t, down_t in zip(up_task_list, down_task_list):
            up_t.set_downstream(down_t)
def cross_downstream(