kaxil commented on code in PR #74580:
URL: https://github.com/apache/airflow/pull/74580#discussion_r4247787879
##########
providers/standard/src/airflow/providers/standard/sensors/external_task.py:
##########
@@ -75,6 +76,10 @@ class ExternalDagLink(BaseOperatorLink):
name = "External DAG"
+ @property
+ def xcom_key(self) -> str:
Review Comment:
Overriding `xcom_key` changes what this key holds. On Airflow 3,
`finalize()` in `task_runner.py` stores the finished URL from `get_link()`
under `oe.xcom_key` after every run, including the one that ends in DEFERRED,
and XComs aren't cleared when a task resumes from deferral. So for a deferrable
sensor the first finalize writes `.../dags/ext/runs/<own run_id>` to
`_external_run_id`, the second finalize reads that URL back as the run id, and
the link becomes `http://host/dags/ext/runs/http:/host/dags/ext/runs/<run_id>`.
Could you keep the default link key and store the run id under its own
constant, the way `TriggerDagRunLink` does with `XCOM_RUN_ID`?
`ExternalTaskMarker` shares this link class, so it picks up the change too.
##########
providers/standard/src/airflow/providers/standard/sensors/external_task.py:
##########
@@ -387,7 +407,10 @@ def _get_count(states: list[str]) -> int:
count = _get_count(self.allowed_states)
count_allowed = self._calculate_count(count, dttm_filter)
- return count_allowed == len(dttm_filter)
+ if count_allowed == len(dttm_filter):
+ self._push_external_run_ids_to_xcom(context, dttm_filter)
Review Comment:
With `deferrable=True` this never runs. `execute()` goes straight to
`defer()`, `WorkflowTrigger` only yields `{"status": "success"}`, and
`execute_complete` only logs on success, so a deferrable sensor with
`execution_date_fn` still links to its own run. Could the trigger event carry
the matched run id so `execute_complete` can store it?
##########
providers/standard/src/airflow/providers/standard/sensors/external_task.py:
##########
@@ -96,13 +101,15 @@ def get_link(self, operator: BaseOperator, *, ti_key:
TaskInstanceKey) -> str:
if template_fields :=
RenderedTaskInstanceFields.get_templated_fields(core_ti_key):
external_dag_id: str = template_fields.get("external_dag_id",
operator.external_dag_id) # type: ignore[no-redef]
+ external_run_id = XCom.get_value(key=self.xcom_key, ti_key=ti_key) or
ti_key.run_id
Review Comment:
This breaks the existing link tests: `test_external_task_sensor_extra_link`
and `TestExternalDagLink::test_link_points_to_external_dag_run` fail in CI with
`ImportError: cannot import name 'SUPERVISOR_COMMS'`, since `XCom.get_value`
needs supervisor comms those tests don't set up. The `TriggerDagRunLink` tests
patch `XCom.get_one` for this. Could you also add a test for the scenario from
the issue (an `execution_date_fn` that maps to a different date) and one for
deferrable mode?
##########
providers/standard/src/airflow/providers/standard/sensors/external_task.py:
##########
@@ -346,7 +353,20 @@ def poke(self, context: Context) -> bool:
if AIRFLOW_V_3_0_PLUS:
return self._poke_af3(context, dttm_filter)
- return self._poke_af2(dttm_filter)
+ return self._poke_af2(context, dttm_filter)
+
+ def _push_external_run_ids_to_xcom(
+ self, context: Context, dttm_filter: Sequence[datetime.datetime]
+ ) -> None:
+ try:
+ if dttm_filter:
+ logical_date = dttm_filter[0]
+ external_run_id = f"scheduled__{logical_date.isoformat()}"
+
+ ti = context.get("ti")
+ ti.xcom_push(key=ExternalDagLink().xcom_key,
value=external_run_id)
Review Comment:
MyPy providers fails on this line because `context.get("ti")` is typed as
optional; `context["ti"]` fixes it. This also only uses `dttm_filter[0]`, so
when `execution_date_fn` returns several dates the link goes to the first one.
The plural `_push_external_run_ids_to_xcom` name suggests otherwise, so a
singular name or a short note would help.
##########
providers/standard/src/airflow/providers/standard/sensors/external_task.py:
##########
@@ -346,7 +353,20 @@ def poke(self, context: Context) -> bool:
if AIRFLOW_V_3_0_PLUS:
return self._poke_af3(context, dttm_filter)
- return self._poke_af2(dttm_filter)
+ return self._poke_af2(context, dttm_filter)
+
+ def _push_external_run_ids_to_xcom(
+ self, context: Context, dttm_filter: Sequence[datetime.datetime]
+ ) -> None:
+ try:
+ if dttm_filter:
+ logical_date = dttm_filter[0]
+ external_run_id = f"scheduled__{logical_date.isoformat()}"
Review Comment:
A logical date doesn't determine the run id, so this can point at a run that
doesn't exist, or at the wrong one. Scheduled run ids are built from
`run_after` (`Timetable.generate_run_id`), and for interval timetables
`run_after` is the interval end while `logical_date` is the start: a daily
external run with logical date Oct 10 has run id
`scheduled__2026-10-11T00:00:00+00:00`. Backfill, manual and asset-triggered
runs never match this format either. Since this now also runs without
`execution_date_fn`, it can replace a link that was correct before. Can we
record the `run_id` of the run that actually matched instead? `_poke_af2` has a
session for a `DagRun` query, and on Airflow 3 `ti.get_task_states(...)`
returns a map keyed by run id (the task-group path already uses it).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
