AMQP 1.0 - consumer QPID Proton python blocked by RabbitMQ on a durable Queue.

[sylvainrichardcool]

Hi all,
I try to consume message with AMQP 1.0 protocol via a QPID proton python receiver.
If the target is a Durable Queue (exampleQueue), RabbitMQ block the receiver.

I use the Exchange (AMQP default)

RabbitMQ log

2024-05-02 10:13:44.181972+00:00 [info] <0.22973.0> accepting AMQP connection <0.22973.0> (172.1.1.41:58070 -> 172.1.1.41:5676)
2024-05-02 10:13:44.183984+00:00 [info] <0.22973.0> AMQP 1.0 connection <0.22973.0>: user 'test' authenticated
2024-05-02 10:13:44.186736+00:00 [info] <0.22973.0> AMQP 1.0 connection <0.22973.0>: user 'test' authenticated and granted access to vhost '/'
2024-05-02 10:13:44.187536+00:00 [error] <0.23002.0> Channel error on connection <0.22982.0> (172.1.1.41:58070 -> 172.1.1.41:5676 (0), vhost: '/', user: 'test'), channel 2:
**2024-05-02 10:13:44.187536+00:00 [error] <0.23002.0> operation queue.declare caused a channel exception precondition_failed: inequivalent arg 'durable' for queue 'exampleQueue' in vhost '/': received 'false' but current is 'true'**
2024-05-02 10:13:44.187729+00:00 [warning] <0.22979.0> Closing session for connection <0.22973.0>:
2024-05-02 10:13:44.187729+00:00 [warning] <0.22979.0> {'v1_0.error',{symbol,<<"amqp:precondition-failed">>},
**2024-05-02 10:13:44.187729+00:00 [warning] <0.22979.0>               {utf8,<<"PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'exampleQueue' in vhost '/': received 'false' but current is 'true'">>},**
2024-05-02 10:13:44.187729+00:00 [warning] <0.22979.0>               undefined}

Log in QPID Proton Python client

   -I- Receiver receiver1 is created
[0x56361fe22f30]: SASL:FRAME:  -> SASL
[0x56361fe22f30]: SASL:FRAME:  <- SASL
[0x56361fe22f30]: AMQP:FRAME:0 <- @sasl-mechanisms(64) [sasl-server-mechanisms=@<symbol>[:ANONYMOUS, :PLAIN, :AMQPLAIN]]
[0x56361fe22f30]: AMQP:FRAME:0 -> @sasl-init(65) [mechanism=:PLAIN, initial-response=b"\x00test\x00test"]
[0x56361fe22f30]: AMQP:FRAME:0 <- @sasl-outcome(68) [code=0x0]
[0x56361fe22f30]: AMQP:FRAME:  -> AMQP
[0x56361fe22f30]: AMQP:FRAME:0 -> @open(16) [container-id="Client Python QIP recv - sylvain", hostname="myhost.com", channel-max=0x7fff]
[0x56361fe22f30]: AMQP:FRAME:0 -> @begin(17) [next-outgoing-id=0x0, incoming-window=0x7fffffff, outgoing-window=0x7fffffff]
[0x56361fe22f30]: AMQP:FRAME:0 -> @attach(18) [name="receiver1", handle=0x0, role=true, snd-settle-mode=0x2, rcv-settle-mode=0x0, source=@source(40) [address="exampleQueue", durable=0x0, timeout=0x0, dynamic=false, capabilities=:topic], target=@target(41) [durable=0x0, timeout=0x0, dynamic=false], initial-delivery-count=0x0, max-message-size=0x0]
[0x56361fe22f30]: AMQP:FRAME:0 -> @flow(19) [incoming-window=0x7fffffff, next-outgoing-id=0x0, outgoing-window=0x7fffffff, handle=0x0, delivery-count=0x0, link-credit=0xa, drain=false]
[0x56361fe22f30]: AMQP:FRAME:  <- AMQP
[0x56361fe22f30]: AMQP:FRAME:0 <- @open(16) [container-id="rabbit@myhost", channel-max=0x7fff, idle-time-out=0xea60, properties={:"cluster_name"="rabbit@myhost", :copyright="Copyright (c) 2007-2024 Broadcom Inc and/or its subsidiaries", :information="Licensed under the MPL 2.0. Website: https://rabbitmq.com", :platform="Erlang/OTP 26.2.4", :product="RabbitMQ", :version="3.13.1"}]
   -I- Connection <proton._endpoints.Connection 0x7f2c9c8e2eb8 ~ 0x56361fe1ef00> is open
