Event exchange plugin not generating consumer.deleted events

[arshad01]

Describe the bug

Hello

We just upgraded from RabbitMQ version 3.9.24 to 3.13.4. After the upgrade, it seems that the event exchange plugin no longer generates the consumer.deleted event. There are some other events (e.g. binding.created etc.) that are also not getting fired. The consumer.deleted event is important for our use since it has the consumer tag available. Note that this is a single node Rocky8 system.

3.13.4

# rabbitmq-diagnostics consume_event_stream | grep consumer

{"event":"connection.created","user_who_performed_action":"System","node":"rabbit@my-host-h1","connected_at":1725854133894,"client_properties":["{<<\"product\">>,longstr,<<\"Pika Python Client Library\">>}","{<<\"platform\">>,longstr,<<\"Python 3.11.9\">>}","{<<\"capabilities\">>,table,[{<<\"authentication_failure_close\">>,bool,true},{<<\"basic.nack\">>,bool,true},{<<\"connection.blocked\">>,bool,true},{<<\"consumer_cancel_notify\">>,bool,true},{<<\"publisher_confirms\">>,bool,true}]}","{<<\"information\">>,longstr,<<\"See http://pika.rtfd.org\">>}","{<<\"version\">>,longstr,<<\"1.1.0\">>}"],"channel_max":2047,"frame_max":131072,"timeout":60,"vhost":"/","user":"System","protocol":"{0,9,1}","ssl_hash":"","ssl_cipher":"","ssl_key_exchange":"","ssl_protocol":"","auth_mechanism":"PLAIN","peer_cert_validity":"","peer_cert_issuer":"","peer_cert_subject":"","ssl":false,"peer_host":"{127,0,0,1}","host":"{127,0,0,1}","peer_port":45554,"port":5672,"name":"127.0.0.1:45554 -> 127.0.0.1:5672","pid":"<rabbit@my-host-h1.1725845209.7721.0>","type":"network","timestamp_in_ms":1725854133897}
{"event":"consumer.created","user_who_performed_action":"System","arguments":[],"prefetch_count":0,"queue":"test.session.abcd","vhost":"/","channel":"<rabbit@my-host-h1.1725845209.7730.0>","ack_required":false,"exclusive":false,"consumer_tag":"TEST_AMQP","timestamp_in_ms":1725854133908}
{"event":"connection.closed","client_properties":["{<<\"product\">>,longstr,<<\"Pika Python Client Library\">>}","{<<\"platform\">>,longstr,<<\"Python 3.11.9\">>}","{<<\"capabilities\">>,table,[{<<\"authentication_failure_close\">>,bool,true},{<<\"basic.nack\">>,bool,true},{<<\"connection.blocked\">>,bool,true},{<<\"consumer_cancel_notify\">>,bool,true},{<<\"publisher_confirms\">>,bool,true}]}","{<<\"information\">>,longstr,<<\"See http://pika.rtfd.org\">>}","{<<\"version\">>,longstr,<<\"1.1.0\">>}"],"node":"rabbit@my-host-h1","pid":"<rabbit@my-host-h1.1725845209.7721.0>","name":"127.0.0.1:45554 -> 127.0.0.1:5672","timestamp_in_ms":1725854138417}
{"event":"channel.closed","consumer_count":1,"user_who_performed_action":"System","pid":"<rabbit@my-host-h1.1725845209.7730.0>","timestamp_in_ms":1725854138430}

3.9.24

# rabbitmq-diagnostics consume_event_stream | grep consumer

