You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "anishgirianish (via GitHub)" <gi...@apache.org> on 2026/01/21 04:52:10 UTC
[PR] Fix running task marked as failed on Celery broker redelivery #58441 [airflow]
anishgirianish opened a new pull request, #60855:
URL: https://github.com/apache/airflow/pull/60855
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
## Summary
- Handle broker redelivery gracefully when a task is already running on another worker
- Detect 409 "already running" response from API and raise `TaskAlreadyRunningError`
- Celery executor catches this and raises `Ignore()` to prevent state reporting
closes #58441
---
* 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` or `{issue_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60855:
URL: https://github.com/apache/airflow/pull/60855#issuecomment-4060118073
@eladkal sure will do
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2791189749
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2088,6 +2102,10 @@ def supervise(
final_state=process.final_state,
)
return exit_code
+ except TaskAlreadyRunningError:
+ # Let the executor handle this (e.g., Celery will ignore it)
+ log.info("Exiting due to broker redelivery", task_instance_id=str(ti.id))
Review Comment:
Thanks for catching that. Replaced all broker/redelivery references in the task-sdk with executor-agnostic language:
- Log: "Task is already running elsewhere, exiting."
- Comment: "Let the executor handle this."
- TaskAlreadyRunningError docstring: "Raised when a task is already running on another worker."
The celery provider still uses "redelivered message" in its own log since that's the appropriate context there.
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #60855:
URL: https://github.com/apache/airflow/pull/60855#issuecomment-4117504227
@jscheffl would you like to proceed with this PR? We need to make a decision so that https://github.com/apache/airflow/pull/64052 would be included in the up coming provider wave
--
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] Raise ``TaskAlreadyRunningError`` when starting an already-running task instance [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl merged PR #60855:
URL: https://github.com/apache/airflow/pull/60855
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2970332754
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -189,12 +189,18 @@ def on_celery_worker_ready(*args, **kwargs):
# and deserialization for us
@app.task(name="execute_workload")
def execute_workload(input: str) -> None:
+ from celery.exceptions import Ignore
from pydantic import TypeAdapter
from airflow.configuration import conf
from airflow.executors import workloads
from airflow.sdk.execution_time.supervisor import supervise
+ try:
+ from airflow.sdk.exceptions import TaskAlreadyRunningError
+ except ImportError:
+ TaskAlreadyRunningError = None # type: ignore[misc,assignment]
+
Review Comment:
@eladkal Created a seprate pr: https://github.com/apache/airflow/pull/64052 for provider changes. 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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #60855:
URL: https://github.com/apache/airflow/pull/60855#issuecomment-4043179786
@anishgirianish 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: **Merge conflicts**: This PR has merge conflicts with the `main` branch. Your branch is 823 commits behind `main`. Please rebase your branch (`git fetch origin && git rebase origin/main`), resolve the conflicts, and push again. See [contributing quick start](https://github.com/apache/airflow/blob/main/contributing-docs/03a_contributors_quick_start_beginners.rst).
> **Note:** Your branch is **823 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.
- 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. 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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2970223461
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2107,6 +2107,10 @@ def supervise(
final_state=process.final_state,
)
return exit_code
+ except TaskAlreadyRunningError:
+ # Let the executor handle this
+ log.info("Task is already running elsewhere, exiting", task_instance_id=str(ti.id))
+ raise
Review Comment:
Yep, we do not need this besides for informational reasons. Removed it, thank 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
Re: [PR] Raise ``TaskAlreadyRunningError`` when starting an already-running task instance [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on PR #60855:
URL: https://github.com/apache/airflow/pull/60855#issuecomment-4129884162
@vatsrahul1001 This is tagged mileston 3.1.9, will you consider this for 3.2.0 as well?
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #60855:
URL: https://github.com/apache/airflow/pull/60855#issuecomment-4060113880
@anishgirianish can you please rebase and resolve conflicts?
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60855:
URL: https://github.com/apache/airflow/pull/60855#issuecomment-4102333215
> @anishgirianish can you please rebase and resolve conflicts?
Hi @eladkal , I hav rebased the branch and resolved the conflicts and pushed, would like to request you for the re-review. thank 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
Re: [PR] Fix running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2969527074
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -189,12 +189,18 @@ def on_celery_worker_ready(*args, **kwargs):
# and deserialization for us
@app.task(name="execute_workload")
def execute_workload(input: str) -> None:
+ from celery.exceptions import Ignore
from pydantic import TypeAdapter
from airflow.configuration import conf
from airflow.executors import workloads
from airflow.sdk.execution_time.supervisor import supervise
+ try:
+ from airflow.sdk.exceptions import TaskAlreadyRunningError
+ except ImportError:
+ TaskAlreadyRunningError = None # type: ignore[misc,assignment]
+
Review Comment:
This is a bug fix I think this can go to 3.1.9 and maybe we should consider expediting 3.1.9 but regardless if we are having 3.1.9 or not the marker can be for >=3.1.9
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2791189749
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2088,6 +2102,10 @@ def supervise(
final_state=process.final_state,
)
return exit_code
+ except TaskAlreadyRunningError:
+ # Let the executor handle this (e.g., Celery will ignore it)
+ log.info("Exiting due to broker redelivery", task_instance_id=str(ti.id))
Review Comment:
Thanks for catching that. Replaced all broker/redelivery references in the task-sdk with executor-agnostic language:
- Log: "Task is already running elsewhere, exiting."
- Comment: "Let the executor handle this."
- TaskAlreadyRunningError docstring: "Raised when a task is already running on another worker."
The celery provider still uses "redelivered message" in its own log since that's the appropriate context there.
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2985562216
##########
task-sdk/tests/task_sdk/execution_time/test_supervisor.py:
##########
@@ -731,40 +731,6 @@ def mock_monotonic():
"task_instance_id": str(ti.id),
} in captured_logs
- def test_supervisor_handles_already_running_task(self):
- """Test that Supervisor prevents starting a Task Instance that is already running."""
- ti = TaskInstance(
- id=uuid7(), task_id="b", dag_id="c", run_id="d", try_number=1, dag_version_id=uuid7()
- )
-
- # Mock API Server response indicating the TI is already running
- # The API Server would return a 409 Conflict status code if the TI is not
- # in a "queued" state.
- def handle_request(request: httpx.Request) -> httpx.Response:
- if request.url.path == f"/task-instances/{ti.id}/run":
- return httpx.Response(
- 409,
- json={
- "reason": "invalid_state",
- "message": "TI was not in a state where it could be marked as running",
- "previous_state": "running",
- },
- )
-
- return httpx.Response(status_code=204)
-
- client = make_client(transport=httpx.MockTransport(handle_request))
-
- with pytest.raises(ServerResponseError, match="Server returned error") as err:
- ActivitySubprocess.start(dag_rel_path=os.devnull, bundle_info=FAKE_BUNDLE, what=ti, client=client)
-
- assert err.value.response.status_code == 409
- assert err.value.detail == {
- "reason": "invalid_state",
- "message": "TI was not in a state where it could be marked as running",
- "previous_state": "running",
- }
-
@pytest.mark.parametrize("captured_logs", [logging.ERROR], indirect=True, ids=["log_level=error"])
def test_state_conflict_on_heartbeat(self, captured_logs, monkeypatch, mocker, make_ti_context_dict):
Review Comment:
added test
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2791199602
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -159,15 +165,21 @@ def execute_workload(input: str) -> None:
base_url = f"http://localhost:8080{base_url}"
default_execution_api_server = f"{base_url.rstrip('/')}/execution/"
- supervise(
- # This is the "wrong" ti type, but it duck types the same. TODO: Create a protocol for this.
- ti=workload.ti, # type: ignore[arg-type]
- dag_rel_path=workload.dag_rel_path,
- bundle_info=workload.bundle_info,
- token=workload.token,
- server=conf.get("core", "execution_api_server_url", fallback=default_execution_api_server),
- log_path=workload.log_path,
- )
+ try:
+ supervise(
+ # This is the "wrong" ti type, but it duck types the same. TODO: Create a protocol for this.
+ ti=workload.ti, # type: ignore[arg-type]
+ dag_rel_path=workload.dag_rel_path,
+ bundle_info=workload.bundle_info,
+ token=workload.token,
+ server=conf.get("core", "execution_api_server_url", fallback=default_execution_api_server),
+ log_path=workload.log_path,
+ )
+ except Exception as e:
+ if TaskAlreadyRunningError is not None and isinstance(e, TaskAlreadyRunningError):
+ log.info("[%s] Task already running elsewhere, ignoring redelivered message", celery_task_id)
Review Comment:
Thanks for the question. This is new for Airflow 3. In 2.x, the scheduler only processed executor failure events for QUEUED tasks, so redelivery didn't affect running tasks. In 3.x, it also handles RUNNING so a redelivered message hitting a 409 gets recorded as FAILURE by Celery, and the scheduler marks the still-running task as failed. Ignore() prevents that.
Added a comment explaining this. Verified locally with a forced Redis redelivery — without the fix, the task gets marked as failed; with it, it completes successfully.
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -159,15 +165,21 @@ def execute_workload(input: str) -> None:
base_url = f"http://localhost:8080{base_url}"
default_execution_api_server = f"{base_url.rstrip('/')}/execution/"
- supervise(
- # This is the "wrong" ti type, but it duck types the same. TODO: Create a protocol for this.
- ti=workload.ti, # type: ignore[arg-type]
- dag_rel_path=workload.dag_rel_path,
- bundle_info=workload.bundle_info,
- token=workload.token,
- server=conf.get("core", "execution_api_server_url", fallback=default_execution_api_server),
- log_path=workload.log_path,
- )
+ try:
+ supervise(
+ # This is the "wrong" ti type, but it duck types the same. TODO: Create a protocol for this.
+ ti=workload.ti, # type: ignore[arg-type]
+ dag_rel_path=workload.dag_rel_path,
+ bundle_info=workload.bundle_info,
+ token=workload.token,
+ server=conf.get("core", "execution_api_server_url", fallback=default_execution_api_server),
+ log_path=workload.log_path,
+ )
+ except Exception as e:
+ if TaskAlreadyRunningError is not None and isinstance(e, TaskAlreadyRunningError):
+ log.info("[%s] Task already running elsewhere, ignoring redelivered message", celery_task_id)
Review Comment:
Thanks for the question. This is new for Airflow 3. In 2.x, the scheduler only processed executor failure events for QUEUED tasks, so redelivery didn't affect running tasks. In 3.x, it also handles RUNNING so a redelivered message hitting a 409 gets recorded as FAILURE by Celery, and the scheduler marks the still-running task as failed. Ignore() prevents that.
Added a comment explaining this. Verified locally with a forced Redis redelivery without the fix, the task gets marked as failed; with it, it completes successfully.
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2969504190
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -189,12 +189,18 @@ def on_celery_worker_ready(*args, **kwargs):
# and deserialization for us
@app.task(name="execute_workload")
def execute_workload(input: str) -> None:
+ from celery.exceptions import Ignore
from pydantic import TypeAdapter
from airflow.configuration import conf
from airflow.executors import workloads
from airflow.sdk.execution_time.supervisor import supervise
+ try:
+ from airflow.sdk.exceptions import TaskAlreadyRunningError
+ except ImportError:
+ TaskAlreadyRunningError = None # type: ignore[misc,assignment]
+
Review Comment:
If we know which version this fix is being released (can this go into 3.2.0?) then we should rather use the `AIRFLOW_V_3_2_PLUS` pattern such that in future it is easier to clean-up and it marks it clear which core version is needed for a feature.
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2107,6 +2107,10 @@ def supervise(
final_state=process.final_state,
)
return exit_code
+ except TaskAlreadyRunningError:
+ # Let the executor handle this
+ log.info("Task is already running elsewhere, exiting", task_instance_id=str(ti.id))
+ raise
Review Comment:
THe only purpose of catching and re-raising is posting an information? Do we need this?
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on PR #60855:
URL: https://github.com/apache/airflow/pull/60855#issuecomment-4121020450
> @jscheffl would you like to proceed with this PR? We need to make a decision so that #64052 would be included in the up coming provider wave
I know there are still ongoing problems but still would like to have more feedback fro other contributors on this. Feeling myself un-comfortable to merge and thus would rather risk to keep the problem open before merging too early. I think CoPilit responses are ok if being ignored.
@kaxil WDYT? Can we ad-hoc merge to get Celery fixes into provider wave?
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2985565830
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -216,7 +216,18 @@ def start(self, id: uuid.UUID, pid: int, when: datetime) -> TIRunContext:
"""Tell the API server that this TI has started running."""
body = TIEnterRunningPayload(pid=pid, hostname=get_hostname(), unixname=getuser(), start_date=when)
- resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ try:
+ resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.CONFLICT:
+ detail = e.detail
+ if (
+ isinstance(detail, dict)
+ and detail.get("reason") == "invalid_state"
+ and detail.get("previous_state") == "running"
+ ):
+ raise TaskAlreadyRunningError(f"Task {id} is already running") from e
+ raise
Review Comment:
seprate pr
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2985568123
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -216,7 +216,18 @@ def start(self, id: uuid.UUID, pid: int, when: datetime) -> TIRunContext:
"""Tell the API server that this TI has started running."""
body = TIEnterRunningPayload(pid=pid, hostname=get_hostname(), unixname=getuser(), start_date=when)
- resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ try:
+ resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.CONFLICT:
+ detail = e.detail
+ if (
+ isinstance(detail, dict)
+ and detail.get("reason") == "invalid_state"
+ and detail.get("previous_state") == "running"
+ ):
+ raise TaskAlreadyRunningError(f"Task {id} is already running") from e
Review Comment:
added test
--
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] Raise ``TaskAlreadyRunningError`` when starting an already-running task instance [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on PR #60855:
URL: https://github.com/apache/airflow/pull/60855#issuecomment-4131648915
> #58441
Yes, as this is a bug fix, we should include this in 3.2.0rc1 rc
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2732024592
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2088,6 +2102,10 @@ def supervise(
final_state=process.final_state,
)
return exit_code
+ except TaskAlreadyRunningError:
+ # Let the executor handle this (e.g., Celery will ignore it)
+ log.info("Exiting due to broker redelivery", task_instance_id=str(ti.id))
Review Comment:
"broker" as a concept only exist for Celery, so this log message doesn't make sense for other executors.
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1007,9 +1007,23 @@ def _on_child_started(
ti_context = self.client.task_instances.start(ti.id, self.pid, start_date)
self._should_retry = ti_context.should_retry
self._last_successful_heartbeat = time.monotonic()
- except Exception:
+ except Exception as e:
# On any error kill that subprocess!
self.kill(signal.SIGKILL)
+
+ # Handle broker redelivery: task already running on another worker
+ if isinstance(e, ServerResponseError) and e.response.status_code == 409:
+ # FastAPI wraps HTTPException detail in {"detail": {...}}
+ detail = e.detail
+ if isinstance(detail, dict) and "detail" in detail:
+ detail = detail["detail"]
+ if (
+ isinstance(detail, dict)
+ and detail.get("reason") == "invalid_state"
+ and detail.get("previous_state") == "running"
+ ):
Review Comment:
This is rather fragile and not a good idea to write things this way.
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60855:
URL: https://github.com/apache/airflow/pull/60855#issuecomment-3902784673
@ashb Thanks for the review, I've pushed updates addressing all your feedback. Quick summary:
Why the "ignore" behaviour: In Airflow 3.x, the scheduler now processes executor events for RUNNING tasks (not just QUEUED as in 2.x). When Celery redelivers a message for an already-running task, the worker hits a 409 from the API server. Without this fix, Celery records it as a FAILURE event, and the scheduler marks the still-running task as failed. The fix lets the supervisor detect the 409 (task already running) and exit gracefully instead of propagating a failure.
Changes made:
- Moved 409 handling into TaskInstanceOperations.start() (no more fragile dict parsing)
- Replaced all broker/Celery-specific language with executor-agnostic terms
Ready for another look when you have a chance.
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2969527074
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -189,12 +189,18 @@ def on_celery_worker_ready(*args, **kwargs):
# and deserialization for us
@app.task(name="execute_workload")
def execute_workload(input: str) -> None:
+ from celery.exceptions import Ignore
from pydantic import TypeAdapter
from airflow.configuration import conf
from airflow.executors import workloads
from airflow.sdk.execution_time.supervisor import supervise
+ try:
+ from airflow.sdk.exceptions import TaskAlreadyRunningError
+ except ImportError:
+ TaskAlreadyRunningError = None # type: ignore[misc,assignment]
+
Review Comment:
This is a bug fix I think this can go to 3.1.9 and maybe we should consider expediting 3.1.9
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2981033364
##########
task-sdk/tests/task_sdk/execution_time/test_supervisor.py:
##########
@@ -731,40 +731,6 @@ def mock_monotonic():
"task_instance_id": str(ti.id),
} in captured_logs
- def test_supervisor_handles_already_running_task(self):
- """Test that Supervisor prevents starting a Task Instance that is already running."""
- ti = TaskInstance(
- id=uuid7(), task_id="b", dag_id="c", run_id="d", try_number=1, dag_version_id=uuid7()
- )
-
- # Mock API Server response indicating the TI is already running
- # The API Server would return a 409 Conflict status code if the TI is not
- # in a "queued" state.
- def handle_request(request: httpx.Request) -> httpx.Response:
- if request.url.path == f"/task-instances/{ti.id}/run":
- return httpx.Response(
- 409,
- json={
- "reason": "invalid_state",
- "message": "TI was not in a state where it could be marked as running",
- "previous_state": "running",
- },
- )
-
- return httpx.Response(status_code=204)
-
- client = make_client(transport=httpx.MockTransport(handle_request))
-
- with pytest.raises(ServerResponseError, match="Server returned error") as err:
- ActivitySubprocess.start(dag_rel_path=os.devnull, bundle_info=FAKE_BUNDLE, what=ti, client=client)
-
- assert err.value.response.status_code == 409
- assert err.value.detail == {
- "reason": "invalid_state",
- "message": "TI was not in a state where it could be marked as running",
- "previous_state": "running",
- }
-
@pytest.mark.parametrize("captured_logs", [logging.ERROR], indirect=True, ids=["log_level=error"])
def test_state_conflict_on_heartbeat(self, captured_logs, monkeypatch, mocker, make_ti_context_dict):
Review Comment:
The supervisor-level test for handling a 409 conflict when starting an already-running TI was removed, but there isn't a replacement assertion at the supervisor/celery boundary. Since the PR intent is to make broker redelivery of an already-running task not propagate as a failure, please add/update a test that exercises the supervisor path (e.g. `ActivitySubprocess.start`/`supervise`) with this scenario and asserts the expected non-failure behavior.
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -216,7 +216,18 @@ def start(self, id: uuid.UUID, pid: int, when: datetime) -> TIRunContext:
"""Tell the API server that this TI has started running."""
body = TIEnterRunningPayload(pid=pid, hostname=get_hostname(), unixname=getuser(), start_date=when)
- resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ try:
+ resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.CONFLICT:
+ detail = e.detail
+ if (
+ isinstance(detail, dict)
+ and detail.get("reason") == "invalid_state"
+ and detail.get("previous_state") == "running"
+ ):
+ raise TaskAlreadyRunningError(f"Task {id} is already running") from e
Review Comment:
`TaskInstanceOperations.start()` now raises `TaskAlreadyRunningError` on a 409/invalid_state(previous_state=running), but there are no callers in the repo that catch/handle this exception (searching for `TaskAlreadyRunningError` only finds this file and tests). In the Celery redelivery scenario this will still surface as a task failure to Celery/Scheduler unless the supervisor / celery worker path explicitly treats it as a non-error (e.g., exit cleanly without reporting failure).
```suggestion
# In this case the task is already running. Treat this as a benign race and
# attempt to return a TIRunContext instead of surfacing a failure.
structlog.get_logger(__name__).info(
"Task already running when starting; treating as no-op",
task_instance_id=str(id),
)
try:
return TIRunContext.model_validate_json(e.response.read())
except Exception:
# If we can't interpret the response body, fall back to raising the
# dedicated error to avoid masking unexpected API responses.
raise TaskAlreadyRunningError(f"Task {id} is already running") from e
```
##########
task-sdk/tests/task_sdk/api/test_client.py:
##########
@@ -333,6 +333,55 @@ def handle_request(request: httpx.Request) -> httpx.Response:
assert resp == ti_context
assert call_count == 3
+ def test_task_instance_start_already_running(self):
+ """Test that start() raises TaskAlreadyRunningError when TI is already running."""
+ from airflow.sdk.exceptions import TaskAlreadyRunningError
+
Review Comment:
This test introduces an import inside the test function body (`from airflow.sdk.exceptions import TaskAlreadyRunningError`). The codebase generally keeps imports at module scope; please move this import to the top of the file unless there's a specific reason it must be local.
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -216,7 +216,18 @@ def start(self, id: uuid.UUID, pid: int, when: datetime) -> TIRunContext:
"""Tell the API server that this TI has started running."""
body = TIEnterRunningPayload(pid=pid, hostname=get_hostname(), unixname=getuser(), start_date=when)
- resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ try:
+ resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.CONFLICT:
+ detail = e.detail
+ if (
+ isinstance(detail, dict)
+ and detail.get("reason") == "invalid_state"
+ and detail.get("previous_state") == "running"
+ ):
+ raise TaskAlreadyRunningError(f"Task {id} is already running") from e
+ raise
Review Comment:
PR description mentions that the Celery executor catches the "already running" case and raises `Ignore()` to prevent state reporting, but there are no corresponding changes in the celery provider/executor code (and no repo references to `Ignore`/`TaskAlreadyRunningError` outside task-sdk). Either update the PR description to match the actual scope, or add the missing executor-side handling so the reported bug is actually fixed end-to-end.
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -216,7 +216,18 @@ def start(self, id: uuid.UUID, pid: int, when: datetime) -> TIRunContext:
"""Tell the API server that this TI has started running."""
body = TIEnterRunningPayload(pid=pid, hostname=get_hostname(), unixname=getuser(), start_date=when)
- resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ try:
+ resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.CONFLICT:
+ detail = e.detail
+ if (
+ isinstance(detail, dict)
+ and detail.get("reason") == "invalid_state"
+ and detail.get("previous_state") == "running"
+ ):
+ raise TaskAlreadyRunningError(f"Task {id} is already running") from e
Review Comment:
The new `TaskAlreadyRunningError` message says "Task {id} is already running", but `id` here is a task *instance* id. Consider adjusting the wording to avoid confusion with a DAG task id (e.g. "Task instance {id}...").
```suggestion
raise TaskAlreadyRunningError(f"Task instance {id} is already running") from e
```
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2970240655
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -189,12 +189,18 @@ def on_celery_worker_ready(*args, **kwargs):
# and deserialization for us
@app.task(name="execute_workload")
def execute_workload(input: str) -> None:
+ from celery.exceptions import Ignore
from pydantic import TypeAdapter
from airflow.configuration import conf
from airflow.executors import workloads
from airflow.sdk.execution_time.supervisor import supervise
+ try:
+ from airflow.sdk.exceptions import TaskAlreadyRunningError
+ except ImportError:
+ TaskAlreadyRunningError = None # type: ignore[misc,assignment]
+
Review Comment:
added AIRFLOW_V_3_1_9_PLUS guard. thank 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
Re: [PR] Fix running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2985563919
##########
task-sdk/tests/task_sdk/api/test_client.py:
##########
@@ -333,6 +333,55 @@ def handle_request(request: httpx.Request) -> httpx.Response:
assert resp == ti_context
assert call_count == 3
+ def test_task_instance_start_already_running(self):
+ """Test that start() raises TaskAlreadyRunningError when TI is already running."""
+ from airflow.sdk.exceptions import TaskAlreadyRunningError
+
Review Comment:
updated
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2985569414
##########
task-sdk/tests/task_sdk/execution_time/test_supervisor.py:
##########
@@ -731,40 +731,6 @@ def mock_monotonic():
"task_instance_id": str(ti.id),
} in captured_logs
- def test_supervisor_handles_already_running_task(self):
Review Comment:
add test 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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2731589850
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -159,15 +165,21 @@ def execute_workload(input: str) -> None:
base_url = f"http://localhost:8080{base_url}"
default_execution_api_server = f"{base_url.rstrip('/')}/execution/"
- supervise(
- # This is the "wrong" ti type, but it duck types the same. TODO: Create a protocol for this.
- ti=workload.ti, # type: ignore[arg-type]
- dag_rel_path=workload.dag_rel_path,
- bundle_info=workload.bundle_info,
- token=workload.token,
- server=conf.get("core", "execution_api_server_url", fallback=default_execution_api_server),
- log_path=workload.log_path,
- )
+ try:
+ supervise(
+ # This is the "wrong" ti type, but it duck types the same. TODO: Create a protocol for this.
+ ti=workload.ti, # type: ignore[arg-type]
+ dag_rel_path=workload.dag_rel_path,
+ bundle_info=workload.bundle_info,
+ token=workload.token,
+ server=conf.get("core", "execution_api_server_url", fallback=default_execution_api_server),
+ log_path=workload.log_path,
+ )
+ except Exception as e:
+ if TaskAlreadyRunningError is not None and isinstance(e, TaskAlreadyRunningError):
+ log.info("[%s] Task already running elsewhere, ignoring redelivered message", celery_task_id)
Review Comment:
Is this a new feature or did something like this exist in Airflow 2.x?
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2791187801
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1007,9 +1007,23 @@ def _on_child_started(
ti_context = self.client.task_instances.start(ti.id, self.pid, start_date)
self._should_retry = ti_context.should_retry
self._last_successful_heartbeat = time.monotonic()
- except Exception:
+ except Exception as e:
# On any error kill that subprocess!
self.kill(signal.SIGKILL)
+
+ # Handle broker redelivery: task already running on another worker
+ if isinstance(e, ServerResponseError) and e.response.status_code == 409:
+ # FastAPI wraps HTTPException detail in {"detail": {...}}
+ detail = e.detail
+ if isinstance(detail, dict) and "detail" in detail:
+ detail = detail["detail"]
+ if (
+ isinstance(detail, dict)
+ and detail.get("reason") == "invalid_state"
+ and detail.get("previous_state") == "running"
+ ):
Review Comment:
Thanks for the feedback. Moved the 409 handling into TaskInstanceOperations.start() in client.py, following the same pattern as DagRunOperations.trigger(). The supervisor now catches TaskAlreadyRunningError with no dict parsing at all.
Also fixed ServerResponseError.from_response() to properly handle dict-type detail values previously, it fell through to a raw msgspec.json.decode fallback, which caused double-wrapping and required callers to unwrap e.detail["detail"]. That's no longer needed.
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2970241880
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -189,12 +189,18 @@ def on_celery_worker_ready(*args, **kwargs):
# and deserialization for us
@app.task(name="execute_workload")
def execute_workload(input: str) -> None:
+ from celery.exceptions import Ignore
from pydantic import TypeAdapter
from airflow.configuration import conf
from airflow.executors import workloads
from airflow.sdk.execution_time.supervisor import supervise
+ try:
+ from airflow.sdk.exceptions import TaskAlreadyRunningError
+ except ImportError:
+ TaskAlreadyRunningError = None # type: ignore[misc,assignment]
+
Review Comment:
@anishgirianish is it possible to keep core changes in this PR and move all providers/ changes to a new PR?
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2970289143
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -189,12 +189,18 @@ def on_celery_worker_ready(*args, **kwargs):
# and deserialization for us
@app.task(name="execute_workload")
def execute_workload(input: str) -> None:
+ from celery.exceptions import Ignore
from pydantic import TypeAdapter
from airflow.configuration import conf
from airflow.executors import workloads
from airflow.sdk.execution_time.supervisor import supervise
+ try:
+ from airflow.sdk.exceptions import TaskAlreadyRunningError
+ except ImportError:
+ TaskAlreadyRunningError = None # type: ignore[misc,assignment]
+
Review Comment:
Sure on 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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2970338236
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -189,12 +189,18 @@ def on_celery_worker_ready(*args, **kwargs):
# and deserialization for us
@app.task(name="execute_workload")
def execute_workload(input: str) -> None:
+ from celery.exceptions import Ignore
from pydantic import TypeAdapter
from airflow.configuration import conf
from airflow.executors import workloads
from airflow.sdk.execution_time.supervisor import supervise
+ try:
+ from airflow.sdk.exceptions import TaskAlreadyRunningError
+ except ImportError:
+ TaskAlreadyRunningError = None # type: ignore[misc,assignment]
+
Review Comment:
Great, 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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2985452026
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -216,7 +216,18 @@ def start(self, id: uuid.UUID, pid: int, when: datetime) -> TIRunContext:
"""Tell the API server that this TI has started running."""
body = TIEnterRunningPayload(pid=pid, hostname=get_hostname(), unixname=getuser(), start_date=when)
- resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ try:
+ resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.CONFLICT:
+ detail = e.detail
+ if (
+ isinstance(detail, dict)
+ and detail.get("reason") == "invalid_state"
+ and detail.get("previous_state") == "running"
+ ):
+ raise TaskAlreadyRunningError(f"Task {id} is already running") from e
Review Comment:
`TaskAlreadyRunningError` inherits from `AirflowException`, not `ServerResponseError`. Anyone currently catching `ServerResponseError` from `start()` to detect this 409 case will stop catching it. The deleted supervisor test (`test_supervisor_handles_already_running_task`) was doing exactly that -- catching `ServerResponseError`.
This is probably the right behavior change, but should there be a supervisor-level test that verifies `TaskAlreadyRunningError` propagates through `ActivitySubprocess.start()` and kills the subprocess? The deleted test covered that integration path; the new `test_client.py` tests only cover the `TaskInstanceOperations.start()` layer.
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -216,7 +216,18 @@ def start(self, id: uuid.UUID, pid: int, when: datetime) -> TIRunContext:
"""Tell the API server that this TI has started running."""
body = TIEnterRunningPayload(pid=pid, hostname=get_hostname(), unixname=getuser(), start_date=when)
- resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ try:
+ resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.CONFLICT:
+ detail = e.detail
+ if (
+ isinstance(detail, dict)
+ and detail.get("reason") == "invalid_state"
+ and detail.get("previous_state") == "running"
+ ):
+ raise TaskAlreadyRunningError(f"Task {id} is already running") from e
Review Comment:
Nit: `id` here is a task *instance* UUID, not a task ID. `f"Task instance {id} is already running"` would be clearer in logs and error messages.
##########
task-sdk/tests/task_sdk/execution_time/test_supervisor.py:
##########
@@ -731,40 +731,6 @@ def mock_monotonic():
"task_instance_id": str(ti.id),
} in captured_logs
- def test_supervisor_handles_already_running_task(self):
Review Comment:
The deleted test verified the supervisor's end-to-end behavior when the API returns 409 for an already-running task (subprocess gets killed, error propagates). The new client-level tests cover the exception translation, but nothing tests the supervisor's reaction to `TaskAlreadyRunningError` specifically.
Consider adding a replacement that asserts `ActivitySubprocess.start()` raises `TaskAlreadyRunningError` (not `ServerResponseError`) and that the child process is killed.
##########
task-sdk/tests/task_sdk/api/test_client.py:
##########
@@ -333,6 +333,55 @@ def handle_request(request: httpx.Request) -> httpx.Response:
assert resp == ti_context
assert call_count == 3
+ def test_task_instance_start_already_running(self):
+ """Test that start() raises TaskAlreadyRunningError when TI is already running."""
+ from airflow.sdk.exceptions import TaskAlreadyRunningError
Review Comment:
`TaskAlreadyRunningError` is imported inside the test body. Other exceptions like `ErrorType` are imported at the top of the file (line 50). Move this up there for consistency.
--
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 running task marked as failed on Celery broker redelivery #58441 [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60855:
URL: https://github.com/apache/airflow/pull/60855#discussion_r2985568419
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -216,7 +216,18 @@ def start(self, id: uuid.UUID, pid: int, when: datetime) -> TIRunContext:
"""Tell the API server that this TI has started running."""
body = TIEnterRunningPayload(pid=pid, hostname=get_hostname(), unixname=getuser(), start_date=when)
- resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ try:
+ resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.CONFLICT:
+ detail = e.detail
+ if (
+ isinstance(detail, dict)
+ and detail.get("reason") == "invalid_state"
+ and detail.get("previous_state") == "running"
+ ):
+ raise TaskAlreadyRunningError(f"Task {id} is already running") from e
Review Comment:
updated
##########
task-sdk/tests/task_sdk/api/test_client.py:
##########
@@ -333,6 +333,55 @@ def handle_request(request: httpx.Request) -> httpx.Response:
assert resp == ti_context
assert call_count == 3
+ def test_task_instance_start_already_running(self):
+ """Test that start() raises TaskAlreadyRunningError when TI is already running."""
+ from airflow.sdk.exceptions import TaskAlreadyRunningError
Review Comment:
updated 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