Testing MQTT Shared Subscriptions and Load Balancing

A shared subscription ($share/<group>/<topic>) turns several subscribers into a worker pool: each message on the topic is delivered to one member of the group instead of all of them. To test it, start several workers in the same group, publish a numbered sequence, and check three things: every message arrived exactly once across the group, the load was spread, and nothing was lost when a worker crashed mid-stream.

This guide shows a quick CLI check, a Python test with a crash simulation and the edge cases worth asserting. The concepts are covered in MQTT shared subscriptions. Here we only test them.

What a correct broker does

The distribution strategy is up to the broker. Mosquitto uses round robin. Other brokers offer random, sticky, or hash-based strategies, often configurable. Your test should check fairness within tolerance, not an exact pattern, unless you rely on a specific strategy.

Step 0: check that the broker supports it

An MQTT 5 broker announces support in the CONNACK with the Shared Subscription Available property. If the property is missing, support is assumed. If it is 0, the broker will reject or ignore $share subscriptions, and SUBACK returns 0x9E Shared Subscriptions not supported. Either way, the distribution test below is the real proof. The reason code lookup decodes whatever SUBACK you get.

The group name is a single topic level: no /, + or #. The rest is a normal topic filter, so wildcards work after the group: $share/workers/testmqtt/share/+/jobs. Check filters with the topic matcher before blaming the broker.

Quick check with the Mosquitto CLI

Open two terminals with members of the same group, and a third with a normal subscriber:

mosquitto_sub -h broker.connectmq.com -V mqttv5 -q 1 \
  -t '$share/demo/testmqtt/share/9b1e/jobs' -v

mosquitto_sub -h broker.connectmq.com -V mqttv5 -q 1 \
  -t 'testmqtt/share/9b1e/jobs' -v

for i in $(seq 1 10); do
  mosquitto_pub -h broker.connectmq.com -V mqttv5 -q 1 \
    -t testmqtt/share/9b1e/jobs -m "job $i"
done

Terminals 1 and 2 split the ten jobs between them (with round robin, odd numbers on one side, even on the other). Terminal 3 sees all ten. Always put the topic in single quotes: unquoted or in double quotes, your shell would expand $share as a variable.

Automated distribution and failover test

This paho-mqtt 2.x script starts three workers in one group, publishes 100 numbered QoS 1 messages, and reports per-worker counts, gaps and duplicates. With --failover it crashes worker-0 halfway: it stops reading, so its last messages stay unacknowledged, then drops the TCP connection without a DISCONNECT, as a crash would.

import sys, time, uuid, collections
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
RUN = uuid.uuid4().hex[:8]
TOPIC = f"testmqtt/share/{RUN}/jobs"
GROUP = f"workers{RUN}"
WORKERS, MESSAGES, RATE = 3, 100, 10     # RATE in messages per second
FAILOVER = "--failover" in sys.argv          # crash worker-0 halfway through

got = collections.defaultdict(list)          # worker name -> message ids received

def make_worker(name):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{name}-{RUN}", protocol=mqtt.MQTTv5)

    def on_connect(cl, userdata, flags, rc, props):
        if getattr(props, "SharedSubscriptionAvailable", 1) == 0:
            print(f"{name}: CONNACK says shared subscriptions are not available")
        cl.subscribe(f"$share/{GROUP}/{TOPIC}", qos=1)

    def on_subscribe(cl, userdata, mid, reason_codes, props):
        print(f"{name}: SUBACK {reason_codes[0]}")

    c.on_connect, c.on_subscribe = on_connect, on_subscribe
    c.on_message = lambda cl, userdata, msg: got[name].append(int(msg.payload))
    c.connect(HOST, PORT)
    c.loop_start()
    return c

workers = {f"worker-{i}": make_worker(f"worker-{i}") for i in range(WORKERS)}
time.sleep(2)                                 # let all subscriptions settle

pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"share-pub-{RUN}", protocol=mqtt.MQTTv5)
pub.connect(HOST, PORT)
pub.loop_start()
for i in range(MESSAGES):
    if FAILOVER and i == MESSAGES // 2:
        crashed = workers.pop("worker-0")
        crashed.loop_stop()                   # stop reading: unacked messages stay in flight
        crashed.socket().close()              # drop TCP without DISCONNECT, like a crash
        print(f"crashed worker-0 after message {i - 1}")
    pub.publish(TOPIC, str(i), qos=1)
    time.sleep(1 / RATE)
time.sleep(5)

all_ids = [m for ids in got.values() for m in ids]
missing = sorted(set(range(MESSAGES)) - set(all_ids))
dupes = len(all_ids) - len(set(all_ids))
for name in sorted(got):
    print(f"{name}: {len(got[name]):4d} messages, in order: {got[name] == sorted(got[name])}")
print(f"delivered {len(set(all_ids))}/{MESSAGES}, missing {len(missing)}, duplicates {dupes}")
if missing:
    print("first missing ids:", missing[:10])
for c in workers.values():
    c.disconnect(); c.loop_stop()
pub.disconnect(); pub.loop_stop()

A run against a round-robin broker looks like this:

worker-0: SUBACK Granted QoS 1
worker-1: SUBACK Granted QoS 1
worker-2: SUBACK Granted QoS 1
crashed worker-0 after message 49
worker-0:   17 messages, in order: True
worker-1:   42 messages, in order: True
worker-2:   41 messages, in order: True
delivered 100/100, missing 0, duplicates 0

How to read it:

• Without --failover, the counts should be roughly equal (33/33/34 with round robin). A heavily skewed split points to a sticky or hash strategy, or to a slow worker that the broker skips.
• With --failover, missing 0 means the broker re-routed the crashed member's work. Missing IDs right around the crash point mean in-flight messages were dropped with the session. Some brokers wait for the client to reconnect first, so test with your real session settings.
• Duplicates are legal at QoS 1: a message can be delivered to the crashed worker and re-sent to another. Workers in a shared group must be idempotent. See duplicate MQTT messages.
• Missing messages without any crash usually mean a rate limit or queue limit on the broker side. Lower RATE and run again. Shared public endpoints are especially likely to throttle.

Edge cases worth asserting

1. Ordering. Each member sees its own messages in order, but there is no global order across the group. Message 41 can be processed before message 40 on another worker. If order matters per device, use a hash-by-topic strategy where available or partition topics. The message ordering test shows how to measure it.
2. Retained messages. Publish a retained message, then start a shared subscriber: it must receive nothing. Workers that need initial state must get it another way.
3. Mixed QoS. The delivered QoS is the minimum of publish QoS and each member's granted QoS. A member that subscribed at QoS 0 loses messages on crash, even if the others use QoS 1.
4. Subscription options. No Local on a shared subscription is a protocol error. The other options are explained in MQTT subscription options.
5. Scaling the group. Add a fourth worker mid-run: it should start receiving its share right away. Remove one: the others pick up the slack.
6. Persistent members. With Session Expiry above 0, a disconnected member keeps its seat in the group, and some brokers keep sending it messages that queue up while it is offline. Decide whether workers should be persistent at all.

Testing the throughput gain

Shared subscriptions exist to scale consumers. Measure that the scaling is real: run the script with 1, 2 and 4 workers that each sleep a fixed processing time per message, and compare the total time to drain a burst. Throughput should grow almost linearly until the broker or the network becomes the limit. A fan-in test is the natural companion: many publishers feeding one group.

Watch the distribution live

Open the online MQTT client in two browser tabs, subscribe both to the same $share/... filter and publish from a third tab. You see the split in real time, which makes a good sanity check before running the script.

Publish test jobs from the browser

A shared subscription test takes a few minutes and answers the questions that matter in production: does every job get processed, does the load really spread, and what happens to work in progress when a worker dies. Keep the failover run in your CI suite, and repeat it whenever you change brokers or session settings.

Frequently asked questions