You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "hkc-8010 (via GitHub)" <gi...@apache.org> on 2026/05/14 07:53:02 UTC
[PR] Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry [airflow]
hkc-8010 opened a new pull request, #66911:
URL: https://github.com/apache/airflow/pull/66911
Avoid retrying the non-idempotent trigger Dag run POST through the generic execution API retry layer.
When the trigger POST raises an ambiguous transport-level `httpx.RequestError`, the client now probes the requested Dag run ID. If the run exists and `reset_dag_run=False`, it treats the trigger as successful. If the run is missing, it re-raises the original request error.
This preserves direct `409 Conflict` behavior, including `DAGRUN_ALREADY_EXISTS` and the existing `reset_dag_run=True` clear path.
closes: #66905
Tests:
- `uv run --project task-sdk ruff format task-sdk/src/airflow/sdk/api/client.py task-sdk/tests/task_sdk/api/test_client.py`
- `uv run --project task-sdk ruff check --fix task-sdk/src/airflow/sdk/api/client.py task-sdk/tests/task_sdk/api/test_client.py`
- `uv run --project task-sdk pytest task-sdk/tests/task_sdk/api/test_client.py::TestDagRunOperations -q`
- `prek run --files task-sdk/src/airflow/sdk/api/client.py task-sdk/tests/task_sdk/api/test_client.py`
- `breeze testing task-sdk-tests --python 3.10 task-sdk/tests/task_sdk/api/test_client.py::TestDagRunOperations`
- `breeze testing task-sdk-tests --python 3.10`
- `breeze ci selective-check --commit-ref HEAD --default-branch main --github-repository apache/airflow`
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: OpenAI Codex following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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] Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry [airflow]
Posted by "hkc-8010 (via GitHub)" <gi...@apache.org>.
hkc-8010 commented on PR #66911:
URL: https://github.com/apache/airflow/pull/66911#issuecomment-4488299860
@jason810496 Addressed all four points — see commit 8a9259d5c0f:
- Pre-check and ambiguous-POST follow-up now both use self.get_count(dag_id=dag_id, run_ids=[run_id]) directly, removing _dag_run_exists and its ~40 lines of compat handling.
- _is_legacy_previous_dag_run_route_response was removed along with _dag_run_exists (it was only used there).
- Added a TODO at the clear() call site for the idempotency follow-up.
Would appreciate a re-review. Thanks!
--
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] Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry [airflow]
Posted by "hkc-8010 (via GitHub)" <gi...@apache.org>.
hkc-8010 commented on code in PR #66911:
URL: https://github.com/apache/airflow/pull/66911#discussion_r3266710056
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -1125,6 +1208,19 @@ def _update_auth(self, response: httpx.Response):
log.debug("Execution API issued us a refreshed Task token")
self.auth = BearerAuth(new_token)
+ @staticmethod
+ def _ensure_json_content_type(kwargs: dict[str, Any]) -> None:
+ # Set content type as convenience if not already set
+ if kwargs.get("content", None) is not None and "content-type" not in (
+ kwargs.get("headers", {}) or {}
+ ):
Review Comment:
Since _dag_run_exists was removed entirely, _is_legacy_previous_dag_run_route_response (which was only used by it) was also removed. The duplication no longer exists.
--
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] Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #66911:
URL: https://github.com/apache/airflow/pull/66911#discussion_r3265070898
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -869,22 +869,104 @@ def trigger(
run_after=run_after,
)
+ dag_run_exists_before_trigger = self._dag_run_exists(dag_id=dag_id, run_id=run_id)
+ if dag_run_exists_before_trigger is True:
+ if reset_dag_run:
+ log.info("Dag Run already exists; Resetting Dag Run.", dag_id=dag_id, run_id=run_id)
+ return self.clear(run_id=run_id, dag_id=dag_id)
+ log.info("Dag Run already exists!", dag_id=dag_id, run_id=run_id)
+ return ErrorResponse(error=ErrorType.DAGRUN_ALREADY_EXISTS)
+
try:
- self.client.post(
- f"dag-runs/{dag_id}/{run_id}", content=body.model_dump_json(exclude_defaults=True)
+ self.client._request_without_retry(
+ "POST", f"dag-runs/{dag_id}/{run_id}", content=body.model_dump_json(exclude_defaults=True)
)
+ except (httpx.ReadError, httpx.ReadTimeout, httpx.RemoteProtocolError):
+ if (
+ dag_run_exists_before_trigger is False
+ and self._dag_run_exists(dag_id=dag_id, run_id=run_id, retry=True) is True
+ ):
+ log.info(
+ "Dag Run exists after ambiguous trigger response; treating trigger as successful.",
+ dag_id=dag_id,
+ run_id=run_id,
+ )
+ return OKResponse(ok=True)
+ raise
except ServerResponseError as e:
if e.response.status_code == HTTPStatus.CONFLICT:
if reset_dag_run:
- log.info("Dag Run already exists; Resetting Dag Run.", dag_id=dag_id, run_id=run_id)
+ log.info(
+ "Dag Run already exists after trigger attempt; Resetting Dag Run.",
+ detail=e.detail,
+ dag_id=dag_id,
+ run_id=run_id,
+ )
return self.clear(run_id=run_id, dag_id=dag_id)
-
- log.info("Dag Run already exists!", detail=e.detail, dag_id=dag_id, run_id=run_id)
+ log.info(
+ "Dag Run already exists after trigger attempt.",
+ detail=e.detail,
+ dag_id=dag_id,
+ run_id=run_id,
+ )
return ErrorResponse(error=ErrorType.DAGRUN_ALREADY_EXISTS)
raise
return OKResponse(ok=True)
+ def _dag_run_exists(self, dag_id: str, run_id: str, *, retry: bool = False) -> bool | None:
+ """Return whether the Dag run exists, or None when the detail endpoint is unavailable."""
+ if self.client._dry_run:
+ return None
+
+ try:
+ if retry:
+ self.client.get(f"dag-runs/{dag_id}/{run_id}")
+ else:
+ self.client._request_without_retry("GET", f"dag-runs/{dag_id}/{run_id}")
+ except httpx.RequestError:
+ return None
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.NOT_FOUND:
+ return False
+ if e.response.status_code == HTTPStatus.METHOD_NOT_ALLOWED:
+ # Older execution API servers may not support the Dag run detail endpoint yet.
+ return None
+ if (
+ e.response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
+ and run_id == "previous"
+ and self._is_legacy_previous_dag_run_route_response(e.response)
+ ):
+ return None
+ if e.response.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR:
+ return None
+ raise
+ except httpx.HTTPStatusError as e:
+ if e.response.status_code == HTTPStatus.NOT_FOUND:
+ return False
+ if e.response.status_code == HTTPStatus.METHOD_NOT_ALLOWED:
+ return None
+ if e.response.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR:
+ return None
+ raise
+ return True
Review Comment:
I will prefer to use `get_dr_count` (pass with the `run_id` paramter) to make sure whether the DagRun existed or not.
We could document down the history context here then we don't need to handle complex the compatible logic here.
Since `get_dr_count` was introduced since the first Airflow 3.0 release.
https://github.com/apache/airflow/blob/v3-0-stable/airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py#L160-L181
--
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] Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry [airflow]
Posted by "hkc-8010 (via GitHub)" <gi...@apache.org>.
hkc-8010 commented on code in PR #66911:
URL: https://github.com/apache/airflow/pull/66911#discussion_r3266708428
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -869,22 +869,104 @@ def trigger(
run_after=run_after,
)
+ dag_run_exists_before_trigger = self._dag_run_exists(dag_id=dag_id, run_id=run_id)
+ if dag_run_exists_before_trigger is True:
+ if reset_dag_run:
+ log.info("Dag Run already exists; Resetting Dag Run.", dag_id=dag_id, run_id=run_id)
+ return self.clear(run_id=run_id, dag_id=dag_id)
+ log.info("Dag Run already exists!", dag_id=dag_id, run_id=run_id)
+ return ErrorResponse(error=ErrorType.DAGRUN_ALREADY_EXISTS)
+
try:
- self.client.post(
- f"dag-runs/{dag_id}/{run_id}", content=body.model_dump_json(exclude_defaults=True)
+ self.client._request_without_retry(
+ "POST", f"dag-runs/{dag_id}/{run_id}", content=body.model_dump_json(exclude_defaults=True)
)
+ except (httpx.ReadError, httpx.ReadTimeout, httpx.RemoteProtocolError):
+ if (
+ dag_run_exists_before_trigger is False
+ and self._dag_run_exists(dag_id=dag_id, run_id=run_id, retry=True) is True
+ ):
+ log.info(
+ "Dag Run exists after ambiguous trigger response; treating trigger as successful.",
+ dag_id=dag_id,
+ run_id=run_id,
+ )
+ return OKResponse(ok=True)
+ raise
except ServerResponseError as e:
if e.response.status_code == HTTPStatus.CONFLICT:
if reset_dag_run:
- log.info("Dag Run already exists; Resetting Dag Run.", dag_id=dag_id, run_id=run_id)
+ log.info(
+ "Dag Run already exists after trigger attempt; Resetting Dag Run.",
+ detail=e.detail,
+ dag_id=dag_id,
+ run_id=run_id,
+ )
return self.clear(run_id=run_id, dag_id=dag_id)
-
- log.info("Dag Run already exists!", detail=e.detail, dag_id=dag_id, run_id=run_id)
+ log.info(
+ "Dag Run already exists after trigger attempt.",
+ detail=e.detail,
+ dag_id=dag_id,
+ run_id=run_id,
+ )
return ErrorResponse(error=ErrorType.DAGRUN_ALREADY_EXISTS)
raise
return OKResponse(ok=True)
+ def _dag_run_exists(self, dag_id: str, run_id: str, *, retry: bool = False) -> bool | None:
+ """Return whether the Dag run exists, or None when the detail endpoint is unavailable."""
+ if self.client._dry_run:
+ return None
+
+ try:
+ if retry:
+ self.client.get(f"dag-runs/{dag_id}/{run_id}")
+ else:
+ self.client._request_without_retry("GET", f"dag-runs/{dag_id}/{run_id}")
+ except httpx.RequestError:
+ return None
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.NOT_FOUND:
+ return False
+ if e.response.status_code == HTTPStatus.METHOD_NOT_ALLOWED:
+ # Older execution API servers may not support the Dag run detail endpoint yet.
+ return None
+ if (
+ e.response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
+ and run_id == "previous"
+ and self._is_legacy_previous_dag_run_route_response(e.response)
+ ):
+ return None
+ if e.response.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR:
+ return None
+ raise
+ except httpx.HTTPStatusError as e:
+ if e.response.status_code == HTTPStatus.NOT_FOUND:
+ return False
+ if e.response.status_code == HTTPStatus.METHOD_NOT_ALLOWED:
+ return None
+ if e.response.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR:
+ return None
+ raise
+ return True
Review Comment:
Done in 8a9259d5c0f. Replaced _dag_run_exists entirely with self.get_count(dag_id=dag_id, run_ids=[run_id]). The pre-check and the follow-up ambiguous-POST probe both now use the dag-runs/count endpoint.
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -869,22 +869,104 @@ def trigger(
run_after=run_after,
)
+ dag_run_exists_before_trigger = self._dag_run_exists(dag_id=dag_id, run_id=run_id)
Review Comment:
Done in 8a9259d5c0f. Both the pre-check (line 873) and the follow-up probe after an ambiguous POST (line 888) now call self.get_count(dag_id=dag_id, run_ids=[run_id]) directly.
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -869,22 +869,104 @@ def trigger(
run_after=run_after,
)
+ dag_run_exists_before_trigger = self._dag_run_exists(dag_id=dag_id, run_id=run_id)
+ if dag_run_exists_before_trigger is True:
+ if reset_dag_run:
+ log.info("Dag Run already exists; Resetting Dag Run.", dag_id=dag_id, run_id=run_id)
+ return self.clear(run_id=run_id, dag_id=dag_id)
Review Comment:
Added a TODO at the clear() call site in trigger() noting this as a follow-up. Happy to open a tracking issue if that would help.
--
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] Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #66911:
URL: https://github.com/apache/airflow/pull/66911#issuecomment-4529572419
@hkc-8010 A few things need addressing before review — see our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
- :x: **CI fails**: `CI image checks / Static checks` (and possibly other checks — see the [Checks tab](https://github.com/apache/airflow/pull/66911/checks) for the full list).
No rush.
---
_Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this [two-stage triage process](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md#why-the-first-pass-is-automated) so that our maintainers' limited time is spent where it matters most: the conversation with you._
---
Drafted-by: Claude Code (Opus 4.7); reviewed by @potiuk before posting
--
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] Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry [airflow]
Posted by "hkc-8010 (via GitHub)" <gi...@apache.org>.
hkc-8010 commented on code in PR #66911:
URL: https://github.com/apache/airflow/pull/66911#discussion_r3260589635
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -870,9 +870,18 @@ def trigger(
)
try:
- self.client.post(
- f"dag-runs/{dag_id}/{run_id}", content=body.model_dump_json(exclude_defaults=True)
+ self.client._request_without_retry(
+ "POST", f"dag-runs/{dag_id}/{run_id}", content=body.model_dump_json(exclude_defaults=True)
Review Comment:
Done in e30cdc50d16. `trigger()` now performs a `GET dag-runs/{dag_id}/{run_id}` pre-check before the `POST`. If the Dag run already exists, `reset_dag_run=True` clears it and the default path returns `DAGRUN_ALREADY_EXISTS`. If the detail endpoint is unavailable, the client falls back to the existing POST path for compatibility.
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -870,9 +870,18 @@ def trigger(
)
try:
- self.client.post(
- f"dag-runs/{dag_id}/{run_id}", content=body.model_dump_json(exclude_defaults=True)
+ self.client._request_without_retry(
+ "POST", f"dag-runs/{dag_id}/{run_id}", content=body.model_dump_json(exclude_defaults=True)
)
+ except httpx.RequestError:
+ if not reset_dag_run and self._dag_run_exists(dag_id=dag_id, run_id=run_id):
Review Comment:
Done in e30cdc50d16. I implemented the pre-check flow from this comment: pre-existing Dag runs clear or return `DAGRUN_ALREADY_EXISTS` without posting, and the trigger POST only happens when the pre-check says the run is missing or the check is unavailable. Ambiguous read-side POST failures are converted to `OKResponse` only when the pre-check saw the run as missing and the follow-up GET finds it. `ConnectError` and `PoolTimeout` still propagate.
I also added regression coverage for pre-existing conflict and reset, unsupported and failed pre-check fallback, ambiguous `ReadError`/`ReadTimeout`/`RemoteProtocolError`, the conservative pre-check-unknown path, direct `409` behavior, reset conflict clearing, and `ConnectError`/`PoolTimeout` propagation.
--
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] Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #66911:
URL: https://github.com/apache/airflow/pull/66911#issuecomment-5616837656
Closing this one and track the follow-up item in https://github.com/apache/airflow/issues/72876, please feel free to reopen this if necessary.
--
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] Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry [airflow]
Posted by "hkc-8010 (via GitHub)" <gi...@apache.org>.
hkc-8010 commented on PR #66911:
URL: https://github.com/apache/airflow/pull/66911#issuecomment-4554253074
@potiuk @jason810496 The earlier static-check failure appears to have been dependency lock drift after the branch fell behind main. I merged main into this branch, CI reran on the latest head, and all required checks are now green, including `CI image checks / Static checks` and `Task SDK tests / task-sdk:P3.10 tests`.
Would you mind taking another look when you get a chance? Thanks.
--
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] Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry [airflow]
Posted by "hkc-8010 (via GitHub)" <gi...@apache.org>.
hkc-8010 commented on PR #66911:
URL: https://github.com/apache/airflow/pull/66911#issuecomment-5385347763
@jason810496 Fair point on the extra round-trip, I've reworked it to drop the pre-check.
The probe now only runs inside the `except` for the ambiguous transport errors, so a trigger that succeeds costs exactly one request, same as before this PR. `test_trigger` asserts the request list is a single POST so it can't drift back. No extra pressure on the API server or DB on the normal path.
On relying on the FK constraint: I agree the server will only ever hold one DagRun for that run ID. The problem this fixes is on the client side, where after a lost response it can't tell whether the run it finds is the one its own POST just created or one that was already there. The FK guarantees uniqueness but not authorship, so the client still has to ask.
One behaviour change worth calling out now that the pre-check is gone: if a run genuinely already existed and the POST fails with an ambiguous transport error, `trigger()` returns OK instead of `DAGRUN_ALREADY_EXISTS`. That needs both things to happen in the same call. It's written up under "Behaviour change" in the PR description. If you'd rather keep that distinction I can compare the found run's `logical_date`/`conf` against what was sent, still without adding anything to the happy path.
I also removed a few tests that duplicated the existing `test_trigger_conflict` pair, and collapsed the per-exception variants into parametrized ones.
--
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] Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 closed pull request #66911: Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry
URL: https://github.com/apache/airflow/pull/66911
--
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] Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #66911:
URL: https://github.com/apache/airflow/pull/66911#discussion_r3265080004
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -869,22 +869,104 @@ def trigger(
run_after=run_after,
)
+ dag_run_exists_before_trigger = self._dag_run_exists(dag_id=dag_id, run_id=run_id)
Review Comment:
Then we could use `self.get_count` directly. SInce it's definitely safe for retry.
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -869,22 +869,104 @@ def trigger(
run_after=run_after,
)
+ dag_run_exists_before_trigger = self._dag_run_exists(dag_id=dag_id, run_id=run_id)
+ if dag_run_exists_before_trigger is True:
+ if reset_dag_run:
+ log.info("Dag Run already exists; Resetting Dag Run.", dag_id=dag_id, run_id=run_id)
+ return self.clear(run_id=run_id, dag_id=dag_id)
Review Comment:
We should introduce the idempotent logic for `self.clear` as follow-up as well. Make sure the DagRun really existed before having the `clear` API call.
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -869,22 +869,104 @@ def trigger(
run_after=run_after,
)
+ dag_run_exists_before_trigger = self._dag_run_exists(dag_id=dag_id, run_id=run_id)
+ if dag_run_exists_before_trigger is True:
+ if reset_dag_run:
+ log.info("Dag Run already exists; Resetting Dag Run.", dag_id=dag_id, run_id=run_id)
+ return self.clear(run_id=run_id, dag_id=dag_id)
+ log.info("Dag Run already exists!", dag_id=dag_id, run_id=run_id)
+ return ErrorResponse(error=ErrorType.DAGRUN_ALREADY_EXISTS)
+
try:
- self.client.post(
- f"dag-runs/{dag_id}/{run_id}", content=body.model_dump_json(exclude_defaults=True)
+ self.client._request_without_retry(
+ "POST", f"dag-runs/{dag_id}/{run_id}", content=body.model_dump_json(exclude_defaults=True)
)
+ except (httpx.ReadError, httpx.ReadTimeout, httpx.RemoteProtocolError):
+ if (
+ dag_run_exists_before_trigger is False
+ and self._dag_run_exists(dag_id=dag_id, run_id=run_id, retry=True) is True
+ ):
+ log.info(
+ "Dag Run exists after ambiguous trigger response; treating trigger as successful.",
+ dag_id=dag_id,
+ run_id=run_id,
+ )
+ return OKResponse(ok=True)
+ raise
except ServerResponseError as e:
if e.response.status_code == HTTPStatus.CONFLICT:
if reset_dag_run:
- log.info("Dag Run already exists; Resetting Dag Run.", dag_id=dag_id, run_id=run_id)
+ log.info(
+ "Dag Run already exists after trigger attempt; Resetting Dag Run.",
+ detail=e.detail,
+ dag_id=dag_id,
+ run_id=run_id,
+ )
return self.clear(run_id=run_id, dag_id=dag_id)
-
- log.info("Dag Run already exists!", detail=e.detail, dag_id=dag_id, run_id=run_id)
+ log.info(
+ "Dag Run already exists after trigger attempt.",
+ detail=e.detail,
+ dag_id=dag_id,
+ run_id=run_id,
+ )
return ErrorResponse(error=ErrorType.DAGRUN_ALREADY_EXISTS)
raise
return OKResponse(ok=True)
+ def _dag_run_exists(self, dag_id: str, run_id: str, *, retry: bool = False) -> bool | None:
+ """Return whether the Dag run exists, or None when the detail endpoint is unavailable."""
+ if self.client._dry_run:
+ return None
+
+ try:
+ if retry:
+ self.client.get(f"dag-runs/{dag_id}/{run_id}")
+ else:
+ self.client._request_without_retry("GET", f"dag-runs/{dag_id}/{run_id}")
+ except httpx.RequestError:
+ return None
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.NOT_FOUND:
+ return False
+ if e.response.status_code == HTTPStatus.METHOD_NOT_ALLOWED:
+ # Older execution API servers may not support the Dag run detail endpoint yet.
+ return None
+ if (
+ e.response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
+ and run_id == "previous"
+ and self._is_legacy_previous_dag_run_route_response(e.response)
+ ):
+ return None
+ if e.response.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR:
+ return None
+ raise
+ except httpx.HTTPStatusError as e:
+ if e.response.status_code == HTTPStatus.NOT_FOUND:
+ return False
+ if e.response.status_code == HTTPStatus.METHOD_NOT_ALLOWED:
+ return None
+ if e.response.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR:
+ return None
+ raise
+ return True
Review Comment:
I will prefer to use `get_dr_count` (pass with the run_id) to make sure whether the DagRun existed or not.
We could document down the history context here then we don't need to handle complex the compatible logic here.
Since `get_dr_count` was introduced since the first Airflow 3.0 release.
https://github.com/apache/airflow/blob/v3-0-stable/airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py#L160-L181
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -1125,6 +1208,19 @@ def _update_auth(self, response: httpx.Response):
log.debug("Execution API issued us a refreshed Task token")
self.auth = BearerAuth(new_token)
+ @staticmethod
+ def _ensure_json_content_type(kwargs: dict[str, Any]) -> None:
+ # Set content type as convenience if not already set
+ if kwargs.get("content", None) is not None and "content-type" not in (
+ kwargs.get("headers", {}) or {}
+ ):
Review Comment:
We should make this a staticmethod at `Client` class in `task-sdk/src/airflow/sdk/api/client.py` level.
It's duplicated of
https://github.com/apache/airflow/blob/4498582dd1ee553c403db5119ddadd7151d6fde1/task-sdk/src/airflow/sdk/api/client.py#L1138-L1142
--
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] Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #66911:
URL: https://github.com/apache/airflow/pull/66911#issuecomment-4486325853
I marked it as draft to avoid adding this to 3.2.2 release before the follow-up fix.
--
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] Avoid false TriggerDagRunOperator conflicts after ambiguous trigger retry [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #66911:
URL: https://github.com/apache/airflow/pull/66911#discussion_r3258080345
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -870,9 +870,18 @@ def trigger(
)
try:
- self.client.post(
- f"dag-runs/{dag_id}/{run_id}", content=body.model_dump_json(exclude_defaults=True)
+ self.client._request_without_retry(
+ "POST", f"dag-runs/{dag_id}/{run_id}", content=body.model_dump_json(exclude_defaults=True)
)
+ except httpx.RequestError:
+ if not reset_dag_run and self._dag_run_exists(dag_id=dag_id, run_id=run_id):
Review Comment:
More specific:
1. Pre-check GET dag-runs/{dag_id}/{run_id} before the POST.
2. If it already exists:
- reset_dag_run=True -> call clear()
- otherwise -> return ErrorResponse(error=ErrorType.DAGRUN_ALREADY_EXISTS)
3. Only if it did not exist, issue the POST.
4. If the POST later gets a 409 after an ambiguous retry, return OKResponse(ok=True) because the first POST likely created it.
5. If a final RequestError happens, only convert to OK when the pre-check said it did not exist and a follow-up GET now finds it.
Best version: track whether the failed POST was ambiguous, such as ReadTimeout/ReadError, and do not convert ConnectError or
PoolTimeout into success just because the Dag run exists afterward.
Tests should cover:
- pre-existing Dag run + transport error does not return OK
- pre-existing Dag run returns DAGRUN_ALREADY_EXISTS
- first POST creates the Dag run, response is lost, retry sees 409 -> OK
- POST fails before reaching server and Dag run still does not exist -> propagate the error
- reset_dag_run=True still clears pre-existing runs
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -870,9 +870,18 @@ def trigger(
)
try:
- self.client.post(
- f"dag-runs/{dag_id}/{run_id}", content=body.model_dump_json(exclude_defaults=True)
+ self.client._request_without_retry(
+ "POST", f"dag-runs/{dag_id}/{run_id}", content=body.model_dump_json(exclude_defaults=True)
Review Comment:
We should pre-check whether the DagRun exist before the POST API 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