Check that external tasks and task groups exist on Airflow 3 in ExternalTaskSensor

providers/standard/src/airflow/providers/standard/sensors/external_task.py

@@ -43,7 +43,11 @@
 )
 from airflow.providers.standard.operators.empty import EmptyOperator
 from airflow.providers.standard.triggers.external_task import WorkflowTrigger
-from airflow.providers.standard.utils.sensor_helper import _get_count, _get_external_task_group_task_ids
+from airflow.providers.standard.utils.sensor_helper import (
+    _check_external_task_existence,
+    _get_count,
+    _get_external_task_group_task_ids,
+)
 from airflow.providers.standard.version_compat import (
     AIRFLOW_V_3_0_PLUS,
     AIRFLOW_V_3_2_PLUS,
@@ -64,6 +68,7 @@
     from sqlalchemy.orm import Session
 
     from airflow.providers.common.compat.sdk import Context, TaskInstanceKey
+    from airflow.sdk.types import RuntimeTaskInstanceProtocol
 
 
 class ExternalDagLink(BaseOperatorLink):
@@ -171,6 +176,11 @@ class ExternalTaskSensor(BaseSensorOperator):
         external_task_id is not None) or check if the DAG to wait for exists (when
         external_task_id is None), and immediately cease waiting if the external task
         or DAG does not exist (default value: False).
+        On Airflow 3 a worker has no database access, so tasks and task groups are checked
+        through the execution API against each awaited Dag run once that run exists: the API
+        reports a task or task group that the run's Dag version does not define. Until the run
+        exists the sensor keeps waiting, and so does it with an API server that does not report
+        unknown tasks. Whether the Dag itself is registered is not checked on Airflow 3.
     :param poke_interval: polling period in seconds to check for the status
     :param poll_interval: (DEPRECATED) use ``poke_interval`` instead
     :param deferrable: Run sensor in deferrable mode
@@ -351,8 +361,9 @@ def poke(self, context: Context) -> bool:
     def _poke_af3(self, context: Context, dttm_filter: Sequence[datetime.datetime]) -> bool:
         from airflow.providers.standard.utils.sensor_helper import _get_count_by_matched_states
 
-        self._has_checked_existence = True
         ti = context["ti"]
+        if self.check_existence and not self._has_checked_existence:
+            self._check_for_existence_af3(ti, dttm_filter)
 
         def _get_count(states: list[str]) -> int:
             if self.external_task_ids:
@@ -488,6 +499,7 @@ def execute(self, context: Context) -> None:
                         logical_dates=list(dttm_filter),
                         run_ids=None,
                         execution_dates=None,
+                        check_existence=self.check_existence,
                     ),
                     method_name="execute_complete",
                 )
