How to Test Retained Messages (and Clear Them)

To test retained messages, publish with the retain flag, then connect a brand-new subscriber and check that it receives the message immediately with retain=1. Then test the other half of the contract: a second retained publish overwrites the first, a non-retained publish does not, and a retained publish with an empty payload clears the topic so the next subscriber receives nothing.

This guide walks through each case with mosquitto_pub / mosquitto_sub, turns them into an automated pytest suite, and covers the MQTT 5 options that change retained behavior.

What to verify

The basics are explained in MQTT retained messages. Use a unique topic tree such as testmqtt/retain-test/... so you don't read someone else's state on a shared broker — and clean it up afterwards.

Test 1: a late subscriber gets the last value

mosquitto_pub -h broker.connectmq.com -p 1883 \
  -t 'testmqtt/retain-test/door' -m 'closed' -r -q 1

# a brand-new subscriber, started after the publish
mosquitto_sub -h broker.connectmq.com -p 1883 \
  -t 'testmqtt/retain-test/door' -C 1 -W 5 -F '%r %t %p'

Expected output: 1 testmqtt/retain-test/door closed. -F '%r %t %p' prints the retain flag, topic and payload; -C 1 exits after one message and -W 5 gives up after five seconds, so the command never hangs in a script. If the subscriber times out, the message was not stored: check that you really passed -r, that your ACL allows publishing retained messages, and whether the broker supports retain at all (Retain not supported is reason code 0x9A in MQTT 5).

Test 2: live delivery has retain=0

Start a subscriber first and keep it running, then publish a retained message:

# Terminal 1: existing subscriber
mosquitto_sub -h broker.connectmq.com -p 1883 -t 'testmqtt/retain-test/#' -F '%r %t %p'

# Terminal 2
mosquitto_pub -h broker.connectmq.com -p 1883 -t 'testmqtt/retain-test/door' -m 'open' -r

Terminal 1 first prints the stored value with 1, then the new one with 0 testmqtt/retain-test/door open. That is correct: in MQTT 3.1.1 the flag tells a subscriber "this is stored state you received because you subscribed", not "the publisher set retain". Code that relies on msg.retain to tell old state from live events works — code that expects to see the publisher's flag does not. With MQTT 5, subscribe with -V mqttv5 --retain-as-published and Terminal 1 shows 1 for the live message too.

Tests 3–5: overwrite, ignore, clear

B="-h broker.connectmq.com -p 1883"
T='testmqtt/retain-test/door'

mosquitto_pub $B -t "$T" -m 'open' -r              # 3. overwrite "closed"
mosquitto_pub $B -t "$T" -m 'glitch'               # 4. not retained: must not replace
mosquitto_sub $B -t "$T" -C 1 -W 5 -F '%r %p'      #    → 1 open

mosquitto_pub $B -t "$T" -r -n                     # 5. zero-length retained = clear
mosquitto_sub $B -t "$T" -C 1 -W 3 -F '%r %p'      #    → nothing, times out

-n sends a null (zero-length) payload. Three details trip people up:

• The topic must match exactly. Publishing an empty message to testmqtt/retain-test/# is invalid — wildcards are not allowed in publish topics. Clear each topic, or use the bulk method below.
• The empty message is delivered too. Subscribers that are connected at that moment receive a zero-length payload. Make sure your handlers don't crash on b"" (a JSON parser will).
• A space is not empty. -m ' ' stores a one-byte retained message. If a topic "won't clear", this is a common cause — see retained message won't clear.

Test 6: wildcards and bulk cleanup

# list: prints only retained messages, stops at the first live one or after 3 s
mosquitto_sub -h broker.connectmq.com -p 1883 \
  -t 'testmqtt/retain-test/#' --retained-only -W 3 -F '%t = %p'

# remove: publishes an empty retained message for every retained topic it receives
mosquitto_sub -h broker.connectmq.com -p 1883 \
  -t 'testmqtt/retain-test/#' --retained-only --remove-retained -W 3

