galaxyproject / galaxyproject/galaxy

Use AMQP duable queues

Open
#22,641 0 comments 5 reactions 1 assignee Claimed by @mvdbeek View on GitHub
Dominant language
Python
Stars
1.9k
Forks
1.2k
Avg merge
2d 16h
Merged PRs (30d)
169

Description

It looks like for celery workers we use transient queues. After a RabbitMQ 4.2->4.3 upgrade yesterday my celery workers were failing with:

```pytb
[2026-05-05 09:29:07,189: WARNING/MainProcess] consumer: Connection to broker lost. Trying to re-establish the connection...
Traceback (most recent call last):
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/celery/worker/consumer/consumer.py", line 346, in start
blueprint.start(self)
~~~~~~~~~~~~~~~^^^^^^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/celery/bootsteps.py", line 116, in start
step.start(parent)
~~~~~~~~~~^^^^^^^^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/celery/worker/consumer/mingle.py", line 37, in start
self.sync(c)
~~~~~~~~~^^^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/celery/worker/consumer/mingle.py", line 41, in sync
replies = self.send_hello(c)
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/celery/worker/consumer/mingle.py", line 54, in send_hello
replies = inspect.hello(c.hostname, our_revoked._data) or {}
~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/celery/app/control.py", line 389, in hello
return self._request('hello', from_node=from_node, revoked=revoked)
~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/celery/app/control.py", line 106, in _request
return self._prepare(self.app.control.broadcast(
~~~~~~~~~~~~~~~~~~~~~~~~~~^
command,
^^^^^^^^
...<6 lines>...
pattern=self.pattern, matcher=self.matcher,
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
))
^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/celery/app/control.py", line 785, in broadcast
return self.mailbox(conn)._broadcast(
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
command, arguments, destination, reply, timeout,
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
limit, callback, channel=channel,
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/kombu/pidbox.py", line 347, in _broadcast
self._publish(command, arguments, destination=destination,
~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
reply_ticket=reply_ticket,
^^^^^^^^^^^^^^^^^^^^^^^^^^
...<3 lines>...
pattern=pattern,
^^^^^^^^^^^^^^^^
matcher=matcher)
^^^^^^^^^^^^^^^^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/kombu/pidbox.py", line 309, in _publish
maybe_declare(self.reply_queue(chan))
~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/kombu/common.py", line 113, in maybe_declare
return _maybe_declare(entity, channel)
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/kombu/common.py", line 155, in _maybe_declare
entity.declare(channel=channel)
~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/kombu/entity.py", line 617, in declare
self._create_queue(nowait=nowait, channel=channel)
~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/kombu/entity.py", line 626, in _create_queue
self.queue_declare(nowait=nowait, passive=False, channel=channel)
~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/kombu/entity.py", line 655, in queue_declare
ret = channel.queue_declare(
queue=self.name,
...<5 lines>...
nowait=nowait,
)
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/amqp/channel.py", line 1162, in queue_declare
return queue_declare_ok_t(*self.wait(
~~~~~~~~~^
spec.Queue.DeclareOk, returns_tuple=True,
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
))
^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/amqp/abstract_channel.py", line 99, in wait
self.connection.drain_events(timeout=timeout)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/amqp/connection.py", line 526, in drain_events
while not self.blocking_read(timeout):
~~~~~~~~~~~~~~~~~~^^^^^^^^^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/amqp/connection.py", line 532, in blocking_read
return self.on_inbound_frame(frame)
~~~~~~~~~~~~~~~~~~~~~^^^^^^^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/amqp/method_framing.py", line 53, in on_frame
callback(channel, method_sig, buf, None)
~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/amqp/connection.py", line 538, in on_inbound_method
return self.channels[channel_id].dispatch_method(
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
method_sig, payload, content,
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/amqp/abstract_channel.py", line 156, in dispatch_method
listener(*args)
~~~~~~~~^^^^^^^
File "/cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/amqp/connection.py", line 668, in _on_close
raise error_for_code(reply_code, reply_text,
(class_id, method_id), ConnectionError)
amqp.exceptions.InternalError: Queue.declare: (541) INTERNAL_ERROR - Feature `transient_nonexcl_queues` is deprecated.
By default, this feature is not permitted anymore.
The feature will be removed from a future major RabbitMQ version, regardless of the configuration; actual version to be determined.
To...
[2026-05-05 09:29:07,195: WARNING/MainProcess] /cvmfs/main.galaxyproject.org/venv/lib/python3.13/site-packages/celery/worker/consumer/consumer.py:397: 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.
```

This can be worked around for now by setting the following in `rabbitmq.conf`:

```ini
deprecated_features.permit.transient_nonexcl_queues = true
```

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.