ExternalTaskMarker not clearing downstream tasks in Airflow 3.1.6

[LinasData]

I am upgrading from Airflow 2.11.0 to Airflow 3.1.6 and noticed a change in behavior regarding the ExternalTaskMarker or maybe I am just using Airflow 3.1.6 wrong, so any help counts.

In Airflow 2.11.0, when I cleared a task upstream of an ExternalTaskMarker (and selected the "Recursive" or "Downstream" clearing options), the corresponding ExternalTaskSensor in the child DAG would also be cleared. In Airflow 3, this relationship seems broken; clearing the parent task does not trigger a clear on the child task.

We want to maintain task-level cross-DAG dependencies. While we are aware of the new Assets feature in Airflow 3, we specifically need the granular control provided by ExternalTaskMarker to ensure that re-running a specific parent task forces the child sensor to reset.

Environment:
Airflow Version: 3.0.x
Executor: CeleryExecutor
Python Version: 3.12
Ran on docker container setup locally

Minimal working example:

# --- PARENT DAG ---
from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.external_task import ExternalTaskMarker

with DAG(
    dag_id='parent_dag',
    start_date=datetime(2026, 1, 18),
    schedule='@daily',
    catchup=False
) as parent_dag:

    task_1 = BashOperator(
        task_id='echo_hello',
        bash_command='echo HELLO!!!!'
    )

    # Marker to point to child_dag/receive_call
    parent_trigger = ExternalTaskMarker(
        task_id='parent_trigger',
        external_dag_id='child_dag',
        external_task_id='receive_call'
    )

    task_1 >> parent_trigger

# --- CHILD DAG ---
from airflow.sensors.external_task import ExternalTaskSensor

with DAG(
    dag_id='child_dag',
    start_date=datetime(2026, 1, 18),
    schedule='@daily',
    catchup=False
) as child_dag:

    receive_call = ExternalTaskSensor(
        task_id='receive_call',
        external_dag_id='parent_dag',
        external_task_id='parent_trigger'
    )

    child_task = BashOperator(
        task_id='child_echo',
        bash_command='echo DONE'
    )

    receive_call >> child_task