Testing MQTT 5 Request/Response with Response Topic and Correlation Data

MQTT is publish/subscribe, but plenty of real systems need a question and an answer: "read this setting", "run this command and tell me if it worked". MQTT 5 request/response standardises that with two PUBLISH properties: the requester sets a Response Topic (where to send the answer) and Correlation Data (an opaque ID that ties the answer to the question). The responder copies the correlation data into its reply and publishes to the response topic.

This guide implements the pattern in Python and JavaScript, shows a one-line command-line test, and lists the test cases that catch the bugs people actually ship: mismatched IDs, missing timeouts, concurrent requests and ACLs that block the reply. For the design side of the pattern, see MQTT request/response.

How the flow works

The broker does not enforce anything here — it simply forwards the properties. That is good news for portability (any MQTT 5 broker works) and bad news for correctness: if your responder forgets to copy the correlation data, nothing complains until your requester times out.

There are two related CONNECT/CONNACK properties. A client can set Request Response Information in CONNECT, and the broker may answer with Response Information in CONNACK — a topic prefix the client should use for replies. Support varies between brokers, so most applications simply build the response topic from their client ID. Either way, the response topic is a topic name: it must not contain + or #.

Test it from the command line first

Mosquitto ships mosquitto_rr, a request/response client that speaks MQTT 5 by default. Start a fake responder in one terminal — a subscriber that prints the properties — so you can see what arrives:

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 \
  -t 'testmqtt/rr/req/#' \
  -F 'topic=%t response=%R corr=%D payload=%p'

In a second terminal send a request with mosquitto_rr. It subscribes to the response topic, publishes the request and waits for one reply:

mosquitto_rr -h broker.connectmq.com -p 1883 \
  -t testmqtt/rr/req/device-7 \
  -e testmqtt/rr/resp/cli-$$ \
  -m '{"op":"get","key":"firmware"}' -W 5

Terminal 1 shows the response topic and correlation data. To answer by hand, publish to the printed response topic with the same correlation data from a third terminal:

mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 \
  -t testmqtt/rr/resp/cli-12345 \
  -D publish correlation-data '<value printed in terminal 1>' \
  -m '{"firmware":"2.4.1"}'

mosquitto_rr prints the reply and exits; with -W 5 it gives up after five seconds, which is your first timeout test. You can also watch both topics in the online MQTT client — subscribe to testmqtt/rr/# and every request and response appears side by side.

A Python responder and requester

The following uses paho-mqtt 2.x with MQTT 5. The responder subscribes to requests and echoes the correlation data back:

import json
import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

HOST = "broker.connectmq.com"

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe("testmqtt/rr/req/+", qos=1)

def on_message(client, userdata, msg):
    resp_topic = getattr(msg.properties, "ResponseTopic", None)
    corr = getattr(msg.properties, "CorrelationData", None)
    if not resp_topic:
        print("request without Response Topic, ignored:", msg.topic)
        return
    request = json.loads(msg.payload)
    reply = {"ok": True, "echo": request}
    props = Properties(PacketTypes.PUBLISH)
    if corr is not None:
        props.CorrelationData = corr
    client.publish(resp_topic, json.dumps(reply), qos=1, properties=props)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_message = on_message
client.connect(HOST, 1883)
client.loop_forever()

The requester keeps a table of pending requests keyed by correlation ID, so many requests can be in flight at once and replies may arrive in any order:

import json, threading, time, uuid
import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

HOST = "broker.connectmq.com"
CLIENT_ID = f"rr-test-{uuid.uuid4().hex[:8]}"
RESP_TOPIC = f"testmqtt/rr/resp/{CLIENT_ID}"
pending = {}  # correlation id -> [Event, reply]
ready = threading.Event()

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(RESP_TOPIC, qos=1)

def on_subscribe(client, userdata, mid, reason_codes, properties):
    ready.set()

def on_message(client, userdata, msg):
    corr = getattr(msg.properties, "CorrelationData", b"")
    slot = pending.get(corr)
    if slot is None:
        print("unexpected or late reply:", corr)
        return
    slot[1] = json.loads(msg.payload)
    slot[0].set()

