### Apache Airflow version

2.2.3 (latest released)

### What happened

Hello, I quite excited to new [Deferrable ("Async") Operators in AIP-40 ](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=177050929), and going to adapt this by develop new async version of ExternalTaskSensor. But while doing, I find out that Deferrable ("Async") Operators are not canceled when DAG is mark failded/success.
![Screenshot from 2021-12-30 14-07-44](https://user-images.githubusercontent.com/6848311/147730232-90a6a5eb-3b6d-4270-9090-57c65240fe19.png)
When I mark DAG as failed, `wait_task` is canceled (changed to `failed` state), but `wait_task_async` still in `deferred` state and triggerer is keep poking.
![Screenshot from 2021-12-30 14-08-06](https://user-images.githubusercontent.com/6848311/147730385-9c3e8c13-6b21-4bee-b85a-3d064ec1cde5.png)
![image](https://user-images.githubusercontent.com/6848311/147730395-124f8705-adf7-4f1c-832a-15fd3826446c.png)



### What you expected to happen

Deferrable ("Async") Operators should be canceled as sync version of operators

### How to reproduce

Testing DAG.
```python
from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.external_task import ExternalTaskSensor, ExternalTaskSensorAsync

with DAG(
    'tutorial_async_sensor',
    default_args={
        'depends_on_past': False,
        'email': ['airflow@example.com'],
        'email_on_failure': False,
        'email_on_retry': False,
        'retries': 1,
        'retry_delay': timedelta(minutes=5),
    },
    description='A simple tutorial DAG using external sensor async',
    schedule_interval=timedelta(days=1),
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=['example'],
) as dag:
    t1 = ExternalTaskSensorAsync(
        task_id='wait_task_async',
        external_dag_id="tutorial",
        external_task_id="sleep",
        execution_delta=timedelta(hours=1),
        poke_interval=5.0
    )

    t2 = ExternalTaskSensor(
        task_id='wait_task',
        external_dag_id="tutorial",
        external_task_id="sleep",
        execution_delta=timedelta(hours=1),
        poke_interval=5.0
    )

    t3 = BashOperator(
        task_id='echo',
        depends_on_past=False,
        bash_command='echo Hello world',
        retries=3,
    )
    [t1, t2] >> t3
```

for `ExternalTaskSensorAsync`, see #20583

### Operating System

Ubuntu 20.04

### Versions of Apache Airflow Providers

_No response_

### Deployment

Virtualenv installation

### Deployment details

run `airflow standalone`

### Anything else

_No response_

### Are you willing to submit PR?

- [x] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

airflow/api/common/experimental/mark_tasks.py
tasks: Iterable[BaseOperator],
    execution_date: datetime.datetime,
    upstream: bool = False,
    downstream: bool = False,
    future: bool = False,
    past: bool = False,
    state: TaskInstanceState = TaskInstanceState.SUCCESS,
    commit: bool = False,
    session=None,
):
    """
    Set the state of a task instance and if needed its relatives. Can set state
    for future tasks (calculated from execution_date) and retroactively
    for past tasks. Will verify integrity of past dag runs in order to create
    tasks that did not exist. It will not create dag runs that are missing
    on the schedule (but it will as for subdag dag runs if needed).
    :param tasks: the iterable of tasks from which to work. task.task.dag needs to be set
    :param execution_date: the execution date from which to start looking
    :param upstream: Mark all parents (upstream tasks)
    :param downstream: Mark all siblings (downstream tasks) of task_id, including SubDags
    :param future: Mark all future tasks on the interval of the dag up until
        last execution date.
    :param past: Retroactively mark all tasks starting from start_date of the DAG