Persistance of MQTT "Will & Testament" / "Death" message across protocols

[Robbert-Brand]

Describe the bug

Dear RabbitMQ,

We are connecting an MQTT client to the RabbitMQ data broker via the latest MQTT pluging. We set the MQTT "will" / "death" message for MQTT the client with a QoS of 1. Upon disconnection by the client, the will is send across all MQTT clients with QoS 1 who are subscribed to the related topic. (As expected)

However, we also subscribe a RabbitMQ client to the MQTT message exchange via a rabbitMQ queue connection using Pika. We consistently recieve all MQTT messages via the durable rabbitMQ queue, with the exception of MQTT Death messages which we miss 50% of the time.

A code sample:

      self._channel.queue_declare(
            queue=self._rabbitmq_param_queue_name,
            durable=True,
            auto_delete=False,
            arguments={
                "x-queue-type": "quorum",
                "x-max-length-bytes": 500000000,  # max 500MB
                # "x-max-length": 1000000         # max 1M messages
                "x-consumer-timeout": 600000,     # 10 minutes timeout after non-ack
            }
        )

        # Bind the stream to the main exchange
        self._channel.queue_bind(
            queue=self._rabbitmq_param_queue_name,
            exchange=self._rabbitmq_param_exchange_name,
            routing_key=self._rabbitmq_param_routing_key
        )

Reproduction steps

1. Ensured that all messages were send with the right MQTT QoS and RabbitMQ durable flag.
2. Tested with other sorts of messages (Non-death messages)

Expected behavior

A death message is sent via MQTT with QoS 1, to be guaranteed to arrive via the RabbitMQ Quorum queue connection.

Additional context

No response

[lukebakken]

Hello and thanks for using RabbitMQ.

• What version of RabbitMQ are you using?
• Erlang version?
• Python version?
• Would it be possible to provide two complete, runnable Python programs to demonstrate the issue? One that uses MQTT and the other that uses Pika? This will save us time.

Thank you

[michaelklishin]

@Robbert-Brand without an executable way to reproduce we have to guess as to what your routing keys are. And we do not guess in this community.

MQTT topic structure uses different separators/wildcards from AMQP 0-9-1, and the translation has its limits, as has been discussed multiple times over the years.

So it can be that some of the messages happen to route correctly but others use a more complex topic structure that does not route with the AMQP 0-9-1 bindings used.

[Robbert-Brand]

Gentlemen,

Thank you for your quick answers, and rightfully questions. I might have rushed my previous message, and I apologise for the lack of information. To further complete my previous message:

What version of RabbitMQ are you using?

• RabbitMQ 3.12.4
  Erlang version?
• Erlang 25.3.2.6
  Python version?
• Python 3.10.12
• Pika 1.3.2

Set-up:

1. Pika script, just listening
2. Paho script publishing, and triggering "Death message" op on termination with keyboard shortcut ctrl+C
3. MqttX client, to verify that " Death message" is actually sent.

Pika listener:

import pika
import time

def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)

_stream_connection = pika.BlockingConnection(
    pika.ConnectionParameters(
        host='localhost',
        port=5672,
        connection_attempts=2000,
        retry_delay=30,
        heartbeat=600,
        blocked_connection_timeout=300
    )
)
_stream_channel = _stream_connection.channel()
_stream_channel.basic_qos(
    prefetch_count=1000
)


_stream_channel.queue_declare(
    queue='_rabbitmq_param_queue_name',
    durable=True,
    auto_delete=False,
    arguments={
        # "x-queue-type": "stream",
        "x-queue-type": "quorum",
        "x-max-length-bytes": 500000000,  # max 500MB
        # "x-max-length": 1000000         # max 1M messages
        "x-consumer-timeout": 600000,     # 10 minutes timeout after non-ack
    }
    )

_stream_channel.queue_bind(
    queue='_rabbitmq_param_queue_name',
    exchange='amq.topic',
    routing_key="#"
)

_stream_channel.basic_consume(
    queue='_rabbitmq_param_queue_name',
    on_message_callback=callback,
    auto_ack=False,
    consumer_tag='_rabbitmq_consumer_tag'
)

while True:
    # I am aware that this is not the best way to do this
    _stream_connection.process_data_events()

    # a small sleep reduced cpu usage
    time.sleep(0.01)

Paho Mqtt script:

import paho.mqtt.client as mqtt
import time

# Define MQTT broker address and port
broker_address = "localhost"
broker_port = 1883

# Define MQTT client and connect to broker
client = mqtt.Client(client_id="my_client_id", protocol=mqtt.MQTTv311)
client.will_set(topic="my/will/topic", payload="my will payload", qos=1, retain=False)
client.connect(broker_address, broker_port)

