You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Vamsi-klu (via GitHub)" <gi...@apache.org> on 2026/03/21 06:23:04 UTC
[PR] Fix task marked as failed despite successful completion [airflow]
Vamsi-klu opened a new pull request, #64023:
URL: https://github.com/apache/airflow/pull/64023
## Summary
Fixes a race condition where a successfully completed task gets incorrectly marked as FAILED due to network retry behavior.
**Root cause:** When a task completes and the supervisor sends a SUCCESS state update to the API server, if the HTTP response is lost (network timeout), httpx retries the request. The API server sees the task is already in SUCCESS (not RUNNING) and returns 409 Conflict. This error cascades: supervisor sends an error to the task process, task exits with code 1, supervisor interprets exit code 1 as FAILED, and attempts to overwrite the correct SUCCESS state.
**Fix:** Idempotent terminal state handling at three layers:
- **API Server** (`task_instances.py`): Return 204 no-op when a terminal state update matches the current state (e.g., SUCCESS→SUCCESS). Cross-state conflicts (SUCCESS→FAILED) still return 409.
- **Supervisor `_handle_request`** (`supervisor.py`): Catch 409 on `SucceedTask` and `RetryTask` when the task is already in the target state, preventing error propagation to the task process.
- **Supervisor `update_task_state_if_needed`** (`supervisor.py`): Handle 409 during post-exit state reporting to prevent unhandled exceptions from `wait()`.
## Files Changed
| File | Change |
|------|--------|
| `airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py` | Added `_get_requested_state()` helper + idempotency check in `ti_update_state()` |
| `task-sdk/src/airflow/sdk/execution_time/supervisor.py` | Added `_is_already_in_target_state()` helper + 409 handling in `SucceedTask`/`RetryTask` handlers + 409 handling in `update_task_state_if_needed()` |
| `airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py` | 7 new tests: idempotent success/failed/skipped + 4 parametrized cross-state conflict tests |
| `task-sdk/tests/task_sdk/execution_time/test_supervisor.py` | 8 new tests: idempotent 409 for succeed/retry, different-state 409 propagation, `update_task_state_if_needed` conflict handling, `_is_already_in_target_state` unit tests |
## Impact
- **Low risk**: The API change adds an early-return path only when `previous_state == requested_state` — no existing behavior is altered for valid or genuinely conflicting transitions.
- **Defense-in-depth**: The supervisor-side handling is a secondary safety net. Even if the API fix regresses, the supervisor gracefully handles the idempotent 409 case.
## Test Plan
- [x] 7 new API server tests (idempotent + cross-state conflict)
- [x] 8 new supervisor tests (409 handling + helper unit tests)
- [x] Full `TestTIUpdateState` class: 41 passed, 1 skipped
- [ ] `test_ti_update_state_reschedule_mysql_limit` — skipped because it requires a MySQL backend (`BACKEND` env var not set in local test environment). This test is unrelated to the idempotency changes and will pass in CI with the MySQL backend configured.
closes: #63183
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (claude-opus-4-6)
Generated-by: Claude Code (claude-opus-4-6) 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] Fix task marked as failed despite successful completion [airflow]
Posted by "Pyasma (via GitHub)" <gi...@apache.org>.
Pyasma commented on PR #64023:
URL: https://github.com/apache/airflow/pull/64023#issuecomment-4103010579
@Vamsi-klu It looks like the CI is failing. Could you check it?
--
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] Fix task marked as failed despite successful completion [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #64023:
URL: https://github.com/apache/airflow/pull/64023#issuecomment-4195085510
@Vamsi-klu This PR has been converted to **draft** because it does not yet meet our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
**Issues found:**
- :x: **Image build**: Failing: Build CI images / Build CI linux/amd64 image 3.10. Check that Dockerfiles and dependencies are correct. See [Image build docs](https://github.com/apache/airflow/blob/main/contributing-docs/03a_contributors_quick_start_beginners.rst).
> **Note:** Your branch is **378 commits behind `main`**. Some check failures may be caused by changes in the base branch rather than by your PR. Please rebase your branch and push again to get up-to-date CI results.
**What to do next:**
- The comment informs you what you need to do.
- Fix each issue, then mark the PR as "Ready for review" in the GitHub UI - but only after making sure that all the issues are fixed.
- There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates.
- Maintainers will then proceed with a normal review.
Converting a PR to draft is **not** a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack).
--
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] Fix task marked as failed despite successful completion [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #64023:
URL: https://github.com/apache/airflow/pull/64023#discussion_r3025336040
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -448,6 +461,13 @@ def _handle_fail_fast_for_dag(ti: TI, dag_id: str, session: SessionDep, dag_bag:
_stop_remaining_tasks(task_instance=ti, task_teardown_map=task_teardown_map, session=session)
+def _get_requested_state(ti_patch_payload: TIStateUpdate) -> TaskInstanceState | None:
+ """Extract the requested terminal state from a TI state update payload."""
Review Comment:
`_get_requested_state()` claims to extract a "terminal" state, but it also accepts `TIRetryStatePayload` (up_for_retry). Suggest updating the docstring (or renaming the helper) to reflect that it extracts the requested *target* state for idempotency checks, not strictly terminal states.
```suggestion
"""Extract the requested target state from a TI state update payload for idempotency checks."""
```
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -363,6 +363,19 @@ def ti_update_state(
)
if previous_state != TaskInstanceState.RUNNING:
+ # Check for idempotent duplicate: if the TI is already in the exact same
+ # terminal state being requested, treat it as a successful no-op.
+ # This handles network-retry scenarios where the first request succeeded
+ # but the response was lost, causing a retry that sees the TI already
+ # in the target state.
+ requested_state = _get_requested_state(ti_patch_payload)
+ if requested_state is not None and previous_state == requested_state:
+ log.info(
+ "TI is already in the requested terminal state, treating as idempotent success",
Review Comment:
The new idempotency path is described as handling a "terminal" state, but this branch is also used for non-terminal-but-direct states like up_for_retry. Consider rewording these comments (and any related docs) to refer to the "requested target state" rather than "terminal state" to avoid confusion for future maintainers.
```suggestion
# requested target state, treat it as a successful no-op.
# This handles network-retry scenarios where the first request succeeded
# but the response was lost, causing a retry that sees the TI already
# in that requested state.
requested_state = _get_requested_state(ti_patch_payload)
if requested_state is not None and previous_state == requested_state:
log.info(
"TI is already in the requested target state, treating as idempotent success",
```
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1081,12 +1091,23 @@ def update_task_state_if_needed(self):
# For states like `deferred`, `up_for_reschedule`, the process will exit with 0, but the state will be updated
# by the subprocess in the `handle_requests` method.
if self.final_state not in STATES_SENT_DIRECTLY:
- self.client.task_instances.finish(
- id=self.id,
- state=self.final_state,
- when=datetime.now(tz=timezone.utc),
- rendered_map_index=self._rendered_map_index,
- )
+ try:
+ self.client.task_instances.finish(
+ id=self.id,
+ state=self.final_state,
+ when=datetime.now(tz=timezone.utc),
+ rendered_map_index=self._rendered_map_index,
+ )
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.CONFLICT:
+ log.warning(
+ "Failed to update TI state after process exit due to conflict",
+ ti_id=self.id,
+ final_state=self.final_state,
+ detail=getattr(e, "detail", str(e)),
Review Comment:
On conflict, this log field uses `getattr(e, "detail", str(e))`, but `ServerResponseError` details usually live on `e.response` (e.g., JSON body). Logging `e.response.json()`/`e.response.text` (guarded) would make this warning much more actionable when diagnosing why the finish update conflicted.
```suggestion
detail = getattr(e, "detail", str(e))
response_body = None
try:
if e.response is not None:
try:
response_body = e.response.json()
except Exception:
response_body = e.response.text
except Exception:
response_body = None
log.warning(
"Failed to update TI state after process exit due to conflict",
ti_id=self.id,
final_state=self.final_state,
detail=detail,
response_body=response_body,
```
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1262,22 +1283,44 @@ def _handle_request(self, msg: ToSupervisor, log: FilteringBoundLogger, req_id:
self._terminal_state = msg.state
self._task_end_time_monotonic = time.monotonic()
self._rendered_map_index = msg.rendered_map_index
- self.client.task_instances.succeed(
- id=self.id,
- when=msg.end_date,
- task_outlets=msg.task_outlets,
- outlet_events=msg.outlet_events,
- rendered_map_index=self._rendered_map_index,
- )
+ try:
+ self.client.task_instances.succeed(
+ id=self.id,
+ when=msg.end_date,
+ task_outlets=msg.task_outlets,
+ outlet_events=msg.outlet_events,
+ rendered_map_index=self._rendered_map_index,
+ )
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.CONFLICT and _is_already_in_target_state(
+ e, msg.state
+ ):
+ log.info(
+ "TI already in success state, treating as idempotent success",
+ ti_id=self.id,
+ )
Review Comment:
This info log says "treating as idempotent success". Since this is specifically swallowing a 409 conflict, wording like "idempotent no-op" / "already in target state" (and optionally including `previous_state`/`target_state`) would be clearer for operators reading logs during incident triage.
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1262,22 +1283,44 @@ def _handle_request(self, msg: ToSupervisor, log: FilteringBoundLogger, req_id:
self._terminal_state = msg.state
self._task_end_time_monotonic = time.monotonic()
self._rendered_map_index = msg.rendered_map_index
- self.client.task_instances.succeed(
- id=self.id,
- when=msg.end_date,
- task_outlets=msg.task_outlets,
- outlet_events=msg.outlet_events,
- rendered_map_index=self._rendered_map_index,
- )
+ try:
+ self.client.task_instances.succeed(
+ id=self.id,
+ when=msg.end_date,
+ task_outlets=msg.task_outlets,
+ outlet_events=msg.outlet_events,
+ rendered_map_index=self._rendered_map_index,
+ )
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.CONFLICT and _is_already_in_target_state(
+ e, msg.state
+ ):
+ log.info(
+ "TI already in success state, treating as idempotent success",
+ ti_id=self.id,
+ )
+ else:
+ raise
elif isinstance(msg, RetryTask):
self._terminal_state = msg.state
self._task_end_time_monotonic = time.monotonic()
self._rendered_map_index = msg.rendered_map_index
- self.client.task_instances.retry(
- id=self.id,
- end_date=msg.end_date,
- rendered_map_index=self._rendered_map_index,
- )
+ try:
+ self.client.task_instances.retry(
+ id=self.id,
+ end_date=msg.end_date,
+ rendered_map_index=self._rendered_map_index,
+ )
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.CONFLICT and _is_already_in_target_state(
+ e, msg.state
+ ):
+ log.info(
+ "TI already in retry state, treating as idempotent success",
Review Comment:
This log message says "treating as idempotent success" even though the target state here is a retry transition. Consider using neutral phrasing such as "already in retry state, treating as idempotent no-op" to avoid implying the task succeeded when it actually moved to UP_FOR_RETRY.
```suggestion
"TI already in retry state, treating as idempotent no-op",
```
--
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] Fix task marked as failed despite successful completion [airflow]
Posted by "Vamsi-klu (via GitHub)" <gi...@apache.org>.
Vamsi-klu commented on PR #64023:
URL: https://github.com/apache/airflow/pull/64023#issuecomment-4102455736
@jscheffl @potiuk can i get feedback for this PR? 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] Fix task marked as failed despite successful completion [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk closed pull request #64023: Fix task marked as failed despite successful completion
URL: https://github.com/apache/airflow/pull/64023
--
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] Fix task marked as failed despite successful completion [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #64023:
URL: https://github.com/apache/airflow/pull/64023#discussion_r3066477355
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1262,22 +1283,44 @@ def _handle_request(self, msg: ToSupervisor, log: FilteringBoundLogger, req_id:
self._terminal_state = msg.state
self._task_end_time_monotonic = time.monotonic()
self._rendered_map_index = msg.rendered_map_index
- self.client.task_instances.succeed(
- id=self.id,
- when=msg.end_date,
- task_outlets=msg.task_outlets,
- outlet_events=msg.outlet_events,
- rendered_map_index=self._rendered_map_index,
- )
+ try:
+ self.client.task_instances.succeed(
+ id=self.id,
+ when=msg.end_date,
+ task_outlets=msg.task_outlets,
+ outlet_events=msg.outlet_events,
+ rendered_map_index=self._rendered_map_index,
+ )
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.CONFLICT and _is_already_in_target_state(
+ e, msg.state
+ ):
+ log.info(
+ "TI already in success state, treating as idempotent success",
+ ti_id=self.id,
+ )
+ else:
+ raise
elif isinstance(msg, RetryTask):
self._terminal_state = msg.state
self._task_end_time_monotonic = time.monotonic()
self._rendered_map_index = msg.rendered_map_index
- self.client.task_instances.retry(
- id=self.id,
- end_date=msg.end_date,
- rendered_map_index=self._rendered_map_index,
- )
+ try:
+ self.client.task_instances.retry(
+ id=self.id,
+ end_date=msg.end_date,
+ rendered_map_index=self._rendered_map_index,
+ )
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.CONFLICT and _is_already_in_target_state(
+ e, msg.state
+ ):
+ log.info(
+ "TI already in retry state, treating as idempotent success",
+ ti_id=self.id,
+ )
Review Comment:
The log message in the `RetryTask` conflict handler says \"treating as idempotent success\", which is misleading for a retry transition. Update the message to reflect a retry/idempotent state update (e.g., \"idempotent retry\" or \"idempotent state already applied\") so operators can accurately interpret logs.
##########
task-sdk/tests/task_sdk/execution_time/test_supervisor.py:
##########
@@ -2702,6 +2703,211 @@ def test_handle_requests_api_server_error(self, watched_subprocess, mocker):
"detail": error.response.json(),
}
+ def test_succeed_task_idempotent_on_409_already_success(self, watched_subprocess, mocker):
+ """Test that a 409 with previous_state=success is treated as idempotent success for SucceedTask."""
+ watched_subprocess, read_socket = watched_subprocess
+
+ error = ServerResponseError(
+ message="Conflict",
+ request=httpx.Request("PATCH", "http://test"),
+ response=httpx.Response(
+ 409,
+ json={
+ "detail": {
+ "reason": "invalid_state",
+ "message": "TI was not in the running state so it cannot be updated",
+ "previous_state": "success",
+ }
+ },
+ ),
+ )
+ watched_subprocess.client.task_instances.succeed = mocker.Mock(side_effect=error)
+
+ generator = watched_subprocess.handle_requests(log=mocker.Mock())
+ next(generator)
+
+ msg = SucceedTask(end_date=timezone.parse("2024-10-31T12:00:00Z"))
+ req_frame = _RequestFrame(id=randint(1, 2**32 - 1), body=msg.model_dump())
+ generator.send(req_frame)
+
+ # Read response — should be an ack (no error), not an error response
+ read_socket.settimeout(0.1)
+ frame_len = int.from_bytes(read_socket.recv(4), "big")
+ response_bytes = read_socket.recv(frame_len)
+ frame = msgspec.msgpack.Decoder(_ResponseFrame).decode(response_bytes)
Review Comment:
These new tests assume `socket.recv(n)` returns exactly `n` bytes, but `recv` can legally return fewer bytes than requested. This can make the tests flaky under different OS/CI timing; consider using a small helper that loops until exactly 4 bytes (length prefix) and then exactly `frame_len` bytes are read. Also, a 0.1s socket timeout is tight for CI—raising it (or using a retry loop) would further reduce intermittent failures.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -448,6 +461,13 @@ def _handle_fail_fast_for_dag(ti: TI, dag_id: str, session: SessionDep, dag_bag:
_stop_remaining_tasks(task_instance=ti, task_teardown_map=task_teardown_map, session=session)
+def _get_requested_state(ti_patch_payload: TIStateUpdate) -> TaskInstanceState | None:
+ """Extract the requested terminal state from a TI state update payload."""
Review Comment:
The docstring says \"requested terminal state\" but the implementation also accepts `TIRetryStatePayload` (which is not a terminal state). Either adjust the docstring to say \"requested state\" (or similar), or narrow the accepted payload types to terminal-only to keep the documentation accurate.
--
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] Fix task marked as failed despite successful completion [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #64023:
URL: https://github.com/apache/airflow/pull/64023#issuecomment-4331426172
@Vamsi-klu This draft PR has been inactive for 21 days since the last triage comment and no response from the author. Closing to keep the queue clean.
You are welcome to reopen this PR when you resume work, or to open a new one addressing the issues previously raised. There is no rush — take your time.
---
_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._
--
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