{"event":"connection.created","user_who_performed_action":"System","node":"rabbit@my-host-h1","connected_at":1725854183745,"client_properties":["{<<\"product\">>,longstr,<<\"Pika Python Client Library\">>}","{<<\"platform\">>,longstr,<<\"Python 3.9.18\">>}","{<<\"capabilities\">>,table,[{<<\"authentication_failure_close\">>,bool,true},{<<\"basic.nack\">>,bool,true},{<<\"connection.blocked\">>,bool,true},{<<\"consumer_cancel_notify\">>,bool,true},{<<\"publisher_confirms\">>,bool,true}]}","{<<\"information\">>,longstr,<<\"See http://pika.rtfd.org\">>}","{<<\"version\">>,longstr,<<\"1.1.0\">>}"],"channel_max":2047,"frame_max":131072,"timeout":60,"vhost":"/","user":"System","protocol":"{0,9,1}","ssl_hash":"","ssl_cipher":"","ssl_key_exchange":"","ssl_protocol":"","auth_mechanism":"PLAIN","peer_cert_validity":"","peer_cert_issuer":"","peer_cert_subject":"","ssl":false,"peer_host":"{127,0,0,1}","host":"{127,0,0,1}","peer_port":60300,"port":5672,"name":"127.0.0.1:60300 -> 127.0.0.1:5672","pid":"<rabbit@my-host-h1.1725431545.5060.8>","type":"network","timestamp_in_ms":1725854183749}
{"event":"binding.created","user_who_performed_action":"System","vhost":"/","arguments":[],"routing_key":"consumer.deleted","destination_kind":"queue","destination_name":"test.session.abcd","source_kind":"exchange","source_name":"amq.rabbitmq.event","timestamp_in_ms":1725854183778}
{"event":"consumer.created","user_who_performed_action":"System","arguments":[],"prefetch_count":0,"queue":"test.session.abcd","vhost":"/","channel":"<rabbit@my-host-h1.1725431545.5069.8>","ack_required":false,"exclusive":false,"consumer_tag":"TEST_AMQP","timestamp_in_ms":1725854183779}
{"event":"connection.closed","client_properties":["{<<\"product\">>,longstr,<<\"Pika Python Client Library\">>}","{<<\"platform\">>,longstr,<<\"Python 3.9.18\">>}","{<<\"capabilities\">>,table,[{<<\"authentication_failure_close\">>,bool,true},{<<\"basic.nack\">>,bool,true},{<<\"connection.blocked\">>,bool,true},{<<\"consumer_cancel_notify\">>,bool,true},{<<\"publisher_confirms\">>,bool,true}]}","{<<\"information\">>,longstr,<<\"See http://pika.rtfd.org\">>}","{<<\"version\">>,longstr,<<\"1.1.0\">>}"],"node":"rabbit@my-host-h1","pid":"<rabbit@my-host-h1.1725431545.5060.8>","name":"127.0.0.1:60300 -> 127.0.0.1:5672","timestamp_in_ms":1725854189414}
{"event":"consumer.deleted","user_who_performed_action":"rmq-internal","queue":"test.session.abcd","vhost":"/","channel":"<rabbit@my-host-h1.1725431545.5069.8>","consumer_tag":"TEST_AMQP","timestamp_in_ms":1725854189414}
{"event":"binding.deleted","user_who_performed_action":"rmq-internal","vhost":"/","arguments":[],"routing_key":"consumer.deleted","destination_kind":"queue","destination_name":"test.session.abcd","source_kind":"exchange","source_name":"amq.rabbitmq.event","timestamp_in_ms":1725854189417}

rabbitmq.conf

$ cat /etc/rabbitmq/rabbitmq.conf
listeners.tcp.1 = 127.0.0.1:5672
stomp.listeners.tcp.1 = 127.0.0.1:61613
web_stomp.tcp.listener = 127.0.0.1:15674
stomp.proxy_protocol = true
classic_queue.default_version = 2

enabled_plugins

$ cat /etc/rabbitmq/enabled_plugins
[rabbitmq_auth_backend_oauth2,rabbitmq_event_exchange,rabbitmq_management,rabbitmq_web_stomp].

Reproduction steps

Run this test case: (please replace the credential part for your setup). Ctrl-C after running the consumer and check the output of rabbitmq-diagnostics consume_event_stream.

import pika


creds = pika.PlainCredentials(<get creds>)
params = pika.ConnectionParameters('127.0.0.1', 5672, credentials=creds)
conn = pika.BlockingConnection(params)
chan = conn.channel()


chan.exchange_declare('amq.rabbitmq.event', exchange_type='topic', durable=True, internal=True)

chan.queue_declare('test.session.abcd', durable=True, auto_delete=True)

chan.queue_bind('test.session.abcd', 'amq.rabbitmq.event', 'consumer.deleted')

def callback(ch, method, properties, body):
    print(f'consumer_tag={properties.headers["consumer_tag"]}')

chan.basic_consume(queue='test.session.abcd', on_message_callback=callback, auto_ack=True, consumer_tag='TEST_AMQP')

chan.start_consuming()

### Expected behavior

`consumer.deleted` event to be generated by the event exchange plugin.

### Additional context

_No response_

[mkuratczyk]

This was an intentional change: https://github.com/rabbitmq/rabbitmq-server/releases/tag/v3.12.5

If you actually cancel the consumer, the event is emitted, but if the channel is abruptly cancelled, it is not - you get the channel.closed.

More info:
#9356
#9320

[arshad01]

Hello

Thanks for the reply and the links to the relevant PRs.

I am basically using consumer.deleted event to get the consumer tag, that helps me to do cleanup when consumer disconnects abruptly. Is there any way to get the consumer tag from the connection.closed or channel.closed events?
Thanks

[michaelklishin]

@arshad01 when a channel is closed by an application, the consumer.deleted events are still emitted:

{"event":"channel.created","user_who_performed_action":"guest","vhost":"/","user":"guest","number":1,"connection":"<rabbit@sunnyside.1725931885.1029.0>","name":"127.0.0.1:60161 -> 127.0.0.1:5672 (1)","pid":"<rabbit@sunnyside.1725931885.1039.0>","timestamp_in_ms":1725934542658}
{"event":"node.node.stats","send_bytes":2651,"recv_bytes":633,"sock_port":25672,"sock_addr":"127.0.0.1","peer_port":60160,"peer_addr":"127.0.0.1","route":"{rabbit@sunnyside,'rabbitmqcli-646-rabbit@sunnyside'}","timestamp_in_ms":1725934543557}
{"event":"queue.created","user_who_performed_action":"guest","arguments":["{<<\"x-queue-type\">>,longstr,<<\"quorum\">>}"],"type":"rabbit_quorum_queue","exclusive":false,"auto_delete":false,"durable":true,"name":"qq.1","vhost":"/","timestamp_in_ms":1725934545912}
{"event":"node.node.stats","send_bytes":3285,"recv_bytes":633,"sock_port":25672,"sock_addr":"127.0.0.1","peer_port":60160,"peer_addr":"127.0.0.1","route":"{rabbit@sunnyside,'rabbitmqcli-646-rabbit@sunnyside'}","timestamp_in_ms":1725934548619}
{"event":"consumer.created","user_who_performed_action":"guest","arguments":[],"prefetch_count":2000,"queue":"qq.1","vhost":"/","channel":"<rabbit@sunnyside.1725931885.1039.0>","ack_required":false,"exclusive":false,"consumer_tag":"bunny-1725934549000-272473404194","timestamp_in_ms":1725934549857}
{"event":"node.node.stats","send_bytes":3884,"recv_bytes":633,"sock_port":25672,"sock_addr":"127.0.0.1","peer_port":60160,"peer_addr":"127.0.0.1","route":"{rabbit@sunnyside,'rabbitmqcli-646-rabbit@sunnyside'}","timestamp_in_ms":1725934553699}
{"event":"channel.closed","consumer_count":1,"user_who_performed_action":"guest","pid":"<rabbit@sunnyside.1725931885.1039.0>","timestamp_in_ms":1725934554623}
{"event":"consumer.deleted","user_who_performed_action":"rmq-internal","queue":"qq.1","vhost":"/","channel":"<rabbit@sunnyside.1725931885.1039.0>","consumer_tag":"bunny-1725934549000-272473404194","timestamp_in_ms":1725934554626}

They are not emitted when a channel is closed abruptley.

A very quick tests demonstrates that when a connection is closed cleanly, the event is emitted as well:

{"event":"channel.closed","consumer_count":1,"user_who_performed_action":"guest","pid":"<rabbit@sunnyside.1725931885.1081.0>","timestamp_in_ms":1725934640905}
{"event":"connection.closed","client_properties":["{<<\"capabilities\">>,table,[{<<\"publisher_confirms\">>,bool,true},{<<\"consumer_cancel_notify\">>,bool,true},{<<\"exchange_exchange_bindings\">>,bool,true},{<<\"basic.nack\">>,bool,true},{<<\"connection.blocked\">>,bool,true},{<<\"authentication_failure_close\">>,bool,true}]}","{<<\"product\">>,longstr,<<\"Bunny\">>}","{<<\"platform\">>,longstr,<<\"ruby 3.3.1 (2024-04-23 revision c56cd86388) [arm64-darwin23]\">>}","{<<\"version\">>,longstr,<<\"2.24.0.pre\">>}","{<<\"information\">>,longstr,<<\"http://rubybunny.info\">>}","{<<\"connection_name\">>,void,undefined}"],"node":"rabbit@sunnyside","pid":"<rabbit@sunnyside.1725931885.1029.0>","name":"127.0.0.1:60161 -> 127.0.0.1:5672","timestamp_in_ms":1725934640906}
{"event":"consumer.deleted","user_who_performed_action":"rmq-internal","queue":"qq.1","vhost":"/","channel":"<rabbit@sunnyside.1725931885.1081.0>","consumer_tag":"bunny-1725934634000-821373516586","timestamp_in_ms":1725934640908}

#9320 explains the reasoning behind this change.

Since connections and channels are less often closed cleanly all at once — which is the problematic scenario — we still emit consumer.deleted for those cases.

[arshad01]

Thanks @michaelklishin for reply.

In my use case, the closure of consumers can be both clean and abrupt. One use case is a consumer in a browser subscribed through a web stomp connection. The browser or browser tab can be closed suddenly and it may not be possible to do a clean close of web stomp connection.

[michaelklishin]

You can list online consumers using the HTTP API and compute the delta against the registered ones to… well, do what you previously did with the disconnected consumers.

[arshad01]

Thanks for the suggestion @michaelklishin. I'll check it out.