1fanwang opened a new pull request, #69135: URL: https://github.com/apache/airflow/pull/69135
Outcome of the `[DISCUSS] Let TriggerDagRunOperator own its execution logic` thread on the dev list. ## Why On Airflow 3, `TriggerDagRunOperator.execute()` raises `DagRunTriggerException` and the **task runner** does the trigger and the synchronous wait loop. So the trigger-and-wait contract (and any crash recovery) lives in two places — the operator and the runner — and can drift. The poll half is already a first-class Task SDK accessor (`ti.get_dagrun_state()`); only the trigger half was missing, which is why it had to go through the exception side channel. ## What - Add `ti.trigger_dag_run()` — the symmetric counterpart to `ti.get_dagrun_state()`. It routes a trigger through the same execution-API endpoint and scoped token the runner already uses, so there is no new authz surface. - On Airflow 3.3+ non-deferrable runs, `TriggerDagRunOperator` now does the submit and poll itself and subclasses `ResumableJobMixin` directly, so the durability contract lives in one place. - Add an opt-in `durable` flag (default `False`): on a synchronous `wait_for_completion`, the triggered run id is persisted before polling, so a worker crash mid-wait reconnects to the in-flight run on retry instead of triggering a duplicate. Scoped deliberately: deferrable keeps the `DagRunTriggerException` path (it still needs the triggerer handoff). Airflow < 3.3 and Airflow 2 are unchanged — the operator falls back to the existing path. ## Tests `41 passed, 41 skipped` for the operator suite (the 41 skips are Airflow-2-only, on 3.4): - New unit tests for the accessor and the operator-owned path: trigger, skip/reset, sync wait success/failure, fire-and-forget, and the four durable states (persist-before-poll, reconnect, short-circuit on an already-succeeded prior run, resubmit after a failed one). - The existing synchronous-path tests are re-homed to assert the accessor calls instead of the raised exception (the operator now owns that execution). Deferrable / Airflow-2 tests are unchanged. ## E2E — live, Airflow 3.4 standalone (LocalExecutor + Postgres) Two DAGs: a parent `TriggerDagRunOperator(wait_for_completion=True, durable=True, poke_interval=3)` triggers a child whose only task sleeps 45s. **1. Clean path (no crash)** — proves the operator owns its execution via the accessors; the runner's `DagRunTriggerException` path is never taken. | t | parent task | child runs | |---|---|---| | 8s | running (try 1) | **1** (running) | | 48s | **success (try 1)** | 1 (success) | Parent task log shows the operator's own poll loop (`trigger_dagrun.py`), and the side channel is unused: ```text Waiting for e2e_child on manual__…920653+00:00 to reach an allowed state [SUCCESS] ... DagRunTriggerException occurrences in the task log: 0 ``` **2. Worker `SIGKILL` mid-wait (durable reconnect)** — the parent's worker process is `kill -9`ed while it polls; the scheduler retries the task. | t | parent task | child runs created by this parent | |---|---|---| | 6s | running (try 1), triggered child | **1** | | — | `SIGKILL` the parent worker pid | 1 | | 9s | up_for_retry (try 1) | 1 | | 12s | running (**try 2**) | 1 (no new trigger) | | 42s | **success (try 2)** | **1** | On the retry the operator reconnects to the **same** run id instead of triggering a duplicate: ```text attempt=2: Reconnecting to existing job attempt=2: Waiting for e2e_child on manual__…957584+00:00 ... <- same run id as attempt=1 attempt=2: Triggering events: 0 | DagRunTriggerException: 0 parent task: success (try_number=2) child dag runs created: 1 (delta=1 -> reconnected, not duplicated) ``` <details><summary>Raw state progression — crash run</summary> ```text child runs before crash test: 1 (left over from the clean run above) [t=3s] parent=(none) try=0 | child_total=1 [t=6s] parent=running try=1 pid=62403| child_total=2 <- triggered + polling >>> SIGKILL parent worker pid=62403 (simulate crash mid-wait) [t=9s] parent=up_for_retry try=1 | child_total=2 [t=12s] parent=running try=2 pid=62469| child_total=2 <- retry; NO new child [t=15s..39s] parent=running try=2 | child_total=2 <- reconnect-polling the in-flight run [t=42s] parent=success try=2 | child_total=2 === child delta=1 (1=reconnect/no-dup, 2=DUPLICATE) === manual__2026-…T16:51:09.920653+00:00 = success (clean run's child) manual__2026-…T16:54:29.957584+00:00 = success (crash run's child — single run despite the crash+retry) ``` </details> ## Refs This is the accessor-based design the dev-list thread converged on — an alternative to the hand-rolled-in-the-runner durability in #68936 and the share-the-core refactor in #68952 / #68955. --- ##### Was generative AI tooling used to co-author this PR? - [ ] Yes (please specify the tool below) -- 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]
