Change default value for Durable flag on Queue creation for pidbox (RabbitMQ deprecated feature transient_nonexcl_queues)

[Osmoskour]

In 4.0, RabbitMQ added transient_nonexcl_queues to deprecated features:
https://www.rabbitmq.com/release-information/deprecated-features-list
https://www.rabbitmq.com/blog/2021/08/21/4.0-deprecation-announcements#removal-of-transient-non-exclusive-queues

rabbitmq  | 2025-01-27 09:40:55.277393+00:00 [warning] <0.780.0> Deprecated features: `transient_nonexcl_queues`: Feature `transient_nonexcl_queues` is deprecated.
rabbitmq  | 2025-01-27 09:40:55.277393+00:00 [warning] <0.780.0> By default, this feature can still be used for now.
rabbitmq  | 2025-01-27 09:40:55.277393+00:00 [warning] <0.780.0> Its use will not be permitted by default in a future minor RabbitMQ version and the feature will be removed from a future major RabbitMQ version; actual versions to be determined.

By default, the durable flag for new queue is False. This is hardcoded for pidbox.py:

kombu/kombu/pidbox.py

Lines 395 to 405 in a0175b0

	def _get_exchange(self, namespace, type):
	return Exchange(self.exchange_fmt % namespace,
	type=type,
	durable=False,
	delivery_mode='transient')
	def _get_reply_exchange(self, namespace):
	return Exchange(self.reply_exchange_fmt % namespace,
	type='direct',
	durable=False,
	delivery_mode='transient')

kombu/kombu/pidbox.py

Lines 233 to 257 in a0175b0

	def get_reply_queue(self):
	oid = self.oid
	return Queue(
	f'{oid}.{self.reply_exchange.name}',
	exchange=self.reply_exchange,
	routing_key=oid,
	durable=False,
	auto_delete=True,
	expires=self.reply_queue_expires,
	message_ttl=self.reply_queue_ttl,
	)
	@cached_property
	def reply_queue(self):
	return self.get_reply_queue()
	def get_queue(self, hostname):
	return Queue(
	f'{hostname}.{self.namespace}.pidbox',
	exchange=self.exchange,
	durable=False,
	auto_delete=True,
	expires=self.queue_expires,
	message_ttl=self.queue_ttl,
	)

kombu/kombu/pidbox.py

Lines 272 to 274 in a0175b0

	exchange = Exchange(exchange, exchange_type='direct',
	delivery_mode='transient',
	durable=False)

And not hardcoded for qpid.py but setting it to True can be a sensible default:

kombu/kombu/transport/qpid.py

Line 591 in a0175b0

def queue_declare(self, queue, passive=False, durable=False,

kombu/kombu/transport/qpid.py

Line 711 in a0175b0

def exchange_declare(self, exchange='', type='direct', durable=False,

By setting those to True, can we avoid the deprecated feature warning?

[bfdacosta]

Greetings. Have the same issue in my env.
bump!

[trevorboydsmith]

TLDR: i concur with this solution. (FYI "transient_nonexcl_queues" feature has been soft deprecated since 2021 https://www.rabbitmq.com/release-information/deprecated-features-list)

further analysis (see below):

the software requirements from my point of view are:

1. change celery/kombu to not depend on a rabbitmq feature that will be removed
2. keep the celery functionality the same (i.e. celery inspect/control/ping via pidbox)

the proposed change fulfills both of the software requirements by:

1. changing default durable to True
2. keeps the celery functionality all the same

the only downside is that all celery inspect/control/ping command/control is done via queues that are durable instead of in-memory so there is a slight performance hit (not a big deal IMO). and the change is very low risk.

p.s. IMO it does not require a new celery config field/parameter because the rabbitmq feature is going away so no way to turn back on once it goes away.

[auvipy]

thanks for the report. we will consider this for kombu 5.7. anyone is welcome to contribute. if not we will try to handle them by ourselves.

[auvipy]

@Nusnus may be this is related to quorum Q?

[Nusnus]

@auvipy

@Nusnus may be this is related to quorum Q?

From a brief review it seems it’s not related to QQ, so no :)

[trianglesis]

Both options:

• delivery_mode='persistent'
• durable=True

... should be configurable, so the developer can change the behaviour.

Covers queues that are both non-durable and non-exclusive, this combination should be avoided. 
Use [durable queues or non-durable exclusive queues](https://www.rabbitmq.com/docs/queues). 
[Queue TTL](https://www.rabbitmq.com/docs/ttl#queue-ttl) can be used for cleanup of unused durable queues.

Doc: https://www.rabbitmq.com/release-information/deprecated-features-list

I'm trying both options now hardcoded into persistent and True here: https://github.com/trianglesis/kombu_no_transient_nonexcl_queues.git

[auvipy]

Do you have plans to contribute to this issue?

[trianglesis]

Do you have plans to contribute to this issue?

I'm trying to make it work now in the dev environment.
On success, I will create a PR; if not, I'm afraid I have to downgrade and move along.

[auvipy]

If you can share your draft changes, it would be also great

[trianglesis]

Here is a pretty straightforward approach I've made.

According to RabbitMQ recommendations: durable queues are the recommended option: DOC

I've changed all booleans to True for the durable arg everywhere:

• https://github.com/trianglesis/celery
• https://github.com/trianglesis/kombu_no_transient_nonexcl_queues

It did not break anything, at least with the test task.

However, I went an extra mile in my setup, making a celery stop/start routine with an additional method to manually declare and bind everything using pika with a custom opened channel:

def queue_declare(self, queue, exchange, routing_key):
    self.channel.exchange_declare(
        exchange=exchange,
        exchange_type='direct',
        durable=CeleryConstants.CONF_DURABLE
    )
    self.channel.queue_declare(
        queue=queue,
        durable=CeleryConstants.CONF_DURABLE,
        exclusive=False,
        arguments=self.args_d
    )
    self.channel.queue_bind(
        exchange=exchange,
        queue=queue,
        routing_key=routing_key
    )

    # Default binding: only if worker_direct=True
    # self.channel.queue_bind(exchange='C.dq2', queue=queue, routing_key=routing_key)

    return self.channel.confirm_delivery()

So I cannot be sure, at the current level, that the basic celery setup is not affected by those silly changes of booleans all over.

N.B. There is also a problem with celery worker creation transparency. It's not so obvious how the celery worker summons a queue and assigns it to the default exchange. Sometimes it helps when you create an Exchange and Queues explicitly.

default_exchange = Exchange(
    'octopus',
    type='direct',
    durable=CeleryConstants.CONF_DURABLE,
    delivery_mode='persistent',
)

workers_list = [   ]
for _worker in workers:
    workers_list.append(
        Queue(f"{_worker}.dq2",
              alias=_worker,
              routing_key=f"{_worker}.dq2",
              durable=CeleryConstants.CONF_DURABLE,
              exclusive=False,
              exchange=default_exchange,
              arguments={'expires':1000 * 60 * 60 * 12, 'max_priority':10},
              )
    )
app.conf.task_queues = workers_list

Additionally, any non-default option in the following directives breaks the basic setup, and it's also not obvious why. What if I wanted to create an exchange and default queue by myself? But I'm still seeing cellery named exchanges, and default queue where all my messages are sent.

task_default_queue
task_default_exchange
task_default_routing_key

The change on durability option seems simple, but it raises a lot of concerns with related things.

UPD: Works fine with a simple systemd start script and without an extra mile.

Custom exchange octopus

All workers and queues fromthe typical systemd setup script are up and running:

Everything is durable=True

Fake tasks are running:

[auvipy]

what is your take on this now?