def request(client, device, body, timeout=5.0):
    corr = uuid.uuid4().bytes
    slot = pending[corr] = [threading.Event(), None]
    props = Properties(PacketTypes.PUBLISH)
    props.ResponseTopic = RESP_TOPIC
    props.CorrelationData = corr
    props.MessageExpiryInterval = int(timeout)  # stale requests are dropped
    t0 = time.perf_counter()
    client.publish(f"testmqtt/rr/req/{device}", json.dumps(body), qos=1, properties=props)
    ok = slot[0].wait(timeout)
    pending.pop(corr, None)
    if not ok:
        raise TimeoutError(f"no reply from {device} within {timeout}s")
    return slot[1], (time.perf_counter() - t0) * 1000

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID, protocol=mqtt.MQTTv5)
client.on_connect, client.on_subscribe, client.on_message = on_connect, on_subscribe, on_message
client.connect(HOST, 1883)
client.loop_start()
ready.wait(5)

reply, ms = request(client, "device-7", {"op": "get", "key": "firmware"})
print(f"reply in {ms:.1f} ms: {reply}")

Two details matter. The requester waits for SUBACK before sending — otherwise a fast responder can reply before the subscription exists and the answer is lost. And the request carries a Message Expiry Interval equal to the timeout, so a responder that comes online later does not execute a command the caller already gave up on (see message expiry).

The same pattern in MQTT.js

MQTT.js 5 exposes the properties on the publish options and on the received packet:

import mqtt from "mqtt";
import { randomUUID } from "node:crypto";

const id = `rr-js-${randomUUID().slice(0, 8)}`;
const respTopic = `testmqtt/rr/resp/${id}`;
const client = await mqtt.connectAsync("mqtt://broker.connectmq.com:1883", {
  protocolVersion: 5,
  clientId: id,
});
await client.subscribeAsync(respTopic, { qos: 1 });

const pending = new Map();
client.on("message", (topic, payload, packet) => {
  const corr = packet.properties?.correlationData?.toString();
  pending.get(corr)?.(JSON.parse(payload.toString()));
  pending.delete(corr);
});

function request(device, body, timeoutMs = 5000) {
  const corr = randomUUID();
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => {
      pending.delete(corr);
      reject(new Error(`timeout waiting for ${device}`));
    }, timeoutMs);
    pending.set(corr, (reply) => { clearTimeout(timer); resolve(reply); });
    client.publish(`testmqtt/rr/req/${device}`, JSON.stringify(body), {
      qos: 1,
      properties: { responseTopic: respTopic, correlationData: Buffer.from(corr) },
    });
  });
}

console.log(await request("device-7", { op: "get", key: "firmware" }));
await client.endAsync();

Run it against the Python responder above: the two libraries interoperate because both only use standard MQTT 5 properties. The code generator can scaffold the connection part for other languages.

Test cases worth automating

A request/response layer looks trivial until it fails in production. These cases cover most real bugs:

1. Happy path. One request, one reply, payload and correlation data match.
2. Responder offline. The requester must raise a timeout, not hang. Measure that it fires at the configured value.
3. Late reply. Make the responder sleep longer than the timeout. The late reply must be ignored, not delivered to the next request.
4. Concurrency. Fire 100 requests in parallel with a responder that adds random delay; every reply must land on the right caller. Out-of-order replies are normal.
5. Missing properties. Send a request without Response Topic. The responder should log and drop it, not crash.
6. Expired request. Stop the responder, send a request with a 2-second expiry, wait 5 seconds, restart the responder with a persistent session. It must not execute the request.
7. Authorization. With ACLs enabled, the responder needs publish rights on response topics and the requester needs subscribe rights on its own. A denied reply is often dropped without an error — see ACL denied silently.

When the reply never arrives

A useful habit: make the responder log the response topic and correlation data it received before processing. When a test fails, that single log line tells you which side is wrong.

Measuring round-trip latency

Because the requester timestamps the request and receives the reply on the same clock, request/response is an easy place to measure end-to-end latency. Loop the request() call a few hundred times and record p50 and p99. The round trip includes two broker hops, the responder's processing time and two network paths, so compare it against a single-hop latency test to see how much time your responder adds. Repeat at QoS 0 and QoS 1 — the extra PUBACKs show up directly in the numbers.

Watch requests and replies live

Request/response over MQTT 5 needs only two properties, but the details — subscribing before sending, timeouts, expiry, unique correlation IDs and correct ACLs — decide whether it is reliable. Keep the scripts above as a regression suite and run them against every broker or permission change.

Frequently asked questions