bingqin2 opened a new pull request, #73085:
URL: https://github.com/apache/airflow/pull/73085
On Airflow 3, `check_existence=True` on `ExternalTaskSensor` verified
nothing: a worker has no database access, and the Airflow 2 lookup (`DagModel`
plus a `DagBag` of the file) has no equivalent through the execution API.
#72514 tracks this, and #72517 adds the Dag-level lookup (`ti.get_dag`, Airflow
3.2+). This adds the task and task-group half. It is independent of #72517: it
only touches the Airflow 3 branch of the existence check and the trigger, so
whichever merges second has a small rebase in `_check_for_existence`.
**How it decides**
A Dag run's task instances are created together with the run, in the same
transaction and from the run's own Dag version: `_create_orm_dagrun` adds and
flushes the run and calls `verify_integrity` before the caller commits, and
every persisted creation path goes through it. So once an awaited run exists,
its task instances are the version-accurate answer to whether a task is part of
that run, and the worker can read them through the same `task-instances/count`
and `task-instances/states` calls the sensor already polls with. Per awaited
run:
- no run yet (`get_dr_count` is 0): keep waiting; the check repeats on later
pokes until every awaited run has been seen
- the run exists and has no task instance for one of `external_task_ids`:
`ExternalTaskNotFoundError`
- `external_task_group_id`: `get_task_states(task_group_id=...)` answers 404
for a group the Dag does not define (`ExternalTaskGroupNotFoundError`); an
empty answer for a run that exists means the group's tasks are not part of that
run (same exception)
This follows the direction given on #67832: no dedicated existence endpoint,
and no answer before a run exists. Missing tasks and groups are configuration
errors, so they are raised regardless of `soft_fail`, as on Airflow 2.
The deferrable path gets the same rule. `WorkflowTrigger` takes
`check_existence` (serialized, default `False`, so already-serialized triggers
are unaffected) and yields `{"status": "not_found", "kind": "task" |
"task_group", "message": ...}`, which `execute_complete` maps to the matching
exception. The Airflow 2 path is untouched, and no execution API version change
is needed: the count and states calls exist on every Airflow 3 release.
**Known limits**
- The server resolves a task group against the latest Dag version
(`_get_group_tasks`), so a group renamed after a pinned run was created is
reported missing for that run. A follow-up core PR makes `_get_group_tasks`
resolve the group against the version the run resolves to; on servers without
it this limitation stays, and it is the existing behaviour of the states
endpoint.
- The scheduler adds task instances to an unfinished run when the Dag gains
a task after the run was created (`_verify_integrity_if_dag_changed`), so a
poke between that Dag update and the next scheduler pass sees no instance for a
task that appears seconds later. Airflow 2 had the mirror image, since it
checked the current file once.
- Runs that are never created, for example an `execution_date_fn` pointing
at dates that never run, still wait until timeout, as on Airflow 2.
- Whether the external Dag itself is registered is not checked here; that is
#72517.
**Changes**
- `utils/sensor_helper.py`: `_check_external_task_existence(api, ...)`, the
decision rule, usable from the worker (`ti`) and from the triggerer
(`RuntimeTaskInstance`)
- `sensors/external_task.py`: `_check_for_existence_af3` runs the rule on
each poke until every awaited run has been checked; the trigger receives
`check_existence`; `execute_complete` handles `not_found`
- `triggers/external_task.py`: `check_existence` parameter, serialized; the
existence loop before the state polling
- tests: 8 sensor tests (waits for the run then checks once, task missing,
`soft_fail` ignored, group missing, group unknown, other API errors kept,
deferrable hands over to the trigger, `execute_complete` mapping) and 3 trigger
tests (task missing, group missing, waits for the run)
**Testing**
- `providers/standard`:
`tests/unit/standard/sensors/test_external_task_sensor.py`,
`tests/unit/standard/triggers/test_external_task.py`,
`tests/unit/standard/utils/test_sensor_helper.py` (75 tests on Airflow 3)
- mypy on the three changed modules, prek hooks on the changed files
Part of #72514 (the task and task-group half; #72517 covers the Dag level).
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Claude Fable 5.1) following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions).
I reviewed and understand all changes; the tests were run locally as listed
above.
---
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