# Publish message
client.publish(topic="my/topic", payload="Hello, MQTT!", qos=1, retain=False)

# Disconnect from broker
# client.disconnect()
while True:
    time.sleep(10)

Result:

1. Pika script logs all non-death messages and about 10% of death messages.
2. MQTTX receives all death messages, including death messages.

[ansd]

I edited your example to make it work on my local machine.

Whenever I disconnect the Paho Mqtt client by typing Ctrl+C into the terminal, RabbitMQ logs

2023-10-25 09:27:02.358368+02:00 [info] <0.1853.0> MQTT connection <<"[::1]:64588 -> [::1]:1883">> will terminate because peer closed TCP connection
2023-10-25 09:27:02.359038+02:00 [debug] <0.1853.0> sent Will Message to topic my/will/topic for MQTT client ID my_client_id

and the Pika client receives the Will message:

[x] Received b'my will payload'

Hence your example does not demonstrate what you claim. Everything works as expected.

[michaelklishin]

@Robbert-Brand since we cannot reproduce the behavior described, the most productive path forward would be for you to put together an executable example. That is, a public repository on GitHub (or elsewhere) with a set of steps to execute.

You are welcome to use tools such as Docker Compose and run RabbitMQ in a container if that helps. That generally helps with reproducibility.

I would encourage you to investigate the behavior of your Pika-based consumers by taking a traffic capture, collecting more logs and metrics in your own code. It is quite common to see cases where RabbitMQ gets the blame for application behavior.

[Robbert-Brand]

Hello @michaelklishin , @ansd ,

Thank you for the responses. Interesting to see that the same code example has such a different result on your set-up, as it is very reproducible on mine with this demo.

Perhaps some additional information about the RabbitMQ set-up that we are running could make a difference:

docker-compose.yaml:

version: '3'
services:
  rabbitmq-node1:
    image: rabbitmq:latest
    container_name: rabbitmq-node1
    hostname: rabbitmq-node1
    restart: always
    environment:
      RABBITMQ_ERLANG_COOKIE: "test-cookie"
      RABBITMQ_ENABLED_PLUGINS_FILE: /etc/rabbitmq/enabled_plugins
    volumes:
      - ./Config/RabbitMQNodes/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro
      - ./Config/RabbitMQNodes/enabled_plugins:/etc/rabbitmq/enabled_plugins:ro
      - ./Config/RabbitMQNodes/management.conf:/etc/rabbitmq/conf.d/management.conf:ro
      - ./Config/RabbitMQNodes/broker_definition.json:/etc/rabbitmq/broker_definition.json:ro
    networks:
      broker-network:
        ipv4_address: 172.0.0.2
        aliases:
          - rabbitmq-node1

  rabbitmq-node2:
    image: rabbitmq:latest
    container_name: rabbitmq-node2
    hostname: rabbitmq-node2
    restart: always
    environment:
      RABBITMQ_ERLANG_COOKIE: "test-cookie"
      RABBITMQ_ENABLED_PLUGINS_FILE: /etc/rabbitmq/enabled_plugins
    volumes:
      - ./Config/RabbitMQNodes/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro
      - ./Config/RabbitMQNodes/enabled_plugins:/etc/rabbitmq/enabled_plugins:ro
      - ./Config/RabbitMQNodes/management.conf:/etc/rabbitmq/conf.d/management.conf:ro
      - ./Config/RabbitMQNodes/broker_definition.json:/etc/rabbitmq/broker_definition.json:ro
    networks:
      broker-network:
        ipv4_address: 172.0.0.3
        aliases:
          - rabbitmq-node2

  rabbitmq-node3:
    image: rabbitmq:latest
    container_name: rabbitmq-node3
    hostname: rabbitmq-node3
    restart: always
    environment:
      RABBITMQ_ERLANG_COOKIE: "test-cookie"
      RABBITMQ_ENABLED_PLUGINS_FILE: /etc/rabbitmq/enabled_plugins
    volumes:
      - ./Config/RabbitMQNodes/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro
      - ./Config/RabbitMQNodes/enabled_plugins:/etc/rabbitmq/enabled_plugins:ro
      - ./Config/RabbitMQNodes/management.conf:/etc/rabbitmq/conf.d/management.conf:ro
      - ./Config/RabbitMQNodes/broker_definition.json:/etc/rabbitmq/broker_definition.json:ro
    networks:
      broker-network:
        ipv4_address: 172.0.0.4
        aliases:
          - rabbitmq-node3

  haproxy-1:
    image: haproxy:latest
    container_name: haproxy-1
    ports:
      - "5672:5672"
      - "15672:15672"
      - "1883:1883"
      - "5552:5552"
    volumes:
      - ./Config/HaProxy1/haproxy.cfg:/usr/local/etc/haproxy/haproxy.cfg:ro
    networks:
      broker-network:
        ipv4_address: 172.0.0.10
        aliases:
          - haproxy-1
    restart: always

