This is an automated email from the ASF dual-hosted git repository.
eladkal pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 5f213c70e2c Make the celery visibility_timeout redelivery docs
version-conditional (#75373)
5f213c70e2c is described below
commit 5f213c70e2c2f85aa7cd96249e38f3b8ddff3e8f
Author: iralp <[email protected]>
AuthorDate: Fri Oct 30 21:55:26 2026 -0700
Make the celery visibility_timeout redelivery docs version-conditional
(#75373)
* Make the celery visibility_timeout redelivery docs version-conditional
Three doc surfaces describe a broker redelivery as terminating the
still-running original task:
* ``[celery] task_acks_late``'s config description (and its generated
mirror in ``get_provider_info.py``), which also describes the failure
as surfacing only as "'Task Instance Not Running' FAILED: Task is in
the running state'"
* the runtime warning emitted by ``_broker_transport_options`` when no
``visibility_timeout`` is configured
* ``celery_executor.rst``, which asks for a visibility timeout above the
longest task ETA without saying what happens if it is not
That stopped being the only outcome in
``apache-airflow-providers-celery`` 3.18.0: ``celery_executor_utils``
raises ``Ignore()`` on ``TaskAlreadyRunningError``, so the redelivered
message is discarded and the original task keeps running. It needs
Airflow 3.1.9 or newer as well, because the 409 it relies on comes from
the Execution API -- the provider still gates on
``AIRFLOW_V_3_1_9_PLUS``, so the existing text remains correct below
that floor and is now scoped to those versions rather than replaced.
Measured on Airflow 3.3.2 with provider 3.24.0: a 301-second task under
a 60-second visibility timeout reached ``success``, and the duplicate
logged "Task already running elsewhere, ignoring redelivered message".
Neither doc surface named a version floor, and the whole provider doc
set had one bullet about visibility timeout with no explanation of the
mechanism, so this also adds a "Visibility timeout and long-running
tasks" section covering ``task_acks_late``'s interaction with
redelivery, both version floors, and SQS's 43200-second cap.
Follow-up to #62869, which added the warning text. It was accurate for
the provider versions available at the time; this change scopes it to
those versions.
Generated-by: Claude Opus 5
* Correct the version floor to 3.2.0 and attribute it to the Task SDK
Review feedback on #75373: the docs named Airflow 3.1.9, taken from the
provider's ``AIRFLOW_V_3_1_9_PLUS`` gate, but 3.1.9 was never released --
3.1.8 is the last 3.1.x on PyPI and #60855 shipped in 3.2.0. 3.2.0 is
therefore the lowest version a reader can actually be running, so all
four surfaces now say 3.2.0.
#60855 also landed in the Task SDK, not the Execution API: the 409 was
already there, and what changed is ``TaskInstanceOperations.start()``
translating it into ``TaskAlreadyRunningError``
(``task-sdk/src/airflow/sdk/api/client.py``). The rst now names that
exception instead of crediting the Execution API.
Bumping ``AIRFLOW_V_3_1_9_PLUS`` to ``AIRFLOW_V_3_2_PLUS`` is left to a
separate PR, as agreed in review.
Generated-by: Claude Opus 5
---
providers/celery/docs/celery_executor.rst | 29 +++++++++++++++++++++-
providers/celery/provider.yaml | 13 +++++++---
.../providers/celery/executors/default_celery.py | 4 ++-
.../airflow/providers/celery/get_provider_info.py | 2 +-
4 files changed, 41 insertions(+), 7 deletions(-)
diff --git a/providers/celery/docs/celery_executor.rst
b/providers/celery/docs/celery_executor.rst
index d175908549d..6b82484f1e0 100644
--- a/providers/celery/docs/celery_executor.rst
+++ b/providers/celery/docs/celery_executor.rst
@@ -82,12 +82,39 @@ Please note that you must have the ``flower`` python
library already installed o
Some caveats:
- Make sure to use a database backed result backend
-- Make sure to set a visibility timeout in
``[celery_broker_transport_options]`` that exceeds the ETA of your longest
running task
+- Make sure to set a visibility timeout in
``[celery_broker_transport_options]`` that exceeds the ETA of your longest
running task (see `Visibility timeout and long-running tasks`_)
- Make sure to specify the password for Redis Server in the
``[celery_broker_transport_options]`` section if you are using Redis Sentinel
as your broker and the Redis servers are password-protected
- Make sure to set umask in ``[worker_umask]`` to set permissions for newly
created files by workers.
- Tasks can consume resources. Make sure your worker has enough resources to
run ``worker_concurrency`` tasks
- Queue names are limited to 256 characters, but each broker backend might
have its own restrictions
+Visibility timeout and long-running tasks
+-----------------------------------------
+
+Brokers that redeliver unacknowledged messages -- Redis and Amazon SQS -- hide
a delivered message for
+``[celery_broker_transport_options] visibility_timeout`` seconds. Because
``[celery] task_acks_late``
+is enabled by default, a message is acknowledged only once its task finishes,
so any task that runs
+for longer than the visibility timeout is redelivered while it is still
running.
+
+Setting ``task_acks_late`` does not prevent this. For Redis and SQS it does
not override
+``visibility_timeout``: the broker redelivers the message regardless.
+
+Both deliveries carry the same Celery task id, so a failure reported for the
redelivery cannot be
+told apart from one reported for the original task.
+
+With ``apache-airflow-providers-celery`` 3.18.0 or newer running on Airflow
3.2.0 or newer, the
+redelivered message is detected and discarded. The worker logs ``Task already
running elsewhere,
+ignoring redelivered message``, and the original task runs to completion. Both
version floors are
+required: the provider recognizes the condition, but it relies on the
``TaskAlreadyRunningError``
+that the Task SDK raises in Airflow 3.2.0 and newer when a task instance is
already running.
+
+On earlier versions the redelivery terminates the still-running original task,
which surfaces only as
+``'Task Instance Not Running' FAILED: Task is in the running state``.
+
+Set ``visibility_timeout`` to exceed the ETA of your longest running task.
Note that Amazon SQS caps
+the visibility timeout at 43200 seconds (12 hours), so on SQS it cannot be
raised any further and
+tasks running for longer than 12 hours are always redelivered.
+
PostgreSQL result backends
--------------------------
diff --git a/providers/celery/provider.yaml b/providers/celery/provider.yaml
index 635e978db96..1f0b40ada61 100644
--- a/providers/celery/provider.yaml
+++ b/providers/celery/provider.yaml
@@ -385,10 +385,15 @@ config:
default: "1.0"
task_acks_late:
description: |
- If an Airflow task's execution time exceeds the visibility_timeout,
Celery will re-assign the
- task to a Celery worker, even if the original task is still running
successfully. The new task
- instance then runs concurrently with the original task and the
Airflow UI and logs only show an
- error message:
+ If an Airflow task's execution time exceeds the visibility_timeout,
the broker redelivers the
+ message to a Celery worker, even if the original task is still
running successfully. Both
+ deliveries carry the same Celery task id.
+ With apache-airflow-providers-celery 3.18.0 or newer running on
Airflow 3.2.0 or newer, the
+ redelivered message is recognized and discarded: the worker logs
+ 'Task already running elsewhere, ignoring redelivered message' and
the original task is left
+ running. Both version floors are required.
+ On earlier versions the redelivered message runs concurrently with
the original task and fails
+ it, and the Airflow UI and logs only show an error message:
'Task Instance Not Running' FAILED: Task is in the running state'
Setting task_acks_late to True acknowledges the task only after it
completes.
Note: for Redis and SQS brokers, task_acks_late does NOT override
visibility_timeout.
diff --git
a/providers/celery/src/airflow/providers/celery/executors/default_celery.py
b/providers/celery/src/airflow/providers/celery/executors/default_celery.py
index e57719f070e..76362f32ac7 100644
--- a/providers/celery/src/airflow/providers/celery/executors/default_celery.py
+++ b/providers/celery/src/airflow/providers/celery/executors/default_celery.py
@@ -90,7 +90,9 @@ def _broker_transport_options(broker_url: str, conf:
AirflowSDKConfigParser | An
log.warning(
"No visibility_timeout configured in
[celery_broker_transport_options]. "
"Using default of 86400 seconds (24 hours). Celery tasks
running longer than this "
- "will be redelivered by the broker, which terminates the
original task. "
+ "will be redelivered by the broker. On Airflow 3.2.0 and newer
the redelivered "
+ "message is discarded and the original task keeps running; on
earlier versions the "
+ "redelivery terminates the original task. "
"If you have long-running tasks, increase this value in your
Airflow configuration: "
"[celery_broker_transport_options] visibility_timeout =
<seconds>"
)
diff --git a/providers/celery/src/airflow/providers/celery/get_provider_info.py
b/providers/celery/src/airflow/providers/celery/get_provider_info.py
index f5ecc04ddb9..cbb914e55f4 100644
--- a/providers/celery/src/airflow/providers/celery/get_provider_info.py
+++ b/providers/celery/src/airflow/providers/celery/get_provider_info.py
@@ -240,7 +240,7 @@ def get_provider_info():
"default": "1.0",
},
"task_acks_late": {
- "description": "If an Airflow task's execution time
exceeds the visibility_timeout, Celery will re-assign the\ntask to a Celery
worker, even if the original task is still running successfully. The new
task\ninstance then runs concurrently with the original task and the Airflow UI
and logs only show an\nerror message:\n'Task Instance Not Running' FAILED: Task
is in the running state'\nSetting task_acks_late to True acknowledges the task
only after it completes.\nNo [...]
+ "description": "If an Airflow task's execution time
exceeds the visibility_timeout, the broker redelivers the\nmessage to a Celery
worker, even if the original task is still running successfully.
Both\ndeliveries carry the same Celery task id.\nWith
apache-airflow-providers-celery 3.18.0 or newer running on Airflow 3.2.0 or
newer, the\nredelivered message is recognized and discarded: the worker
logs\n'Task already running elsewhere, ignoring redelivered message' a [...]
"version_added": "3.6.0",
"type": "boolean",
"example": "True",
