Repository: apache/airflow
Pull request: Add `task_acks_late` configuration to Celery Executor
<!--
 Licensed to the Apache Software Foundation (ASF) under one
 or more contributor license agreements.  See the NOTICE file
 distributed with this work for additional information
 regarding copyright ownership.  The ASF licenses this file
 to you under the Apache License, Version 2.0 (the
 "License"); you may not use this file except in compliance
 with the License.  You may obtain a copy of the License at

   http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing,
 software distributed under the License is distributed on an
 "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 KIND, either express or implied.  See the License for the
 specific language governing permissions and limitations
 under the License.
 -->

<!--
Thank you for contributing! 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 committers for the review!

In case of an existing issue, reference it using one of the following:



How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->

Celery [task_acks_late](https://docs.celeryq.dev/en/stable/userguide/configuration.html#task-acks-late) is currently hard-coded to be deactivated. However, this causes issues when tasks exceed the `visibility_timeout`.

This PR, adds a `task_acks_late` parameter to the Celery configuration. Default values remain unchanged, but with the new config options users can at least try, if this setting solves there issue if they encounter the `'Task Instance Not Running' FAILED: Task is in the running state` error message.

It is an enhanced re-implementation of #31829 and adresses #16163.

closes #16163
related: #31829


This is not covered by tests (yet). I can add tests but would be happy to receive pointers to e.g. existing test for environment settings, that I could adjust.

<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).


Problem statement:
Confusing log for long running tasks: "dependency 'Task Instance Not Running' FAILED: Task is in the running state"
**Apache Airflow version**: 1.10.* / 2.0.* / 2.1.*


**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): Any

**Environment**:

- **Cloud provider or hardware configuration**: Any
- **OS** (e.g. from /etc/os-release): Any
- **Kernel** (e.g. `uname -a`): Any
- **Install tools**: Any
- **Others**: N/A

**What happened**:

This line in the TaskInstance log is very misleading. It seems to happen for tasks that take longer than one hour. When users are waiting for tasks to finish and see this in the log, they often get confused. They may think something is wrong with their task or with Airflow. In fact, this line is harmless. It's simply saying "the TaskInstance is already running so it cannot be run again". 

```
{taskinstance.py:874} INFO - Dependencies not met for <TaskInstance: ... [running]>, dependency 'Task Instance Not Running' FAILED: Task is in the running state
{taskinstance.py:874} INFO - Dependencies not met for <TaskInstance: ... [running]>, dependency 'Task Instance State' FAILED: Task is in the 'running' state which is not a valid state for execution. The task must be cleared in order to be run.
```

**What you expected to happen**:

The confusion is unnecessary. This line should be silenced in the log. Or it should log something clearer.

**How to reproduce it**:

Any task that takes more than an hour to run has this line in the log.

