Extra Links do not works with mapped operators #25360
Description
Apache Airflow version
main (development)
What happened
I found that Extra Links do not work with dynamic tasks at all - links inaccessible, but same Extra Links works fine with not mapped operators.
I think the nature of that extra links assign to parent task instance (i do not know how to correct name this TI) but not to actual mapped TIs.
As result we only have number extra links defined in operator not (number extra links defined in operator) x number of mapped TIs.
What you think should happen instead
No response
How to reproduce
from pendulum import datetime
from airflow.decorators import dag
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.operators.empty import EmptyOperator
EXTERNAL_DAG_IDS = [f"example_external_dag_{ix:02d}" for ix in range(3)]
DAG_KWARGS = {
"start_date": datetime(2022, 7, 1),
"schedule_interval": "@daily",
"catchup": False,
"tags": ["mapped_extra_links", "AIP-42", "serialization"],
}
def external_dags():
EmptyOperator(task_id="dummy")
@dag(**DAG_KWARGS)
def external_regular_task_sensor():
for external_dag_id in EXTERNAL_DAG_IDS:
ExternalTaskSensor(
task_id=f'wait_for_{external_dag_id}',
external_dag_id=external_dag_id,
poke_interval=5,
)
@dag(**DAG_KWARGS)
def external_mapped_task_sensor():
ExternalTaskSensor.partial(
task_id='wait',
poke_interval=5,
).expand(external_dag_id=EXTERNAL_DAG_IDS)
dag_external_regular_task_sensor = external_regular_task_sensor()
dag_external_mapped_task_sensor = external_mapped_task_sensor()
for dag_id in EXTERNAL_DAG_IDS:
globals()[dag_id] = dag(dag_id=dag_id, **DAG_KWARGS)(external_dags)()mapped_extra_links.mp4
Operating System
macOs 12.5
Versions of Apache Airflow Providers
No response
Deployment
Other
Deployment details
No response
Anything else
No response
Are you willing to submit PR?
- Yes I am willing to submit a PR!
Code of Conduct
- I agree to follow this project's Code of Conduct