**Apache Airflow version**:

2.0.1

**What happened**:

When clearing task across dags using `ExternalTaskMarker` the dag state of the external `DagRun` is not set to active. So cleared tasks in the external dag will not automatically start if the `DagRun` is a Failed or Succeeded state.

**What you expected to happen**:

The external `DagRun` run should also be set to Running state.

**How to reproduce it**:

Clear tasks in an external dag using an ExternalTaskMarker.


**Anything else we need to know**:

Looking at the code is has:

https://github.com/apache/airflow/blob/b23fc137812f5eabf7834e07e032915e2a504c17/airflow/models/dag.py#L1323-L1335

It seems like it intentionally calls the dag member method `set_dag_run_state` instead of letting the helper function `clear_task_insntances` set the `DagRun` state. But the member method will only change the state of `DagRun`s of dag where the original task is, while I believe `clear_task_instances` would correctly change the state of all involved `DagRun`s.

tests/sensors/test_external_task_sensor.py
task_a_2 = ExternalTaskSensor(
        task_id="task_a_2", external_dag_id=dag_1.dag_id, external_task_id=task_b_1.task_id, dag=dag_2
    )
    task_b_2 = ExternalTaskMarker(
        task_id="task_b_2", external_dag_id="dag_3", external_task_id="task_a_3", recursion_depth=1, dag=dag_2
    )
    task_a_2 >> task_b_2
    dag_3 = DAG("dag_3", start_date=DEFAULT_DATE, schedule_interval=None)
    task_a_3 = ExternalTaskSensor(
        task_id="task_a_3", external_dag_id=dag_2.dag_id, external_task_id=task_b_2.task_id, dag=dag_3
    )
    task_b_3 = DummyOperator(task_id="task_b_3", dag=dag_3)
    task_a_3 >> task_b_3
    for dag in [dag_0, dag_1, dag_2, dag_3]:
        dag_bag.bag_dag(dag=dag, root_dag=dag)
    return dag_bag
def run_tasks(dag_bag, execution_date=DEFAULT_DATE):
    """
    Run all tasks in the DAGs in the given dag_bag. Return the TaskInstance objects as a dict
    keyed by task_id.
    """
    tis = {}
    for dag in dag_bag.dags.values():
        for task in dag.tasks:
            ti = TaskInstance(task=task, execution_date=execution_date)
            tis[task.task_id] = ti
            ti.run()
            assert_ti_state_equal(ti, State.SUCCESS)
    return tis
def assert_ti_state_equal(task_instance, state):