Hints from issue discussion:
I bumped into this issue and I misunderstood until now 😅 
thanks for pointing out it's okay to wait
After some more investigation, it's very likely we see this log appearing an hour after a long running task started because of the default `visibility_timeout` setting in Celery. This code in default_celery.py sets `visibility_timeout` to 21600 only if the broker_url starts with redis or sqs. In our case we are using redis sentinels so it's still redis although the URL starts with `sentinel`. Therefore the `visibility_timeout` is left at 3600 which is the default according to [celery documentation](https://docs.celeryproject.org/en/v4.4.7/getting-started/brokers/redis.html#visibility-timeout). The weird thing is that after I tried to manually change `visibility_timeout` to a very large integer in airflow.cfg, the same log still showed up exactly an hour after a task started. So it seems changing `visibility_timeout` in this case does not make any difference. Not sure if anyone experienced the same.

@david30907d maybe try changing `visibility_timeout` to a large number in your setup and see if it still happens after an hour. If it stops for you, it means `visibility_timeout` is probably the cause. There may be something wrong in our own setup causing changing `visibility_timeout` not to take effect.

```python
def _broker_supports_visibility_timeout(url):
    return url.startswith("redis://") or url.startswith("sqs://")


log = logging.getLogger(__name__)

broker_url = conf.get('celery', 'BROKER_URL')

broker_transport_options = conf.getsection('celery_broker_transport_options') or {}
if 'visibility_timeout' not in broker_transport_options:
    if _broker_supports_visibility_timeout(broker_url):
        broker_transport_options['visibility_timeout'] = 21600
```
@yuqian90 thanks, very useful information. I'll give `celery_broker_transport_options` a shot!
Moved it to 2.2 now - I guess we have not addressed it yet in 2.1.1
In the case where the visibility timeout is reached, it's confusing that there is not a clear log line that the task has been killed for taking too long to complete.

(If that's indeed what is happening.)

@potiuk is it the case, that the Celery task is killed or is it simply no longer streaming logs into Airflow at that point?
Not sure. Needs investigation
@malthe That's not quite what is happening. My understanding is this:

- The task message on celery is picked up and the TI starts running
- After 1 hour (the celery visiblity timeout) celery makes the message visible again
- A second worker picks up the celery message
- When the second one runs, it gets this message.


So the first attempt is still running, and this message is the _second_ concurrent worker (but for the same try_number)  saying "I can't run this task, it's already running."
@ashb so regardless of whether the first worker is still running or defunct/dead, shouldn't the second worker be able to "take over" somehow?

Otherwise, what's the point in trying?
This is largely airflow and celery "fighting" over behaviour
Hi!
Just ran into the issue, having an "amqp://" starting broker url.
Will it be fixed in 2.3.0?

Commit message: Update airflow/providers/celery/provider.yaml

Patch to review:
diff --git a/airflow/providers/celery/executors/default_celery.py b/airflow/providers/celery/executors/default_celery.py
index eaaf0fd0b4f31..01b13b1c5a8af 100644
--- a/airflow/providers/celery/executors/default_celery.py
+++ b/airflow/providers/celery/executors/default_celery.py
@@ -72,7 +72,7 @@ def _broker_supports_visibility_timeout(url):
     "accept_content": ["json"],
     "event_serializer": "json",
     "worker_prefetch_multiplier": conf.getint("celery", "worker_prefetch_multiplier", fallback=1),
-    "task_acks_late": True,
+    "task_acks_late": conf.getboolean("celery", "task_acks_late", fallback=True),
     "task_default_queue": conf.get("operators", "DEFAULT_QUEUE"),
     "task_default_exchange": conf.get("operators", "DEFAULT_QUEUE"),
     "task_track_started": conf.getboolean("celery", "task_track_started", fallback=True),
diff --git a/airflow/providers/celery/provider.yaml b/airflow/providers/celery/provider.yaml
index 947dfcb6b1a01..80b7583a5ea62 100644
--- a/airflow/providers/celery/provider.yaml
+++ b/airflow/providers/celery/provider.yaml
@@ -276,6 +276,21 @@ config:
         type: float
         example: ~
         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:
+          'Task Instance Not Running' FAILED: Task is in the running state'
+          Setting task_acks_late to True will force Celery to wait until a task is finished before a
+          new task instance is assigned. This effectively overrides the visibility timeout.
+          See also:
+          https://docs.celeryq.dev/en/stable/reference/celery.app.task.html#celery.app.task.Task.acks_late
+        version_added: ~
+        type: boolean
+        example: "True"
+        default: "True"
       task_track_started:
         description: |
           Celery task will report its status as 'started' when the task is executed by a worker.
diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
index 08d2817eaee1d..8457f8403c489 100644
--- a/docs/spelling_wordlist.txt
+++ b/docs/spelling_wordlist.txt
@@ -7,6 +7,7 @@ Ack
 ack
 ackIds
 acknowledgement
+acks
 actionCard
 Acyclic
 acyclic
diff --git a/tests/providers/celery/executors/test_celery_executor.py b/tests/providers/celery/executors/test_celery_executor.py
index e6446c0f6fa6b..450312bd63198 100644
--- a/tests/providers/celery/executors/test_celery_executor.py
+++ b/tests/providers/celery/executors/test_celery_executor.py
@@ -358,3 +358,12 @@ def test_sentinel_kwargs_loaded_from_string():
     assert default_celery.DEFAULT_CELERY_CONFIG["broker_transport_options"]["sentinel_kwargs"] == {
         "service_name": "mymaster"
     }
+
+
+@conf_vars({("celery", "task_acks_late"): "False"})
+def test_celery_task_acks_late_loaded_from_string():
+    import importlib
+
+    # reload celery conf to apply the new config
+    importlib.reload(default_celery)
+    assert default_celery.DEFAULT_CELERY_CONFIG["task_acks_late"] is False


Code review comments:
diff_hunk: @@ -276,6 +276,21 @@ config:
         type: float
         example: ~
         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:
+          'Task Instance Not Running' FAILED: Task is in the running state'
+          Setting task_acks_late to True will force Celery to wait until a task is finished before a
+          new task instance is assigned. This effectively overrides the visibility timeout.
+          See also:
+          https://docs.celeryq.dev/en/stable/reference/celery.app.task.html#celery.app.task.Task.acks_late
+        version_added: ~
path: airflow/providers/celery/provider.yaml
text: @user1:
```suggestion
        version_added: 3.6.0
```

@user1:
@user2-awala ?

@user3:
yes we can use the next minor release for Celery, I just checked the other providers' configurations and found some version_added set in some of them