[0x56361fe22f30]: AMQP:FRAME:0 <- @begin(17) [remote-channel=0x0, next-outgoing-id=0x0, incoming-window=0xffff, outgoing-window=0xffff, handle-max=0xffffffff]
[0x56361fe22f30]: AMQP:FRAME:0 <- @end(23) [error=@error(29) [condition=:"amqp:precondition-failed", description="PRECONDITION_FAILED - inequivalent arg \x27durable\x27 for queue \x27exampleQueue\x27 in vhost \x27/\x27: received \x27false\x27 but current is \x27true\x27"]]
   -E- Session closed with error

The QPID proton python receiver works with non durable queue. (exampleQueuend)

What configuration is missing to connect a durable queue?

Sylvain

[lukebakken]

Hello, thanks for using RabbitMQ.

We can't assist you without the following information:

• RabbitMQ and Erlang versions
• RabbitMQ configuration
• QPid Python code that we can clone and run to reproduce this issue.

It seems like this issue should be easy to reproduce, so PLEASE provide code we can run to see it. Thank you.

[sylvainrichardcool]

Hi Luke.

please find all the information in attachment files.

receiver.txt

You can clone from https://github.com/sylvainrichardcool/amqp1.0-qpid-proton-python

Test each Queue.
python3 receiver-amqp_rabbitmq.py --address amqp://myhost.com:5676/exampleQueue -m 1 #durable Queue
python3 receiver-amqp_rabbitmq.py --address amqp://myhost.com:5676/exampleQueuend -m 1 #not durable Queue
python3 receiver-amqp_rabbitmq.py --address amqp://myhost.com:5676/exampleStream -m 1 #durable Stream

Thanks.

Sylvain

[kjnilsson]

You need to use the /amq/queue/QUEUE_NAME target format. See: https://github.com/rabbitmq/rabbitmq-server/tree/v3.13.2/deps/rabbitmq_amqp1_0#routing-and-addressing

[sylvainrichardcool]

Hi Karl
as you must have seen the prefix is ​​added in the script.

[lukebakken]

• AMQP sending to RabbitMQ with 1.0 Plugin doesn't respect durable setting smallrye/smallrye-reactive-messaging#174
• Durability conflicts when publishing messages discussions#11

@kjnilsson and @ansd both know a lot more about AMQP 1.0 than I do, and perhaps they could chime in.

UPDATE: see @mkuratczyk's response.

[sylvainrichardcool]

Hi all
according to the AMQP 1.0 spec it should be possible to consume Durable queues.

Is this a problem with the "AMQP 1.0" plugin or a rabbitMQ specificity?

[mkuratczyk]

No, it's a problem with your client app (perhaps the client lib you're using, but more likely some parameters you need to provide to it). It's trying to declare a queue that already exists and it's trying to do that with different arguments (the queue exists as durable, you are trying to declare it as non-durable). You can use https://github.com/rabbitmq/omq to play with AMQP 1.0, for example:

> rabbitmqadmin declare queue name=durable arguments='{"durable": true}'
queue declared

> rabbitmqctl -q list_queues name durable messages
Listing queues for vhost / ...
name	durable	messages
durable	true	0

> omq amqp -t /amq/queue/durable -T /amq/queue/durable -C 10 -D 5
[output removed for brevity]

> rabbitmqctl -q list_queues name durable messages
Listing queues for vhost / ...
name	durable	messages
durable	true	5

In this case omq published 10 messages and consumed 5 of them without problems.

[lukebakken]

Thank you @mkuratczyk

@sylvainrichardcool - please see these search results, I'm sure the solution can be found - https://www.google.com/search?q=qpid+proton+python+rabbitmq+durable+queue

https://www.rabbitmq.com/client-libraries/interoperability

[lukebakken]

@sylvainrichardcool my google search lead to this article:

https://stackoverflow.com/q/42210809

Please see this code which allows your receiver to use a durable queue:
lukebakken/rabbitmq-server-11148-11151@f79ebbd

[sylvainrichardcool]

Thank you Luke.
it's works.

Note : for Steam queue. By default the consumer get only the new messages in the stream.
I need to add this config to read all the stream messages.

class ReceiverFilterOptions(ReceiverOption):
    def apply(self, receiver):
        receiver.source.filter.put_dict({symbol("rabbitmq:stream-offset-spec"): "first"})

    def on_start(self, event):
        receiver_options = [ DurableSubscription(), ReceiverFilterOptions() ] #read from first msg in the stream