### 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
"""Returns dates of DAG execution"""
    latest_execution_date = dag.get_latest_execution_date()
    if latest_execution_date is None:
        raise ValueError(f"Received non-localized date {execution_date}")
    
    end_date = latest_execution_date if future else execution_date
    if 'start_date' in dag.default_args:
        start_date = dag.default_args['start_date']
    elif dag.start_date:
        start_date = dag.start_date
    else:
        start_date = execution_date
    start_date = execution_date if not past else start_date
    if not dag.timetable.can_run:
        
        
        dag_runs = dag.get_dagruns_between(start_date=start_date, end_date=end_date)
        dates = sorted({d.execution_date for d in dag_runs})
    elif not dag.timetable.periodic:
        dates = [start_date]
    else:
        dates = [
            info.logical_date for info in dag.iter_dagrun_infos_between(start_date, end_date, align=False)
        ]
    return dates
@provide_session
def _set_dag_run_state(dag_id, execution_date, state, session=None):