bingqin2 commented on issue #72514:
URL: https://github.com/apache/airflow/issues/72514#issuecomment-5612971660
I'd like to pick up the task and task-group half of this on top of #72517,
which covers the Dag level. @ignacioparicio, if you already have it in
progress, say so and I'll stay out of the way. Before writing code I want to
check the approach, because #67832 tried a dedicated existence endpoint and the
feedback there set two constraints: no separate existence request, and no
answer before a run exists, since with Dag versioning a task only has a
definite existence per run.
**The observation the design rests on**
`create_dagrun` goes through `_create_orm_dagrun`, which calls
`run.verify_integrity(...)` before returning. A run's task instances are
therefore created in the same transaction as the run, from the run's own Dag
version. Once the awaited run exists, its task-instance rows are the
version-accurate answer to "does this task exist for this run", and a worker
can already read them through the same `task-instances/count` and
`task-instances/states` calls that `_poke_af3` makes today. That is the
information the #67832 feedback suggested reusing, with no new endpoint.
**Decision rule, per awaited logical date (unique per Dag, so at most one
run), after the Dag-level check from #72517 has passed**
1. No Dag run for that logical date yet (`get_dr_count` is 0): not an error,
keep waiting.
2. The run exists but has no task instances at all (`get_ti_count` with no
task filter is 0): keep waiting. Given `verify_integrity` this should not
happen; it is only a guard against concluding "missing" too early.
3. The run exists, has task instances, and one of `external_task_ids` has
none in it: raise `ExternalTaskNotFoundError` naming the task and the run.
4. `external_task_group_id`: `get_task_states(task_group_id=...)` already
answers 404 when the group is not in the Dag (the server resolves it through
`_get_group_tasks`), which today surfaces as a generic `AirflowRuntimeError`;
map that to `ExternalTaskGroupNotFoundError`. An empty result for a run that
exists and has task instances means the group's tasks are not part of that
run's version: same exception.
5. The check repeats on each poke until every awaited run has been checked,
then `_has_checked_existence` switches it off, so a run created after the first
poke is still verified. Cost is two or three count calls per run, once.
**Deferrable path**
The same rule goes into `WorkflowTrigger.run()`, which already calls
`RuntimeTaskInstance.get_ti_count` / `get_task_states` / `get_dr_count` from
the triggerer. The trigger gets a `check_existence` argument (serialized,
default `False`, so already-serialized triggers are unaffected) and yields
`TriggerEvent({"status": "not_found", "message": ...})`. `execute_complete`
maps that status to the matching exception. Its existing `else` branch already
turns an unknown status into `ExternalTaskNotFoundError`, so a newer triggerer
paired with an older worker still ends in the right outcome.
**Compatibility**
Provider-only change, no Execution API version change. The count and states
calls are the ones the sensor already relies on for every Airflow 3 release, so
unlike the Dag-level check this part needs no 3.2 gate. The Airflow 2 path is
untouched.
**Known limits and one open question**
- 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.
- `_get_group_tasks` resolves the group against the latest Dag version, not
the run's. A group that exists in the latest version but not in the run's
version is still caught by the empty-result branch of rule 4. The reverse, a
group present in the run's version but since renamed, would get a 404 although
the run has it. That is a pre-existing server-side limit; a small core
follow-up could resolve the group with `get_dag_for_run` when the request
identifies a single run. I would keep it out of this PR unless you want it
together.
- Alternative: make the count/states endpoints themselves answer 404 with
`reason: task_not_found` when a run exists and the task is not in its Dag
version, as suggested on #67832. That is cleaner for the server to own, but it
needs an Execution API version change, and the provider would still need the
client-side rule for older servers. I'd treat it as a later step if maintainers
prefer the server to be authoritative.
**Tests**
Sensor unit tests mocking `get_dr_count` / `get_ti_count` /
`get_task_states` for each outcome above plus "run appears on a later poke";
trigger tests for the `not_found` event; `execute_complete` mapping including
`soft_fail`. Existing tests stay unchanged.
If this looks right, I'll open the PR rebased on #72517 once it merges.
--
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]
