bingqin2 commented on issue #72514:
URL: https://github.com/apache/airflow/issues/72514#issuecomment-5654028667

   Thanks, both points are right.
   
   **Case 2.** I went through every path that persists a run: the REST trigger 
route, the asset-event route, the scheduler (scheduled, asset-triggered and 
backfill runs), `airflow.api.common.trigger_dag` and `dag.test()` all end in 
`DAG.create_dagrun` → `_create_orm_dagrun`, which does `session.add(run)`, 
`session.flush()` and `run.verify_integrity(...)` before returning, and the 
caller commits after that. The only `DagRun(...)` built anywhere else is the 
in-memory run of `airflow tasks test`, which is never written. So a committed 
run always carries the task instances of its Dag version, and a run with none 
is a run of a Dag without tasks, where the awaited task cannot exist either. 
I'll drop the guard: run exists and has no task instance for the task ⇒ not 
found.
   
   The one window that remains is `_verify_integrity_if_dag_changed`: the 
scheduler adds task instances to an unfinished run when the Dag gains a task 
after the run was created, 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 (it checked the current file once, and a task 
present there could still have no instance in the awaited run). I'd document it 
rather than keep waiting on unfinished runs, since failing fast while the 
external Dag is running is the point of the option.
   
   **Task groups.** Agreed, it should not be left for later, and the fix 
belongs in core: `_get_group_tasks` can resolve the group against the version 
each named run resolves to, through `dag_bag.get_dag_for_run(dag_run, 
session)`: the pinned version for runs of a versioned bundle, the latest 
version otherwise, which is also how the scheduler treats those runs when it 
re-verifies them. The latest version stays the answer only when the request 
names no run. That is a behaviour fix inside the existing 
`task-instances/count` and `task-instances/states` endpoints, so no Execution 
API version change. I'll open it as a separate core PR next to the provider 
one; on servers without it the provider keeps today's limitation, which the 
docstring will state.
   
   **Base.** Since #72517 has not moved for nine days, I'll open the 
task/task-group half against `main` rather than stacked on it. It only touches 
the Airflow 3 branch of the existence check and the trigger, so whichever of 
the two merges second has a small rebase in `_check_for_existence`. On its own 
the change still waits, as today, when the external Dag itself does not exist, 
which is exactly the part #72517 adds.
   


-- 
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]

Reply via email to