Repository navigation
airflow workers idle after redis connection errors #631
Description
we are using celery executors and had redis connection issues last night. after redis was online again the workers could reconnect but did not execute any task anymore. all the tasks which should be run timed out waiting in the queue.
worker restart solves the problem immediately
log before restart attached, thats the end of the log (copied at 9 am one today)
...
[2025-05-25 18:05:00,485: INFO/ForkPoolWorker-15] Filling up the DagBag from /stackable/app/git/current/stages/int/apps/platform-core/dags/deadbeef.py
[2025-05-25 18:05:00,648: INFO/ForkPoolWorker-15] Running <TaskInstance: platform-deadbeef.deadbeef scheduled__2025-05-25T16:00:00+00:00 [queued]> on host airflow-worker-default-1.airflow-worker-default.mesh-platform-core.svc.cluster.local
[2025-05-25 18:05:02,045: WARNING/ForkPoolWorker-15] empty cryptography key - values will not be stored encrypted.
[2025-05-25 18:05:02,047: INFO/ForkPoolWorker-15] Using connection ID 'hpo_default' for task execution.
[2025-05-25 18:05:02,047: INFO/ForkPoolWorker-15] AWS Connection (conn_id='hpo_default', conn_type='aws') credentials retrieved from login and password.
[2025-05-25 18:05:02,605: INFO/ForkPoolWorker-15] Task airflow.providers.celery.executors.celery_executor_utils.execute_command[7a457fda-8875-4b9d-add7-8b4711e2e43d] succeeded in 2.17235898389481s: None
[2025-05-25 18:12:38,465: WARNING/MainProcess] consumer: Connection to broker lost. Trying to re-establish the connection...
Traceback (most recent call last):
File "/stackable/app/lib64/python3.9/site-packages/celery/worker/consumer/consumer.py", line 340, in start
blueprint.start(self)
File "/stackable/app/lib64/python3.9/site-packages/celery/bootsteps.py", line 116, in start
step.start(parent)
File "/stackable/app/lib64/python3.9/site-packages/celery/worker/consumer/consumer.py", line 746, in start
c.loop(*c.loop_args())
File "/stackable/app/lib64/python3.9/site-packages/celery/worker/loops.py", line 97, in asynloop
next(loop)
File "/stackable/app/lib64/python3.9/site-packages/kombu/asynchronous/hub.py", line 373, in create_loop
cb(*cbargs)
File "/stackable/app/lib64/python3.9/site-packages/kombu/transport/redis.py", line 1344, in on_readable
self.cycle.on_readable(fileno)
File "/stackable/app/lib64/python3.9/site-packages/kombu/transport/redis.py", line 569, in on_readable
chan.handlers[type]()
File "/stackable/app/lib64/python3.9/site-packages/kombu/transport/redis.py", line 913, in _receive
ret.append(self._receive_one(c))
File "/stackable/app/lib64/python3.9/site-packages/kombu/transport/redis.py", line 923, in _receive_one
response = c.parse_response()
File "/stackable/app/lib64/python3.9/site-packages/redis/client.py", line 837, in parse_response
response = self._execute(conn, try_read)
File "/stackable/app/lib64/python3.9/site-packages/redis/client.py", line 813, in _execute
return conn.retry.call_with_retry(
File "/stackable/app/lib64/python3.9/site-packages/redis/retry.py", line 49, in call_with_retry
fail(error)
File "/stackable/app/lib64/python3.9/site-packages/redis/client.py", line 815, in <lambda>
lambda error: self._disconnect_raise_connect(conn, error),
File "/stackable/app/lib64/python3.9/site-packages/redis/client.py", line 802, in _disconnect_raise_connect
raise error
File "/stackable/app/lib64/python3.9/site-packages/redis/retry.py", line 46, in call_with_retry
return do()
File "/stackable/app/lib64/python3.9/site-packages/redis/client.py", line 814, in <lambda>
lambda: command(*args, **kwargs),
File "/stackable/app/lib64/python3.9/site-packages/redis/client.py", line 835, in try_read
return conn.read_response(disconnect_on_error=False, push_request=True)
File "/stackable/app/lib64/python3.9/site-packages/redis/connection.py", line 512, in read_response
response = self._parser.read_response(disable_decoding=disable_decoding)
File "/stackable/app/lib64/python3.9/site-packages/redis/_parsers/resp2.py", line 15, in read_response
result = self._read_response(disable_decoding=disable_decoding)
File "/stackable/app/lib64/python3.9/site-packages/redis/_parsers/resp2.py", line 25, in _read_response
raw = self._buffer.readline()
File "/stackable/app/lib64/python3.9/site-packages/redis/_parsers/socket.py", line 115, in readline
self._read_from_socket()
File "/stackable/app/lib64/python3.9/site-packages/redis/_parsers/socket.py", line 68, in _read_from_socket
raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
redis.exceptions.ConnectionError: Connection closed by server.
[2025-05-25 18:12:41,476: WARNING/MainProcess] /stackable/app/lib64/python3.9/site-packages/celery/worker/consumer/consumer.py:391: CPendingDeprecationWarning:
In Celery 5.1 we introduced an optional breaking change which
on connection loss cancels all currently executed tasks with late acknowledgement enabled.
These tasks cannot be acknowledged as the connection is gone, and the tasks are automatically redelivered
back to the queue. You can enable this behavior using the worker_cancel_long_running_tasks_on_connection_loss
setting. In Celery 5.1 it is set to False by default. The setting will be set to True by default in Celery 6.0.
warnings.warn(CANCEL_TASKS_BY_DEFAULT, CPendingDeprecationWarning)
[2025-05-25 18:12:41,477: INFO/MainProcess] Temporarily reducing the prefetch count to 12 to avoid over-fetching since 4 tasks are currently being processed.
The prefetch count will be gradually restored to 16 as the tasks complete processing.
[2025-05-25 18:12:41,494: WARNING/MainProcess] /stackable/app/lib64/python3.9/site-packages/celery/worker/consumer/consumer.py:508: CPendingDeprecationWarning: The broker_connection_retry configuration setting will no longer determine
whether broker connection retries are made during startup in Celery 6.0 and above.
If you wish to retain the existing behavior for retrying connections on startup,
you should set broker_connection_retry_on_startup to True.
warnings.warn(
[2025-05-25 18:12:41,497: ERROR/MainProcess] consumer: Cannot connect to redis://airflow-redis-haproxy:6379//: Connection closed by server..
Trying again in 2.00 seconds... (1/100)
[2025-05-25 18:12:43,502: ERROR/MainProcess] consumer: Cannot connect to redis://airflow-redis-haproxy:6379//: Connection closed by server..
Trying again in 4.00 seconds... (2/100)
[2025-05-25 18:12:47,508: ERROR/MainProcess] consumer: Cannot connect to redis://airflow-redis-haproxy:6379//: Connection closed by server..
Trying again in 6.00 seconds... (3/100)
[2025-05-25 18:12:53,523: INFO/MainProcess] Connected to redis://airflow-redis-haproxy:6379//
[2025-05-25 18:12:53,524: WARNING/MainProcess] /stackable/app/lib64/python3.9/site-packages/celery/worker/consumer/consumer.py:508: CPendingDeprecationWarning: The broker_connection_retry configuration setting will no longer determine
whether broker connection retries are made during startup in Celery 6.0 and above.
If you wish to retain the existing behavior for retrying connections on startup,
you should set broker_connection_retry_on_startup to True.
warnings.warn(
[2025-05-25 18:12:53,530: INFO/MainProcess] mingle: searching for neighbors
[2025-05-25 18:12:54,540: INFO/MainProcess] mingle: all alone
[2025-05-25 18:12:59,561: INFO/MainProcess] missed heartbeat from celery@airflow-worker-default-0
[2025-05-25 18:30:46,097: WARNING/ForkPoolWorker-16] empty cryptography key - values will not be stored encrypted.
[2025-05-25 18:30:46,099: INFO/ForkPoolWorker-16] Using connection ID 'hpo_default' for task execution.
[2025-05-25 18:30:46,100: INFO/ForkPoolWorker-16] AWS Connection (conn_id='hpo_default', conn_type='aws') credentials retrieved from login and password.
[2025-05-25 18:30:46,962: INFO/ForkPoolWorker-16] Task airflow.providers.celery.executors.celery_executor_utils.execute_command[0a9c0d57-fa15-41dd-87f3-257f05a5c04f] succeeded in 1846.0727213749196s: None
[2025-05-25 18:30:47,489: WARNING/ForkPoolWorker-14] empty cryptography key - values will not be stored encrypted.
[2025-05-25 18:30:47,492: INFO/ForkPoolWorker-14] Using connection ID 'hpo_default' for task execution.
[2025-05-25 18:30:47,492: INFO/ForkPoolWorker-14] AWS Connection (conn_id='hpo_default', conn_type='aws') credentials retrieved from login and password.
[2025-05-25 18:30:48,363: INFO/ForkPoolWorker-14] Task airflow.providers.celery.executors.celery_executor_utils.execute_command[17fafc80-ab1b-45d9-b46b-ca71f46a3f98] succeeded in 1847.4771845629439s: None
Activity
adwk67 commentedon May 26, 2025
I will try and reproduce this, but that may be difficult. If you are on Airflow 2.9.x it looks as though it could be related to this issue: apache/airflow#41359, celery/kombu#2096.
adwk67 commentedon May 26, 2025
If you are on airflow 2.9.3, then we package version 3.7.2 of the celery provider. The default timeout is quite low, 1s. You could try changing this value if convenient: see https://airflow.apache.org/docs/apache-airflow-providers-celery/3.7.2/configurations-ref.html#operation-timeout.
maxgruber19 commentedon May 26, 2025
thanks for the quick reply, I'll test the timeout config. oh and yes, we're on airflow 2.9.3.
I'm just curios that after 3 retries everything looks good to go but then nothing happens after the redis connection is reestablished
adwk67 commentedon May 26, 2025
I've tried to reproduce this without success. Starting jobs without Redis being available, or shutting it down while jobs are scheduled always gives me a pretty immediate connection error:
But no timeouts (maybe because the DAG does basically nothing and completes very quickly).
maxgruber19 commentedon May 27, 2025
I see, it's hard for me to reproduce that as well. No matter how I break redis, it's always "selfhealing" when it's back online
I'll check again, will keep you updated
maxgruber19 commentedon Aug 26, 2025
@adwk67 we were analyzing that issue further because it happenend again and came to the conclusion that it seems to be related to the redis master switching over. That could be when the node hosting the master is being shutdown or has an unexpected failure.
The error is not occurring when the switch of the master role is done pretty fast. As soon as it takes longer (>10s) to complete, the workers - sometimes only one in the statefulset - say that they're reconnected but they dont read from the queue.
Every task is queued until the worker statefulset is restarted manually or they run into the dag timeout. The workers will never recover themself.
More details on our redis setup: We use the redis-ha (https://artifacthub.io/packages/helm/dandydev-charts/redis-ha) in version 4.33.2 with three replicas and haproxy enabled. We didn't make any configurations so it's pretty default. The redis log during a switchover that is leading to the airflow issue is attached below
worker log at the same time, there is no other log line until we restart the pod. no task is read from that worker ever again.
maxgruber19 commentedon Aug 26, 2025
apache/airflow#32484 seems to be a similar problem, at least I'm not the only one feeling pain. Reason might be as mentioned there: celery/celery#8030
maxgruber19 commentedon Aug 26, 2025
I guess that was a perfect hit. After digging deeper together with @realjump we were not able to reproduce the issue after invoking
pip install --upgrade celery==5.5.3before airflow startup.That issue already led to production outages on our side, it would be nice if you can add that to your image in the future. I would also love to have a patched image if that's possible for you. Otherwise I have to build a custom image by myself which is possible of course but i guess thats a major improvement in point of stability.
adwk67 commentedon Aug 27, 2025
Thanks for the extra info, @maxgruber19! We currently ship these versions of celery:
Are you planning on upgrading from 2.9.3 anytime soon?
We will be bumping to 3.0.5 (at least) for the next release (25.11), so I'll open an issue to bump celery to at least 5.5.3 for that release.
adwk67 commentedon Aug 27, 2025
Can you confirm if the issue also persists with 5.5.2?
maxgruber19 commentedon Aug 27, 2025
We built another image ourself (we do that for all of your images either way because of further customization) with celery upgraded to 5.5.3 and it seems like no simulated outage of redis leads to the issue anymore. The issue is fixed since 5.5.0,
We will stay on 2.9.3 until its running out of LTS, changing to airflow 3 is a change we can't do in near future because of capacities 😄
I'll keep an eye on it but I'm very optimistic. I'll close the issue for now, if you need it for further investigation or fixing / patching on your side feel free to open again