# Migrating from MQTT (/connectors/mqtt/scenarios/migration) You migrate an existing MQTT application to KubeMQ by **changing only the broker endpoint** — the host:port your client already connects to. This is a drop-in, endpoint-only migration: no KubeMQ SDK and no library swap. Your standard MQTT 3.1.1 / 5.0 client keeps publishing and subscribing exactly as it does today. The one thing to plan for is KubeMQ's topic grammar — the connector maps MQTT topics onto four messaging patterns through a topic-prefix convention, and a handful of common MQTT features are intentionally unsupported. Read this guide before cutting over so you know what works, what needs a topic rename, and what does not migrate at all. ## Overview [#overview] KubeMQ ships an embedded MQTT broker that speaks **MQTT 3.1.1 and MQTT 5.0** over TCP, TLS, and WebSocket. Existing MQTT clients point at KubeMQ by changing the broker host — no library swap required. The connector maps MQTT topics onto KubeMQ's messaging patterns through a topic-prefix convention, and several common MQTT features are intentionally unsupported. * **Default ports** — `1883` (TCP), `8883` (TLS), `8083` (WebSocket — serves `ws` or `wss` depending on the `Security` config). * **Canonical client** — Eclipse Paho `paho-mqtt 2.x` (Python). The examples in this guide use it; any conformant MQTT 3.1.1 / 5.0 client works. * **Protocol versions** — MQTT 3.1.1 and MQTT 5.0. MQTT 3.1 (protocol level 3) is rejected at CONNECT. The connector is **opt-in (disabled by default)** — a stock kubemq-server does **not** bind the MQTT listeners until you turn it on. Enable it with its enable variable, then connect: **The enable variable is `CONNECTORSMQTT_ENABLE` — there is no underscore between `CONNECTORS` and `MQTT`, and no `KUBEMQ_` prefix.** Variants like `CONNECTORS_MQTT_ENABLE` do **not** bind to the `Connectors.MQTT.Enable` field and are silently ignored. For Kubernetes, set `spec.mqtt.enabled: true` in the `KubemqCluster` CR. For a first connection, see [Getting started](/connectors/mqtt/tutorials/getting-started). ## Compatibility Matrix [#compatibility-matrix] This matrix summarizes what the MQTT connector supports. It is self-contained — read the footnotes for the load-bearing caveats. | Dimension | Support | Notes | | -------------------------------------- | ------------------------------------------------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------- | | **Drop-in level** | endpoint-only⁵ | Host-swap only; topic names must follow the prefix convention | | **Point-to-point queues** | ✅ | `queues/*` topics → KubeMQ Queues; `$share/{g}/queues/*` for consumption | | **Pub/sub (non-durable)** | ✅ | `events/*` and prefixless topics (when `DefaultPattern=events`) | | **Durable / persistent subscriptions** | ⚠️ | `store/*` → Events Store; subscriptions are always `StartNewOnly` — no replay | | **Request/reply (RPC)** | ✅ (v5 only) | MQTT 5.0 Response Topic + Correlation Data → Commands/Queries; MQTT 3.1.1 lacks Response Topic — RPC not available | | **Ordering guarantee** | ⚠️ QoS-dependent | QoS 2 de-duplicates at the wire level; end-to-end ordering is node-local | | **Transactions** | N/A | MQTT has no transaction concept | | **Dead-letter / redrive** | ❌ no client DLQ⁶ | Queue messages past `MaxReceiveCount` are silently dropped; no consumable dead-letter address over MQTT | | **Selectors / filtering / wildcards** | ⚠️ wildcards Events-only | `+` / `#` wildcards supported on `events/*` subscriptions only; rejected (SUBACK 0xA2) on `store/`, `queues/`, `commands/`, `queries/` | | **Auth model** | JWT (password) | CONNECT password = KubeMQ JWT when auth is enabled; username is audit-only | | **TLS / mTLS** | ✅ 8883 / 8083 | Server TLS and mutual TLS on 8883; WebSocket 8083 serves `wss` when `Security` is configured (same port as plain `ws`) | | **Top unsupported** | MQTT 3.1; retained messages; clients-as-responders; wildcards on non-Events; node-local sessions | | Footnote ⁵: MQTT **3.1** clients (protocol level 3) are rejected at CONNECT. Use 3.1.1 (Paho default, protocol level 4) or 5.0. Footnote ⁶: There is no client-settable DLQ over this protocol. See [No client-settable DLQ on queue subscriptions](#no-client-settable-dlq-on-queue-subscriptions) below for details. ## Connection / Endpoint Migration [#connection--endpoint-migration] Replace the broker host and port. The MQTT protocol and your client library stay the same. **No code changes are needed for the connection itself** — only topic names may need renaming (see [Concept & Destination Mapping](#concept--destination-mapping) below). ```bash title="Broker endpoints" # Before (any MQTT broker) mqtt://broker.example.com:1883 mqtts://broker.example.com:8883 # After (KubeMQ) mqtt://kubemq.example.com:1883 mqtts://kubemq.example.com:8883 # WebSocket — single listener on 8083 # Serves plain ws:// when Security is not configured, # or wss:// (TLS) on the same port when Security is configured. ws://kubemq.example.com:8083/ # plain (no Security block) # wss://kubemq.example.com:8083/ # TLS (when Security is configured) ``` ### Protocol version requirement [#protocol-version-requirement] The connector enforces `MinProtocolVersion = 4` (MQTT 3.1.1) by default. MQTT 3.1 clients (protocol level 3) are rejected at CONNECT. Eclipse Paho 2.x defaults to MQTT 3.1.1, so no version flag is needed unless you want MQTT 5.0. ```python # paho-mqtt 2.x — explicitly request MQTT 5.0 (optional; 3.1.1 is the default) import paho.mqtt.client as mqtt client = mqtt.Client( callback_api_version=mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5, # omit for 3.1.1 (the Paho default) ) ``` ### Authentication [#authentication] When the KubeMQ authentication service is enabled, the CONNECT **password** must be a valid KubeMQ JWT. The username field is accepted and recorded for audit but is not validated. ```python client.username_pw_set(username="device-01", password="") ``` When authentication is disabled (the default), all connections are accepted regardless of credentials — exactly like the gRPC and REST listeners. If the MQTT ports are reachable from untrusted networks, either enable authentication or restrict access at the network layer. ## Concept & Destination Mapping [#concept--destination-mapping] The first segment of every MQTT topic selects a KubeMQ messaging pattern. The remainder of the topic path (after the prefix slash) becomes the KubeMQ **channel name**, with `/` translated to `.`. | MQTT topic | KubeMQ pattern | KubeMQ channel | Notes | | -------------------------- | ---------------- | ------------------ | -------------------------------------- | | `events/sensor/temp` | Events | `sensor.temp` | Pub/sub, at-most-once | | `store/audit/login` | Events Store | `audit.login` | Persistent, `StartNewOnly` — no replay | | `queues/orders/new` | Queues | `orders.new` | At-least-once; consume via `$share` | | `commands/device/reboot` | Commands | `device.reboot` | RPC; MQTT 5.0 only | | `queries/inventory/status` | Queries | `inventory.status` | RPC; MQTT 5.0 only | | `sensor/temp` | `DefaultPattern` | `sensor.temp` | Prefixless topic routed by config | **Prefix collision caution.** If your existing topics already use a top-level segment named `events`, `store`, `queues`, `commands`, or `queries`, those topics will be interpreted as KubeMQ pattern prefixes rather than literal topic names. Rename those segments before migrating. ### Prefixless topics [#prefixless-topics] Topics whose first segment is not a reserved prefix are routed according to `DefaultPattern` (default: `events`), so basic pub/sub keeps working with no topic renames. Set `DefaultPattern = none` to reject prefixless topics explicitly. ### Wildcard subscriptions [#wildcard-subscriptions] KubeMQ supports the two standard MQTT wildcards on **`events/` subscriptions only**: * **`+`** — single-level wildcard (matches exactly one topic segment). * **`#`** — multi-level wildcard (matches the rest of the topic tree). Wildcard use on any other pattern is rejected; exact-filter subscriptions work on all patterns. See [Wildcards on non-Events patterns — REJECTED](#wildcards-on-non-events-patterns--rejected) for the exact SUBACK codes and full behavior. ### Queue consumption (shared subscriptions) [#queue-consumption-shared-subscriptions] Plain subscriptions to `queues/*` are rejected (SUBACK 0x83). Queue consumption requires a shared subscription with QoS ≥ 1: ```text $share/{group}/queues/{path} ``` Example: `$share/workers/queues/orders/new` The group name is recorded for metrics and audit but does not create independent per-group message copies — all groups and all nodes compete in the same KubeMQ queue pool. ### `/` and `.` conflation in channel names [#-and--conflation-in-channel-names] KubeMQ uses `.` as the channel hierarchy separator. The topic mapper replaces `/` with `.`, so both `devices/a/b/temp` and `devices/a.b.temp` map to the same channel `devices.a.b.temp`. Design channel names to avoid ambiguity if you have topics that mix both separators. ## Canonical Client Example [#canonical-client-example] **Client:** Eclipse Paho `paho-mqtt 2.x` The snippets below cover the core flows you migrate: publish an event, subscribe and consume events, consume a queue via `$share`, an MQTT 5.0 RPC request, and a TLS connection. ### Install [#install] ```bash title="Terminal" pip install paho-mqtt>=2.0.0 ``` ### Publish an event [#publish-an-event] ```python import paho.mqtt.client as mqtt def on_connect(client, userdata, flags, reason_code, properties): print(f"Connected: reason_code={reason_code}") client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2) # When KubeMQ authentication is enabled, set password = KubeMQ JWT: # client.username_pw_set(username="device-01", password="") client.on_connect = on_connect client.connect("kubemq.example.com", 1883) client.loop_start() # Publish to Events pattern → channel "sensor.temp" result = client.publish("events/sensor/temp", payload='{"celsius": 21.5}', qos=1) result.wait_for_publish() client.loop_stop() client.disconnect() ``` ### Subscribe and consume events [#subscribe-and-consume-events] ```python import paho.mqtt.client as mqtt def on_connect(client, userdata, flags, reason_code, properties): print(f"Connected: reason_code={reason_code}") # Subscribe to Events with a single-level '+' wildcard (Events-only) client.subscribe("events/sensor/+", qos=1) def on_message(client, userdata, message): print(f"Topic: {message.topic} Payload: {message.payload.decode()}") client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2) client.on_connect = on_connect client.on_message = on_message client.connect("kubemq.example.com", 1883) client.loop_forever() ``` ### Consume a queue (at-least-once) [#consume-a-queue-at-least-once] Queue consumption requires a shared subscription and QoS ≥ 1. The PUBACK doubles as the queue acknowledge — the message is requeued if the PUBACK is not sent within `QueueAckTimeoutSeconds` (default 30 s). ```python import paho.mqtt.client as mqtt def on_connect(client, userdata, flags, reason_code, properties): print(f"Connected: reason_code={reason_code}") # $share group name is audit-only; all groups share the same queue pool client.subscribe("$share/workers/queues/orders/new", qos=1) def on_message(client, userdata, message): print(f"Queue message: {message.payload.decode()}") # paho-mqtt 2.x auto-sends PUBACK for QoS 1 when on_message returns. # The KubeMQ connector treats PUBACK receipt as the queue acknowledgment. client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2) client.on_connect = on_connect client.on_message = on_message client.connect("kubemq.example.com", 1883) client.loop_forever() ``` ### RPC (MQTT 5.0 only) [#rpc-mqtt-50-only] RPC requires MQTT 5.0 (Response Topic + Correlation Data). MQTT 3.1.1 does not support RPC on this connector — see [RPC over MQTT 3.1.1 — NOT AVAILABLE](#rpc-over-mqtt-311--not-available) for the exact behavior. ```python import paho.mqtt.client as mqtt from paho.mqtt.properties import Properties from paho.mqtt.packettypes import PacketTypes import uuid CLIENT_ID = "req-client-01" REPLY_TOPIC = f"$reply/{CLIENT_ID}/inventory" def on_connect(client, userdata, flags, reason_code, properties): print(f"Connected: reason_code={reason_code}") # Subscribe to our own $reply inbox before sending the request client.subscribe(REPLY_TOPIC, qos=1) def on_message(client, userdata, message): print(f"RPC response: {message.payload.decode()}") # Check user properties for command outcome: # kubemq-executed = "true"/"false" (Commands) # kubemq-metadata, kubemq-error (Queries / error detail) client = mqtt.Client( callback_api_version=mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID, protocol=mqtt.MQTTv5, ) client.on_connect = on_connect client.on_message = on_message client.connect("kubemq.example.com", 1883) client.loop_start() # Build MQTT 5.0 publish properties pub_props = Properties(PacketTypes.PUBLISH) pub_props.ResponseTopic = REPLY_TOPIC pub_props.CorrelationData = str(uuid.uuid4()).encode() # The responder is a gRPC/REST/CloudEvents client subscribed to channel "inventory.status" client.publish( "queries/inventory/status", payload='{"sku":"WIDGET-100"}', qos=1, properties=pub_props, ) import time; time.sleep(5) # wait for response client.loop_stop() client.disconnect() ``` ### TLS connection [#tls-connection] ```python client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2) client.tls_set( ca_certs="/path/to/ca.crt", # CA that signed the KubeMQ server cert certfile="/path/to/client.crt", # omit for server-TLS-only (no mTLS) keyfile="/path/to/client.key", ) client.connect("kubemq.example.com", 8883) ``` For WebSocket TLS: `client.connect("kubemq.example.com", 8083, transport="websockets")` with the same `tls_set` call above. Port 8083 serves `wss://` when `Security` is configured (the same port that serves `ws://` without TLS). ## Security [#security] ### Authentication [#authentication-1] Authentication is **connect-time only**. A JWT that expires mid-connection does not terminate the connection — the token is validated once at CONNECT and cached for the connection's lifetime. | When auth is ... | Behavior | | ------------------ | -------------------------------------------------------------------------------------------------------- | | Disabled (default) | All CONNECT attempts accepted; username recorded for audit | | Enabled | CONNECT `password` must be a valid KubeMQ JWT; empty or invalid → CONNACK 0x86 (MQTT 5.0) / rc=5 (3.1.1) | ### Authorization [#authorization] When the KubeMQ authorization service (Casbin) is configured, every publish and subscribe is checked against the client's resolved identity. ACL denial returns PUBACK 0x87 / SUBACK 0x87 (MQTT 5.0) or silent drop / SUBACK 0x80 (MQTT 3.1.1); the `$reply//...` namespace is always allowed. ### TLS / mTLS [#tls--mtls] The dedicated TLS listener on **8883** activates only when the global `Security` block is configured with certificate material; without it, 8883 is skipped at startup with a warning. The **WebSocket listener on 8083** is a single listener that always starts when `WsPort` is set: it serves plain `ws://` when `Security` is not configured, and switches to `wss://` (TLS) on the same port when `Security` is configured. The plain TCP listener on **1883** starts unconditionally when the connector is enabled. * **Server TLS** — `Security.Mode = tls`: presents `Security.Cert` / `Security.Key`; clients verify the server certificate; minimum TLS 1.2. * **Mutual TLS** — `Security.Mode = mtls`: additionally requires clients to present a certificate verified against `Security.Ca`. ### Opt-in activation [#opt-in-activation] The connector is **disabled by default**. Explicitly enable it: ```toml title="config.toml" [Connectors.MQTT] Enable = true Port = "1883" ``` Or via environment variable: ```bash title="Terminal" export CONNECTORSMQTT_ENABLE=true ``` ## What Does NOT Migrate / Documented Deviations [#what-does-not-migrate--documented-deviations] ### MQTT 3.1 (protocol level 3) — REJECTED [#mqtt-31-protocol-level-3--rejected] Clients using MQTT 3.1 (protocol level 3, e.g. very old Paho 1.x defaults) receive `CONNACK` with "unsupported protocol version". Upgrade to 3.1.1 (`mqtt.MQTTv311`) or 5.0 (`mqtt.MQTTv5`). Paho 2.x defaults to 3.1.1 — no change needed for modern clients. ### Retained messages — REJECTED AND AUDITED (not silently dropped) [#retained-messages--rejected-and-audited-not-silently-dropped] `RetainAvailable` is forced to `0` at server construction. Any publish with the retain flag set is **rejected with an audit event** (`publish.error`) and is **not routed and not stored**. The MQTT 5.0 client receives a normal success PUBACK (the retain flag on a PUBACK has no standard meaning in MQTT), but the message is discarded before reaching KubeMQ. MQTT 3.1.1 clients also have their retained publishes discarded with an audit event. There is no path to convert retained messages to Events Store entries. A Will (LWT) message with the retain flag set is rejected at CONNECT (CONNACK 0x9A), since retain is unavailable. Use an LWT without the retain flag. ### MQTT clients as RPC responders — NOT SUPPORTED [#mqtt-clients-as-rpc-responders--not-supported] A `SUBSCRIBE` to `commands/*` or `queries/*` is rejected with SUBACK 0x83. MQTT clients can only **issue** RPC requests (as publishers); RPC responders must be gRPC, REST, or CloudEvents clients. This is an architectural constraint. ### RPC over MQTT 3.1.1 — NOT AVAILABLE [#rpc-over-mqtt-311--not-available] MQTT 3.1.1 has no Response Topic field. A publish to `commands/*` or `queries/*` from a 3.1.1 client is silently dropped and audited (`publish.error` + WARN). RPC is exclusively an MQTT 5.0 capability on this connector. ### Wildcards on non-Events patterns — REJECTED [#wildcards-on-non-events-patterns--rejected] Wildcard subscriptions (`+`, `#`) are supported **only** on the `events/` prefix (and on prefixless topics when `DefaultPattern=events`). Attempts to use wildcards on `store/`, `queues/`, `commands/`, or `queries/` receive SUBACK 0xA2 (MQTT 5.0) or SUBACK 0x80 (MQTT 3.1.1). Exact-filter subscriptions work on all patterns. ### Sessions — node-local, in-memory, lost on restart [#sessions--node-local-in-memory-lost-on-restart] MQTT session state (subscriptions, client-ID registry, LWT, QoS state machines) is held in memory on the node that accepted the connection. Consequences: * Process restart clears all session state, including clean-session=0 sessions. * A clean-session=0 client that reconnects to a **different cluster node** finds no session. * Duplicate client-ID from different nodes keeps both connections live (no cross-node takeover). * Cross-node RPC responses are lost: if the requester's connection migrates to another node mid-request, the response is not forwarded. ### Events Store replay — not available over MQTT [#events-store-replay--not-available-over-mqtt] Subscriptions to `store/*` are always `StartNewOnly`. Historical replay from the Events Store is not exposed over the MQTT wire protocol. Replay is available via gRPC, REST, and CloudEvents clients only. ### No client-settable DLQ on queue subscriptions [#no-client-settable-dlq-on-queue-subscriptions] The connector never sets a per-message receive limit on messages it publishes to KubeMQ Queues. A "poison" message that exceeds the server's `MaxReceiveCount` is dropped by the server; there is no consumable dead-letter address accessible over MQTT. See the AWS or RabbitMQ guides for connectors that support DLQ/redrive. ### `$SYS` topics [#sys-topics] The `$SYS` broker-internal topic tree is not available. Subscriptions to `$SYS/*` are denied. ### Events Store subscriptions and the `$share` prefix [#events-store-subscriptions-and-the-share-prefix] `$share` subscriptions are valid **only** for the `queues/*` pattern. `$share` on `store/*`, `events/*`, `commands/*`, or `queries/*` is rejected with SUBACK 0x83. ### User properties (MQTT 5.0 ↔ KubeMQ Tags) [#user-properties-mqtt-50--kubemq-tags] MQTT 5.0 user properties map to KubeMQ message tags and vice versa. Caps: **32 properties** and **4096 bytes** total. Exceeding either cap returns PUBACK 0x97. MQTT 3.1.1 clients do not support user properties — tag metadata is not surfaced to 3.1.1 subscribers. ## Verification Smoke Test [#verification-smoke-test] This recipe confirms that the connector is reachable, that topics are mapped correctly, and that basic publish/consume works. It uses `paho-mqtt 2.x` (the canonical client for this guide) as a "publish one → consume it → confirm arrival" check. ### Step 1 — Enable the connector [#step-1--enable-the-connector] ```toml title="config.toml" [Connectors.MQTT] Enable = true Port = "1883" ``` Restart KubeMQ. Confirm the log line `mqtt connector started on port 1883`. ### Step 2 — Publish and receive an event [#step-2--publish-and-receive-an-event] Run these two snippets in separate terminals. **Terminal 1 — subscriber:** ```python import paho.mqtt.client as mqtt received = [] def on_connect(client, userdata, flags, rc, props): client.subscribe("events/smoke/test", qos=1) def on_message(client, userdata, msg): received.append(msg.payload.decode()) print(f"RECEIVED: {msg.payload.decode()}") client.disconnect() client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2) client.on_connect = on_connect client.on_message = on_message client.connect("localhost", 1883) client.loop_forever() assert received == ['{"ok":true}'], f"Expected message not received: {received}" print("Smoke test PASSED") ``` **Terminal 2 — publisher (after the subscriber is connected):** ```python import paho.mqtt.client as mqtt client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2) client.connect("localhost", 1883) client.loop_start() result = client.publish("events/smoke/test", payload='{"ok":true}', qos=1) result.wait_for_publish() client.loop_stop() client.disconnect() print("Published") ``` Expected: the subscriber prints `RECEIVED: {"ok":true}` and exits with `Smoke test PASSED`. ### Step 3 — Confirm a queue round-trip [#step-3--confirm-a-queue-round-trip] **Publish to a queue:** ```python client.publish("queues/smoke/test", payload='{"job":1}', qos=1) ``` **Consume from the queue (shared subscription, QoS 1):** ```python client.subscribe("$share/smokers/queues/smoke/test", qos=1) ``` Expected: the consumer receives `{"job":1}` exactly once; no redelivery unless the consumer disconnects before the PUBACK. ## See Also [#see-also]