@@ -533,6 +545,15 @@ def execute_complete(self, context: Context, event: dict[str, typing.Any] | None
             if self.soft_fail:
                 raise AirflowSkipException("External job has failed skipping.")
             raise ExternalDagFailedError("External job has failed.")
+        elif event["status"] == "not_found":
+            # A missing task or task group is a configuration error rather than a sensor failure,
+            # so like the poke path this is raised regardless of soft_fail.
+            message = event.get("message") or (
+                f"The external tasks or task group awaited in Dag {self.external_dag_id} do not exist."
+            )
+            if event.get("kind") == "task_group":
+                raise ExternalTaskGroupNotFoundError(message)
+            raise ExternalTaskNotFoundError(message)
         else:
             if self.soft_fail:
                 raise AirflowSkipException("External job has failed skipping.")
@@ -541,6 +562,32 @@ def execute_complete(self, context: Context, event: dict[str, typing.Any] | None
                 "name of executed task and Dag."
             )
 
+    def _check_for_existence_af3(
+        self, ti: RuntimeTaskInstanceProtocol, dttm_filter: Sequence[datetime.datetime]
+    ) -> None:
+        """
+        Check that the awaited tasks or task group exist, through the execution API.
+
+        A worker has no database access on Airflow 3, so unlike ``_check_for_existence`` this
+        relies on what the execution API reports: a task or task group that the Dag version of
+        an existing awaited run does not define. While a run does not exist yet nothing can be
+        concluded, so the check is repeated on later pokes until every awaited run has been seen.
+
+        :param ti: the task instance running this sensor, used to reach the execution API
+        :param dttm_filter: the logical dates of the awaited Dag runs
+        """
+        if not self.external_task_ids and not self.external_task_group_id:
+            self._has_checked_existence = True
+            return
+
+        self._has_checked_existence = _check_external_task_existence(
+            ti,
+            external_dag_id=self.external_dag_id,
+            external_task_ids=self.external_task_ids,
+            external_task_group_id=self.external_task_group_id,
+            logical_dates=list(dttm_filter),
+        )
+
     def _check_for_existence(self, session: Session) -> None:
         dag_to_wait = DagModel.get_current(self.external_dag_id, session=session)
 

providers/standard/src/airflow/providers/standard/triggers/external_task.py

@@ -25,6 +25,7 @@
 from sqlalchemy import func, select
 
 from airflow.models import DagRun
+from airflow.providers.standard.exceptions import ExternalTaskGroupNotFoundError, ExternalTaskNotFoundError
 from airflow.providers.standard.utils.sensor_helper import _get_count
 from airflow.providers.standard.version_compat import AIRFLOW_V_3_0_PLUS
 from airflow.triggers.base import BaseTrigger, TriggerEvent
@@ -51,6 +52,9 @@ class WorkflowTrigger(BaseTrigger):
     :param poke_interval: The interval (in seconds) for poking the external tasks.
     :param soft_fail: If True, the trigger will not fail the entire dag on external task failure.
     :param logical_dates: A list of logical dates for the external dag.
+    :param check_existence: If True, ask the execution API whether the external tasks or task group
+        exist in each awaited Dag run once that run exists, and fire a ``not_found`` event when it
+        reports them missing. Airflow 3 only.
     """
 
     def __init__(
@@ -66,6 +70,7 @@ def __init__(
         allowed_states: Collection[str] | None = None,
         poke_interval: float = 2.0,
         soft_fail: bool = False,
+        check_existence: bool = False,
         **kwargs,
     ):
         self.external_dag_id = external_dag_id
@@ -79,6 +84,7 @@ def __init__(
         self.soft_fail = soft_fail
         self.execution_dates = execution_dates
         self.logical_dates = logical_dates
+        self.check_existence = check_existence
         super().__init__(**kwargs)
 
     def serialize(self) -> tuple[str, dict[str, Any]]:
@@ -96,6 +102,7 @@ def serialize(self) -> tuple[str, dict[str, Any]]:
         if AIRFLOW_V_3_0_PLUS:
             data["run_ids"] = self.run_ids
             data["logical_dates"] = self.logical_dates
+            data["check_existence"] = self.check_existence
         else:
             data["execution_dates"] = self.execution_dates
 
@@ -110,7 +117,25 @@ async def run(self) -> typing.AsyncIterator[TriggerEvent]:
             get_count_func = self._get_count
             run_id_or_dates = self.execution_dates or []
 
+        # Tasks and task groups can only be verified against a Dag run that exists, so this is
+        # repeated at every poll until every awaited run has been seen.
+        existence_checked = not (
+            AIRFLOW_V_3_0_PLUS
+            and self.check_existence
+            and (self.external_task_ids or self.external_task_group_id)
+        )
+
         while True:
+            if not existence_checked:
+                try:
+                    existence_checked = await self._check_existence_af_3()
+                except ExternalTaskGroupNotFoundError as e:
+                    yield TriggerEvent({"status": "not_found", "kind": "task_group", "message": str(e)})
+                    return
+                except ExternalTaskNotFoundError as e:
+                    yield TriggerEvent({"status": "not_found", "kind": "task", "message": str(e)})
+                    return
+
             if self.failed_states:
                 failed_count = await get_count_func(self.failed_states)
                 if failed_count > 0:
@@ -130,6 +155,20 @@ async def run(self) -> typing.AsyncIterator[TriggerEvent]:
             self.log.info("Sleeping for %s seconds", self.poke_interval)
             await asyncio.sleep(self.poke_interval)
 
+    async def _check_existence_af_3(self) -> bool:
+        """Check the awaited tasks or task group against the runs that exist; True once every run was checked."""
+        from airflow.providers.standard.utils.sensor_helper import _check_external_task_existence
+        from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
+
+        return await sync_to_async(_check_external_task_existence)(
+            RuntimeTaskInstance,
+            external_dag_id=self.external_dag_id,
+            external_task_ids=self.external_task_ids,
+            external_task_group_id=self.external_task_group_id,
+            logical_dates=self.logical_dates,
+            run_ids=self.run_ids,
+        )
+
     async def _get_count_af_3(self, states: Collection[str] | None) -> int:
         from airflow.providers.standard.utils.sensor_helper import _get_count_by_matched_states
         from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance

providers/standard/src/airflow/providers/standard/utils/sensor_helper.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 from collections.abc import Collection
+from http import HTTPStatus
 from typing import TYPE_CHECKING, Any, cast
 
 from sqlalchemy import func, select, tuple_
@@ -27,6 +28,8 @@
 from airflow.utils.session import NEW_SESSION, provide_session
 
 if TYPE_CHECKING:
+    from datetime import datetime
+
     from sqlalchemy.orm import Session
     from sqlalchemy.sql import Select
 
@@ -153,3 +156,96 @@ def _get_count_by_matched_states(
             count += 1
 
     return count
+
+
+def _not_found_message(error: Any) -> str | None:
+    """Return the execution API's message when ``error`` wraps a 404 response, ``None`` for any other error."""
+    detail = error.error.detail or {}
+    if detail.get("status_code") != HTTPStatus.NOT_FOUND:
+        return None
+    # The supervisor forwards the server's JSON body under ``detail``; FastAPI nests an
+    # ``HTTPException`` detail under a ``detail`` key of its own.
+    payload = detail.get("detail")
+    while isinstance(payload, dict) and "message" not in payload and isinstance(payload.get("detail"), dict):
+        payload = payload["detail"]
+    if isinstance(payload, dict) and isinstance(payload.get("message"), str):
+        return payload["message"]
+    return str(detail.get("message") or "not found")
+
+
+def _check_external_task_existence(
+    api: Any,
+    *,
+    external_dag_id: str,
+    external_task_ids: Collection[str] | None,
+    external_task_group_id: str | None,
+    logical_dates: Collection[datetime] | None = None,
+    run_ids: Collection[str] | None = None,
+) -> bool:
+    """
+    Verify that the awaited tasks or task group exist in the awaited Dag runs, through the execution API.
+
+    The execution API answers ``task-instances/states`` with 404 when the Dag version an existing
+    run resolves to defines neither the requested tasks nor the requested task group and the run
+    has no task instance for them (apache/airflow#73086). A normal answer means they exist for
+    that run, even while their task instances have not been created yet, so nothing is inferred
+    from task-instance counts. Nothing can be concluded about a run that does not exist yet,
+    which is why the caller repeats the check until this function returns True. API servers
+    without that validation answer normally for unknown tasks, in which case the sensor keeps
+    waiting as it did before.
+
+    :param api: an object exposing ``get_dr_count`` and ``get_task_states`` the way
+        ``RuntimeTaskInstance`` does: the running task instance, or the class itself.
+    :param external_dag_id: The ID of the external Dag.
+    :param external_task_ids: The task IDs that must exist in every awaited run.
+    :param external_task_group_id: The task group ID that must exist in every awaited run.
+    :param logical_dates: Logical dates identifying the awaited runs, used when ``run_ids`` is empty.
+    :param run_ids: Run IDs identifying the awaited runs.
+    :return: True once every awaited run exists and passed the check, False while at least one
+        awaited run does not exist yet.
+    :raises ExternalTaskNotFoundError: when the API reports one of the tasks missing from an awaited run.
+    :raises ExternalTaskGroupNotFoundError: when the API reports the task group missing from an awaited run.
+    """
+    from airflow.providers.standard.exceptions import (
+        ExternalTaskGroupNotFoundError,
+        ExternalTaskNotFoundError,
+    )
+    from airflow.sdk.exceptions import AirflowRuntimeError
+
+    awaited_runs: list[tuple[str, dict[str, list[Any]]]]
+    if run_ids:
+        awaited_runs = [(run_id, {"run_ids": [run_id]}) for run_id in run_ids]
+    else:
+        awaited_runs = [(dt.isoformat(), {"logical_dates": [dt]}) for dt in logical_dates or []]
+
+    all_runs_checked = True
+    for run_label, run_filter in awaited_runs:
+        if api.get_dr_count(dag_id=external_dag_id, **run_filter) == 0:
+            all_runs_checked = False
+            continue
+
+        if external_task_ids:
+            try:
+                api.get_task_states(dag_id=external_dag_id, task_ids=list(external_task_ids), **run_filter)
+            except AirflowRuntimeError as e:
+                if (message := _not_found_message(e)) is None:
+                    raise
+                raise ExternalTaskNotFoundError(
+                    f"The external tasks {list(external_task_ids)} in Dag {external_dag_id} "
+                    f"do not all exist for run {run_label}: {message}"
+                ) from None
+
+        if external_task_group_id:
+            try:
+                api.get_task_states(
+                    dag_id=external_dag_id, task_group_id=external_task_group_id, **run_filter
+                )
+            except AirflowRuntimeError as e:
+                if (message := _not_found_message(e)) is None:
+                    raise
+                raise ExternalTaskGroupNotFoundError(
+                    f"The external task group '{external_task_group_id}' in Dag '{external_dag_id}' "
+                    f"does not exist for run {run_label}: {message}"
+                ) from None
+
+    return all_runs_checked