You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "nathadfield (via GitHub)" <gi...@apache.org> on 2026/05/21 08:28:31 UTC
[PR] Apply `rerun_with_latest_version` to TriggerDagRunOperator reruns [airflow]
nathadfield opened a new pull request, #67273:
URL: https://github.com/apache/airflow/pull/67273
## Motivation
PR #63884 introduced a 4-tier `rerun_with_latest_version` config hierarchy
(explicit > DAG-level > global config > fallback) controlling whether
clearing/rerunning a Dag run uses the latest bundle version. That hierarchy
is wired into the UI clear dialogs, the public clear/backfill API endpoints,
and the `airflow backfill create` CLI command — but **not** into
`TriggerDagRunOperator`'s `reset_dag_run=True` path.
When `TriggerDagRunOperator(reset_dag_run=True)` is invoked and the target
Dag run already exists, the task SDK trigger client gets a 409 and falls
back to `client.clear(dag_id, run_id)`. That hits
`POST /execution/dag-runs/{dag_id}/{run_id}/clear` on the execution API,
which previously called `dag.clear(run_id=run_id)` with no
`run_on_latest_version` argument. As a result, the global and DAG-level
`rerun_with_latest_version` config had no effect on operator-driven reruns.
## What this PR does
Wires the existing `resolve_run_on_latest_version` resolver into the
execution-API `clear_dag_run` endpoint, so operator-driven reruns honor
the same hierarchy as UI-, API-, and CLI-driven reruns.
```python
resolved_run_on_latest = resolve_run_on_latest_version(None, dag_id, session)
dag.clear(run_id=run_id, run_on_latest_version=resolved_run_on_latest)
```
`None` is passed as the explicit override because the operator does not
(intentionally) accept a per-call override — the DAG-level toggle and
global config from #63884 are already sufficient, and adding an operator
parameter that accepts arbitrary versions would create a security gap (a
git bundle could be made to run a commit that was never on the tracking
branch).
Brand-new Dag runs from `TriggerDagRunOperator` are unaffected: they
already use the latest bundle version because `_create_orm_dagrun` reads
`DagModel.bundle_version` at trigger time. Only the rerun/clear path
needed wiring.
No payload/schema change; no client change; no Cadwyn version bump.
## Test plan
- [x] New unit test `test_dag_run_clear_invokes_resolver` verifies the
resolver is invoked with `None` as the explicit override.
- [x] Existing `TestDagRunClear` tests still pass unchanged.
- [x] `prek run --from-ref apache/main --stage pre-commit` — all hooks pass
(ruff, mypy-airflow-core, etc.).
closes: #60880
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (Opus 4.7)
Generated-by: Claude Code (Opus 4.7) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Apply `rerun_with_latest_version` to TriggerDagRunOperator reruns [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on PR #67273:
URL: https://github.com/apache/airflow/pull/67273#issuecomment-4647827103
@ashb @kaxil @amoghrajesh Might one of you be able to review this at some point? Thanks very much!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Apply `rerun_with_latest_version` to TriggerDagRunOperator reruns [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #67273:
URL: https://github.com/apache/airflow/pull/67273#discussion_r3439589449
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py:
##########
@@ -30,6 +30,7 @@
from airflow.api_fastapi.common.db.common import SessionDep
from airflow.api_fastapi.common.types import UtcDateTime
from airflow.api_fastapi.compat import HTTP_422_UNPROCESSABLE_CONTENT
+from airflow.api_fastapi.core_api.services.public.common import resolve_run_on_latest_version
Review Comment:
Can we avoid importing from the public REST service layer here? The execution API is worker-facing, and its existing `core_api` dependencies are only `core_api.base` and `core_api.openapi.exceptions` (shared leaf modules). `core_api.services.public.common` is the REST service tier, and it imports the `Bulk*` REST datamodels at module level, so this one-line import now pulls the public bulk-operations layer into the worker API.
`resolve_run_on_latest_version` has no REST dependency of its own (it only touches `conf` and `SerializedDagModel`), so it doesn't really belong in the public-service module. Moving it to `airflow.api_fastapi.common` (which both `core_api` and `execution_api` already depend on) and importing it from there in both this endpoint and the public clear endpoint keeps it shared without crossing the execution -> core_api.services boundary.
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_dag_runs.py:
##########
@@ -322,6 +324,27 @@ def test_dag_run_not_found(self, client):
assert response.status_code == 404
+ def test_dag_run_clear_invokes_resolver(self, client, session, dag_maker):
+ """Clearing invokes resolve_run_on_latest_version with no explicit override."""
+ dag_id = "test_clear_invokes_resolver"
+ run_id = "test_run_id"
+
+ with dag_maker(dag_id=dag_id, session=session, serialized=True):
+ EmptyOperator(task_id="test_task")
+ dag_maker.create_dagrun(run_id=run_id, state=DagRunState.SUCCESS)
+ session.commit()
+
+ with mock.patch(
+ "airflow.api_fastapi.execution_api.routes.dag_runs.resolve_run_on_latest_version",
+ return_value=False,
+ ) as mock_resolver:
+ response = client.post(f"/execution/dag-runs/{dag_id}/{run_id}/clear")
+
+ assert response.status_code == 204
+ mock_resolver.assert_called_once()
Review Comment:
This asserts the resolver is *called* with `None`, but not that its return value reaches `dag.clear()`. The resolver is mocked to return `False`, yet nothing checks that `False` is what `dag.clear` receives, so an implementation that computes `resolved_run_on_latest` and then calls `dag.clear(run_id=run_id)` (dropping it) would still pass. That dropped-value case is exactly the regression this PR guards against. Consider spying on `SerializedDAG.clear` (autospec) and asserting it gets `run_on_latest_version=<the resolved value>`, so the test pins the data flow and not just the call.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Apply `rerun_with_latest_version` to TriggerDagRunOperator reruns [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #67273:
URL: https://github.com/apache/airflow/pull/67273#discussion_r3441207652
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py:
##########
@@ -30,6 +30,7 @@
from airflow.api_fastapi.common.db.common import SessionDep
from airflow.api_fastapi.common.types import UtcDateTime
from airflow.api_fastapi.compat import HTTP_422_UNPROCESSABLE_CONTENT
+from airflow.api_fastapi.core_api.services.public.common import resolve_run_on_latest_version
Review Comment:
Good catch. I've moved `resolve_run_on_latest_version` into `airflow.api_fastapi.common.dagbag` and repointed all the callers so the execution API no longer reaches into `core_api.services`.
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_dag_runs.py:
##########
@@ -322,6 +324,27 @@ def test_dag_run_not_found(self, client):
assert response.status_code == 404
+ def test_dag_run_clear_invokes_resolver(self, client, session, dag_maker):
+ """Clearing invokes resolve_run_on_latest_version with no explicit override."""
+ dag_id = "test_clear_invokes_resolver"
+ run_id = "test_run_id"
+
+ with dag_maker(dag_id=dag_id, session=session, serialized=True):
+ EmptyOperator(task_id="test_task")
+ dag_maker.create_dagrun(run_id=run_id, state=DagRunState.SUCCESS)
+ session.commit()
+
+ with mock.patch(
+ "airflow.api_fastapi.execution_api.routes.dag_runs.resolve_run_on_latest_version",
+ return_value=False,
+ ) as mock_resolver:
+ response = client.post(f"/execution/dag-runs/{dag_id}/{run_id}/clear")
+
+ assert response.status_code == 204
+ mock_resolver.assert_called_once()
Review Comment:
Agreed. I've updated the test to spy on `SerializedDAG.clear` and assert the resolved value is actually passed through as `run_on_latest_version`, so it now catches the dropped-value case rather than just checking the resolver was called.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Apply `rerun_with_latest_version` to TriggerDagRunOperator reruns [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil merged PR #67273:
URL: https://github.com/apache/airflow/pull/67273
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org