A wildcard subscription receives one retained message per matching topic, so subscribing to a broad filter after a deployment can deliver thousands of messages at once. Check which topics a filter covers with the topic matcher before running --remove-retained — on a production broker it deletes everything that matches.

Automating it with pytest

The same contract as an automated suite (paho-mqtt 2.x). Each test gets its own random topic and the fixture clears it afterwards:

import threading, time, uuid
import paho.mqtt.client as mqtt
import pytest

HOST, PORT = "broker.connectmq.com", 1883

def connect():
    ready = threading.Event()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
    # c.username_pw_set("username", "password")
    c.on_connect = lambda cl, ud, flags, rc, props: ready.set()
    c.connect(HOST, PORT)
    c.loop_start()
    assert ready.wait(5), "no CONNACK"
    return c

def publish(topic, payload, retain=True):
    c = connect()
    c.publish(topic, payload, qos=1, retain=retain).wait_for_publish(5)
    c.disconnect()
    c.loop_stop()

def fetch(topic, wait=1.5):
    """Subscribe with a brand-new client; return [(payload, retain_flag), ...]."""
    got = []
    c = connect()
    c.on_message = lambda cl, ud, msg: got.append((msg.payload, msg.retain))
    c.subscribe(topic, qos=1)
    time.sleep(wait)
    c.disconnect()
    c.loop_stop()
    return got

@pytest.fixture
def topic():
    t = f"testmqtt/retain-test/{uuid.uuid4().hex[:8]}"
    yield t
    publish(t, b"")                     # always clean up

def test_late_subscriber_gets_retained(topic):
    publish(topic, b"closed")
    assert fetch(topic) == [(b"closed", True)]

def test_overwrite_keeps_only_last(topic):
    publish(topic, b"one")
    publish(topic, b"two")
    assert fetch(topic) == [(b"two", True)]

def test_non_retained_publish_does_not_replace(topic):
    publish(topic, b"closed")
    publish(topic, b"glitch", retain=False)
    assert fetch(topic) == [(b"closed", True)]

def test_empty_payload_clears(topic):
    publish(topic, b"closed")
    publish(topic, b"")
    assert fetch(topic) == []

The fixed wait keeps the example short; in CI, replace it with an event that fires on the first message plus a short grace period. For reusable broker fixtures see pytest fixtures for MQTT.

MQTT 5 options that change the result

MQTT 5 adds subscription options that a complete test should cover (details in subscription options):

• Retain Handling — 0 sends retained messages on every subscribe (default), 1 only when the subscription is new, 2 never.
• Retain As Published — keeps the publisher's retain flag on live deliveries (Test 2 then shows 1).
• Message Expiry Interval — a retained message with an expiry is removed when it expires; test it with message expiry tests.

import paho.mqtt.client as mqtt
from paho.mqtt.subscribeoptions import SubscribeOptions

c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
c.on_message = lambda cl, ud, msg: print("got", msg.topic, msg.payload, msg.retain)
c.connect("broker.connectmq.com", 1883)
c.subscribe("testmqtt/retain-test/door",
            options=SubscribeOptions(qos=1, retainHandling=SubscribeOptions.RETAIN_DO_NOT_SEND))
c.loop_forever()   # prints only messages published from now on

Test 7: does retained state survive a restart?

Retained messages live in broker memory unless the broker persists them to disk. On a self-hosted broker, publish a retained value, restart the broker, and run Test 1 again. If the value is gone, enable persistence (for Mosquitto: persistence true and a persistence_location) or accept that devices must republish state after a broker restart. Managed brokers handle this for you, but the test is still worth running once.

Inspect retained messages in your browser

Retained messages are a small contract — stored on publish with retain, delivered once per new subscription, replaced by the next retained publish, cleared by an empty one. Test each clause once by hand with the online MQTT client or mosquitto_sub -F '%r', then keep the pytest suite so a broker change never silently breaks your devices' "last known state".

Frequently asked questions