networks:
  broker-network:
    driver: bridge
    ipam:
      driver: default
      config:
        - subnet: 172.0.0.0/24
    driver_opts:
      com.docker.network.enable_ipv6: "false"
      com.docker.network.bridge.name: "brok_nw"      # Set Linux network interface name

rabbitmq.conf

listeners.tcp.default = 5672
cluster_formation.peer_discovery_backend = rabbit_peer_discovery_classic_config
cluster_formation.classic_config.nodes.1 = rabbit@rabbitmq-node1
cluster_formation.classic_config.nodes.2 = rabbit@rabbitmq-node2
cluster_formation.classic_config.nodes.3 = rabbit@rabbitmq-node3
cluster_name = rabbit@rabbitmq-cluster

load_definitions = /etc/rabbitmq/broker_definition.json

management.tcp.port = 15672
mqtt.listeners.tcp.default = 1883
stream.listeners.tcp.1 = 5552

# Stream plugin defaults
stream.tcp_listen_options.backlog = 4096
stream.tcp_listen_options.recbuf  = 131072
stream.tcp_listen_options.sndbuf  = 131072

stream.tcp_listen_options.keepalive = true
stream.tcp_listen_options.nodelay   = true

stream.tcp_listen_options.exit_on_close = true
stream.tcp_listen_options.send_timeout  = 120

mqtt.allow_anonymous  = true
mqtt.default_user     = guest
mqtt.default_pass     = guest

mqtt.vhost            = /
mqtt.exchange         = main_topic_exchange
# 24 hours by default
mqtt.subscription_ttl = 86400000
mqtt.prefetch         = 10

mqtt.durable_queue_type = quorum

mqtt.retained_message_store = rabbit_mqtt_retained_msg_store_dets
mqtt.retained_message_store_dets_sync_interval = 2000

management.conf

management_agent.disable_metrics_collector = false

enabled_plugins.conf

[rabbitmq_management, rabbitmq_mqtt, rabbitmq_prometheus, rabbitmq_stream].

broker_definition.json
please note: I removed the user section in the json below.

{
	"vhosts": [{
		"name": "/"
	}],
	"permissions": [{
		"user": "guest",
		"vhost": "/",
		"configure": ".*",
		"write": ".*",
		"read": ".*"
	}],
	"topic_permissions": [],
	"parameters": [],
	"policies": [{
		"ha-sync-mode":"automatic",
		"ha-mode":"nodes"
	}],
	"queues": [],
	"exchanges": [{
		"name": "main_topic_exchange",
		"vhost": "/",
		"type": "topic",
		"durable": true,
		"auto_delete": false,
		"internal": false,
		"arguments": {}
	}],
	"bindings": []
}

haproxy.conf

frontend rabbitmq-frontend-management
    bind *:15672
	mode tcp
    default_backend rabbitmq-backend-management

backend rabbitmq-backend-management
	mode tcp
    balance roundrobin
    option tcp-check
    server rabbitmq-node1 172.0.0.2:15672 check
	server rabbitmq-node2 172.0.0.3:15672 check
	server rabbitmq-node3 172.0.0.4:15672 check



frontend rabbitmq-frontend-amqp
	bind *:5672
	mode tcp
	default_backend rabbitmq-backend-amqp

backend rabbitmq-backend-amqp
	mode tcp
	balance roundrobin
	option redispatch
	server rabbitmq-node1 172.0.0.2:5672 check
	server rabbitmq-node2 172.0.0.3:5672 check
	server rabbitmq-node3 172.0.0.4:5672 check



frontend rabbitmq-frontend-mqtt
    bind *:1883
	mode tcp
    default_backend rabbitmq-backend-mqtt

backend rabbitmq-backend-mqtt
	mode tcp
    balance roundrobin
    option redispatch
    server rabbitmq-node1 172.0.0.2:1883 check
	server rabbitmq-node2 172.0.0.3:1883 check
	server rabbitmq-node3 172.0.0.4:1883 check



frontend rabbitmq-frontend-stream
    bind *:5552
	mode tcp
    default_backend rabbitmq-backend-stream

backend rabbitmq-backend-stream
	mode tcp
    balance roundrobin
    option tcp-check
    server rabbitmq-node1 172.0.0.2:5552 check
	server rabbitmq-node2 172.0.0.3:5552 check
	server rabbitmq-node3 172.0.0.4:5552 check

global
    daemon
    maxconn 256

defaults
    mode http
    timeout connect 5000ms
    timeout client 50000ms
    timeout server 50000ms