# MQTT (/connectors/mqtt)



Point your MQTT 3.1.1 / 5.0 application at KubeMQ by changing only the broker address.
The **MQTT connector** is a built-in, wire-protocol bridge inside kubemq-server: an
embedded MQTT broker that speaks the standard protocol natively, so any off-the-shelf
MQTT client (paho, MQTT.js, MQTTnet, rumqttc) reaches KubeMQ's Events, Events-Store,
Queues, Commands, and Queries with no KubeMQ SDK, no library swap, and no code rewrite.

## What is the MQTT connector [#what-is-the-mqtt-connector]

[MQTT](https://mqtt.org/) is a lightweight publish/subscribe protocol built around
*topics*: a client connects, subscribes to topic filters, and publishes messages with a
chosen *Quality of Service* (QoS 0/1/2). The KubeMQ MQTT connector exposes this protocol
on its own dedicated ports and maps it onto **all five KubeMQ messaging patterns**.

The **first segment of the MQTT topic** selects the KubeMQ pattern, and the remaining
segments become the KubeMQ channel with `/` translated to `.`. A publish to
`events/site1/temp` lands on the Events channel `site1.temp`; `queues/jobs` targets a
KubeMQ Queue; `commands/restart` and `queries/status` drive RPC. The connector is a
*gateway*, not a client library — your application only needs a stock MQTT client.

Key capabilities:

* **All five patterns over one wire** — Events, Events-Store, Queues, Commands, and
  Queries, selected by the topic prefix.
* **Topic-driven pattern routing** — `events/`, `store/`, `queues/`, `commands/`,
  `queries/` prefixes map a topic to a KubeMQ pattern; a bare (prefixless) topic routes
  to the configured `DefaultPattern` (`events` by default).
* **MQTT 3.1.1 and 5.0** — both protocol levels on the same listener; MQTT 5.0 unlocks
  User-Properties (carried as KubeMQ Tags) and RPC (Commands / Queries).
* **Cross-protocol interop** — a message published over MQTT to `events/it/cross` is
  consumable by a gRPC or REST KubeMQ client on channel `it.cross`, and vice-versa.

<Callout type="warn">
  **Retain is silently dropped.** The connector forces `RetainAvailable=0` in `CONNACK`. A
  runtime publish with the retain flag set returns a success PUBACK (`0x00`) but the message
  is stripped and **never delivered or stored**. A *Will*-retain requested at CONNECT time is
  rejected outright with CONNACK `0x9A`. There are also **no durable subscriptions** — MQTT
  sessions are in-memory and node-local. See
  [QoS and sessions](/connectors/mqtt/concepts/qos-and-sessions).
</Callout>

## How it works [#how-it-works]

An MQTT client connects to the connector and publishes to (or subscribes to) a topic. The
connector resolves the topic prefix to a KubeMQ `(pattern, channel)` pair, hands the
message to the message broker, and consumers on the same channel — over MQTT or any other
KubeMQ transport — receive it.

<Mermaid
  chart="`
graph LR
APP[&#x22;MQTT client<br/>(any language)&#x22;]
CONN[&#x22;MQTT connector<br/>:1883 / :8883 / :8083&#x22;]
MAP[&#x22;Topic mapper<br/>prefix → pattern&#x22;]
BROKER[&#x22;Message Broker&#x22;]
SUB[&#x22;Consumer<br/>(MQTT / gRPC / REST)&#x22;]

APP -- &#x22;PUBLISH events/site1/temp&#x22; --> CONN
CONN --> MAP
MAP -- &#x22;(pattern, channel)&#x22; --> BROKER
BROKER -. &#x22;deliver&#x22; .-> SUB

class APP,SUB client
class CONN,MAP connector
class BROKER broker
`"
/>

*The embedded broker accepts a standard MQTT connection, the topic mapper resolves the prefix to a KubeMQ pattern and channel, and the message broker fans it out to consumers on any transport.*

## Ports & protocol surface [#ports--protocol-surface]

| Port   | Transport            | Protocol         | Notes                                                                                                                                   |
| ------ | -------------------- | ---------------- | --------------------------------------------------------------------------------------------------------------------------------------- |
| `1883` | Plain TCP            | MQTT 3.1.1 & 5.0 | Default listener. Disable by setting `CONNECTORSMQTT_PORT=""`.                                                                          |
| `8883` | TLS over TCP         | MQTT 3.1.1 & 5.0 | Binds only when the server-global `Security` block (cert + key) is configured; otherwise the port is open but the listener is inactive. |
| `8083` | WebSocket (path `/`) | MQTT 3.1.1 & 5.0 | MQTT-over-WebSocket at path `/`. Upgraded to `wss://` when `Security` is configured.                                                    |

Clients select the transport by URL scheme: `tcp://host:1883`, `tls://host:8883`, or
`ws://host:8083/`. The examples read one environment variable, `KUBEMQ_MQTT_URL` (default
`tcp://localhost:1883`). MQTT 3.1 (protocol level 3) is always rejected; the minimum
accepted level is 3.1.1. See [Configuration](/connectors/mqtt/concepts/configuration) for the
listener and capability settings.

## Publish an event [#publish-an-event]

The example below connects an MQTT 5.0 client and publishes one message to
`events/<channel>` at QoS 1. The topic prefix `events/` selects the Events pattern, and
`/` becomes `.` in the channel, so `events/demo/x` lands on the KubeMQ channel `demo.x`.
Every client reads the broker endpoint from `KUBEMQ_MQTT_URL` (default
`tcp://localhost:1883`).

<Tabs groupId="language" items="['Go','Python','Java','JavaScript','C#','Ruby','Rust']">
  <Tab value="Go">
    ```go
    package main

    import (
    	"context"
    	"fmt"
    	"log"
    	"net"
    	"os"
    	"strings"
    	"time"

    	"github.com/eclipse/paho.golang/paho"
    )

    func brokerURL() string {
    	if u := os.Getenv("KUBEMQ_MQTT_URL"); u != "" {
    		return u
    	}
    	return "tcp://localhost:1883"
    }

    // tcpAddr strips the scheme prefix to obtain "host:port".
    func tcpAddr(rawURL string) string {
    	for _, pfx := range []string{"tcp://", "ws://", "tls://"} {
    		if strings.HasPrefix(rawURL, pfx) {
    			return rawURL[len(pfx):]
    		}
    	}
    	return rawURL
    }

    func main() {
    	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
    	defer cancel()

    	// Prefix "events/" selects the Events pattern; '/' becomes '.' so
    	// "events/demo/x" maps to the KubeMQ channel "demo.x".
    	const topic = "events/demo/x"

    	conn, err := net.Dial("tcp", tcpAddr(brokerURL()))
    	if err != nil {
    		log.Fatalf("dial: %v", err)
    	}
    	client := paho.NewClient(paho.ClientConfig{Conn: conn})

    	connAck, err := client.Connect(ctx, &paho.Connect{
    		ClientID:   "go-mqtt-events-pub",
    		KeepAlive:  30,
    		CleanStart: true,
    	})
    	if err != nil {
    		log.Fatalf("connect: %v", err)
    	}
    	if connAck.ReasonCode != 0 {
    		log.Fatalf("CONNACK reason=0x%02X", connAck.ReasonCode)
    	}

    	// QoS 1 returns a PUBACK. Do NOT set Retain — a retained publish is
    	// silently dropped (PUBACK 0x00, message never delivered).
    	pubAck, err := client.Publish(ctx, &paho.Publish{
    		Topic:   topic,
    		QoS:     1,
    		Payload: []byte("hello from MQTT"),
    		Properties: &paho.PublishProperties{
    			// MQTT 5.0 User Properties round-trip as KubeMQ Tags.
    			User: []paho.UserProperty{{Key: "sensor", Value: "thermometer"}},
    		},
    	})
    	if err != nil {
    		log.Fatalf("publish: %v", err)
    	}
    	if pubAck.ReasonCode != 0 {
    		log.Fatalf("PUBACK reason=0x%02X", pubAck.ReasonCode)
    	}
    	fmt.Printf("published to %q (channel demo.x)\n", topic)

    	_ = client.Disconnect(&paho.Disconnect{ReasonCode: 0})
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    import os

    import paho.mqtt.client as mqtt
    from paho.mqtt.enums import CallbackAPIVersion
    from paho.mqtt.properties import Properties
    from paho.mqtt.packettypes import PacketTypes


    def parse_url(url: str) -> tuple[str, int]:
        scheme, rest = url.split("://", 1)
        host_port = rest.rstrip("/")
        if ":" in host_port:
            host, port = host_port.rsplit(":", 1)
            return host, int(port)
        return host_port, {"tcp": 1883, "tls": 8883, "ws": 8083}.get(scheme, 1883)


    def main() -> None:
        url = os.environ.get("KUBEMQ_MQTT_URL", "tcp://localhost:1883")
        host, port = parse_url(url)

        # Prefix "events/" selects the Events pattern; "/" becomes "." so
        # "events/demo/x" maps to the KubeMQ channel "demo.x".
        topic = "events/demo/x"

        # paho-mqtt 2.x requires an explicit callback API version.
        client = mqtt.Client(
            callback_api_version=CallbackAPIVersion.VERSION2,
            client_id="python-mqtt-events-pub",
            protocol=mqtt.MQTTv5,
        )
        client.connect(host, port, keepalive=30, clean_start=True)
        client.loop_start()

        # MQTT 5.0 User Properties round-trip as KubeMQ Tags.
        props = Properties(PacketTypes.PUBLISH)
        props.UserProperty = [("sensor", "thermometer")]

        # Do not set retain=True — a retained publish is silently dropped.
        info = client.publish(topic, payload=b"hello from MQTT", qos=1, properties=props)
        info.wait_for_publish(timeout=10)
        print(f"published to {topic!r} (channel demo.x)")

        client.loop_stop()
        client.disconnect()


    if __name__ == "__main__":
        main()
    ```
  </Tab>

  <Tab value="Java">
    ```java
    import java.nio.charset.StandardCharsets;
    import java.util.List;

    import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
    import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
    import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
    import org.eclipse.paho.mqttv5.common.MqttMessage;
    import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
    import org.eclipse.paho.mqttv5.common.packet.UserProperty;

    public final class Main {
        public static void main(String[] args) throws Exception {
            String broker = System.getenv().getOrDefault("KUBEMQ_MQTT_URL", "tcp://localhost:1883");

            // Prefix "events/" selects the Events pattern; "/" becomes "." so
            // "events/demo/x" maps to the KubeMQ channel "demo.x".
            String topic = "events/demo/x";

            MqttConnectionOptions opts = new MqttConnectionOptions();
            opts.setCleanStart(true);
            opts.setKeepAliveInterval(30);

            // org.eclipse.paho.mqttv5.client always uses MQTT protocol level 5.
            MqttAsyncClient client = new MqttAsyncClient(broker, "java-mqtt-events-pub", new MemoryPersistence());
            client.connect(opts).waitForCompletion(10_000);

            MqttMessage msg = new MqttMessage("hello from MQTT".getBytes(StandardCharsets.UTF_8));
            msg.setQos(1);
            // Do NOT call msg.setRetained(true) — a retained publish is silently dropped.

            // MQTT 5.0 User Properties round-trip as KubeMQ Tags.
            MqttProperties props = new MqttProperties();
            props.setUserProperties(List.of(new UserProperty("sensor", "thermometer")));
            msg.setProperties(props);

            client.publish(topic, msg).waitForCompletion(10_000);
            System.out.printf("published to %s (channel demo.x)%n", topic);

            client.disconnect().waitForCompletion(5_000);
            client.close();
        }
    }
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    import mqtt, { type MqttClient } from "mqtt";

    function brokerUrl(): string {
      return process.env["KUBEMQ_MQTT_URL"] ?? "tcp://localhost:1883";
    }

    async function main(): Promise<void> {
      // Prefix "events/" selects the Events pattern; "/" becomes "." so
      // "events/demo/x" maps to the KubeMQ channel "demo.x".
      const topic = "events/demo/x";

      const client: MqttClient = mqtt.connect(brokerUrl(), {
        clientId: "js-mqtt-events-pub",
        protocolVersion: 5,
        clean: true,
        keepalive: 30,
      });

      await new Promise<void>((resolve, reject) => {
        client.once("connect", () => resolve());
        client.once("error", reject);
      });

      await new Promise<void>((resolve, reject) => {
        client.publish(
          topic,
          "hello from MQTT",
          {
            qos: 1,
            retain: false, // retain is NOT supported; a retained publish is silently dropped.
            // MQTT 5.0 User Properties round-trip as KubeMQ Tags.
            properties: { userProperties: { sensor: "thermometer" } },
          },
          (err) => (err ? reject(err) : resolve()),
        );
      });
      console.log(`published to "${topic}" (channel demo.x)`);

      await client.endAsync();
    }

    main().catch((err) => {
      console.error(err);
      process.exit(1);
    });
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    using System.Text;
    using MQTTnet;
    using MQTTnet.Client;
    using MQTTnet.Protocol;

    static (string host, int port) ParseEndpoint(string url)
    {
        foreach (var prefix in new[] { "tcp://", "tls://", "ws://" })
            if (url.StartsWith(prefix, StringComparison.OrdinalIgnoreCase))
                url = url[prefix.Length..];
        var parts = url.TrimEnd('/').Split(':');
        return (parts[0], parts.Length > 1 ? int.Parse(parts[1]) : 1883);
    }

    var url = Environment.GetEnvironmentVariable("KUBEMQ_MQTT_URL") ?? "tcp://localhost:1883";
    var (host, port) = ParseEndpoint(url);

    // Prefix "events/" selects the Events pattern; "/" becomes "." so
    // "events/demo/x" maps to the KubeMQ channel "demo.x".
    const string topic = "events/demo/x";

    var factory = new MqttFactory();
    using var client = factory.CreateMqttClient();

    var options = new MqttClientOptionsBuilder()
        .WithTcpServer(host, port)
        .WithClientId("csharp-mqtt-events-pub")
        .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500)
        .WithCleanSession(true)
        .Build();

    await client.ConnectAsync(options);

    var pubResult = await client.PublishAsync(
        new MqttApplicationMessageBuilder()
            .WithTopic(topic)
            .WithPayload(Encoding.UTF8.GetBytes("hello from MQTT"))
            .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
            // MQTT 5.0 User Property round-trips as a KubeMQ Tag.
            // Do NOT call WithRetainFlag(true) — a retained publish is silently dropped.
            .WithUserProperty("sensor", "thermometer")
            .Build());

    if (pubResult.ReasonCode != MqttClientPublishReasonCode.Success)
        throw new Exception($"PUBACK reason: {pubResult.ReasonCode}");
    Console.WriteLine($"published to '{topic}' (channel demo.x)");

    await client.DisconnectAsync();
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # The Ruby `mqtt` gem speaks MQTT 3.1.1 only (no User Properties, no v5 RPC).
    require "mqtt"
    require "uri"

    uri = URI.parse(ENV.fetch("KUBEMQ_MQTT_URL", "tcp://localhost:1883"))

    # Prefix "events/" selects the Events pattern; "/" becomes "." so
    # "events/demo/x" maps to the KubeMQ channel "demo.x".
    topic = "events/demo/x"

    MQTT::Client.connect(
      host:          uri.host,
      port:          uri.port,
      ssl:           uri.scheme == "tls",
      client_id:     "ruby-mqtt-events-pub",
      clean_session: true,
      keep_alive:    30
    ) do |client|
      # retain=false is mandatory — the broker silently drops retained publishes.
      client.publish(topic, "hello from MQTT", false, 1)
      puts "published to '#{topic}' (channel demo.x)"
    end
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    use rumqttc::v5::mqttbytes::v5::PublishProperties;
    use rumqttc::v5::mqttbytes::QoS;
    use rumqttc::v5::{AsyncClient, MqttOptions};
    use std::env;
    use std::time::Duration;

    fn parse_host_port(url: &str) -> (String, u16) {
        let stripped = url
            .trim_start_matches("tcp://")
            .trim_start_matches("tls://")
            .trim_start_matches("ws://");
        let host_port = stripped.split('/').next().unwrap_or(stripped);
        let mut parts = host_port.splitn(2, ':');
        let host = parts.next().unwrap_or("localhost").to_string();
        let port: u16 = parts.next().and_then(|p| p.parse().ok()).unwrap_or(1883);
        (host, port)
    }

    #[tokio::main]
    async fn main() -> Result<(), Box<dyn std::error::Error>> {
        let url = env::var("KUBEMQ_MQTT_URL").unwrap_or_else(|_| "tcp://localhost:1883".to_string());
        let (host, port) = parse_host_port(&url);

        // Prefix "events/" selects the Events pattern; '/' becomes '.' so
        // "events/demo/x" maps to the KubeMQ channel "demo.x".
        let topic = "events/demo/x";

        let mut opts = MqttOptions::new("rust-mqtt-events-pub", host, port);
        opts.set_keep_alive(Duration::from_secs(30));
        let (client, mut eventloop) = AsyncClient::new(opts, 10);

        // Drive the event loop so the PUBLISH is flushed and the PUBACK is processed.
        tokio::spawn(async move { while eventloop.poll().await.is_ok() {} });

        // MQTT 5.0 User Properties round-trip as KubeMQ Tags.
        let props = PublishProperties {
            user_properties: vec![("sensor".to_string(), "thermometer".to_string())],
            ..Default::default()
        };

        // retain = false — a retained publish is silently dropped.
        client
            .publish_with_properties(topic, QoS::AtLeastOnce, false, b"hello from MQTT".as_ref(), props)
            .await?;
        println!("published to '{topic}' (channel demo.x)");

        Ok(())
    }
    ```
  </Tab>
</Tabs>

## Supported languages [#supported-languages]

The connector speaks standard MQTT, so any conformant client works. The examples pin one
native MQTT client per language — there is no KubeMQ SDK and no proto bindings.

| Language                | Client library                                                                                  | Protocol   | Notes                                                          |
| ----------------------- | ----------------------------------------------------------------------------------------------- | ---------- | -------------------------------------------------------------- |
| Go                      | [`eclipse/paho.golang`](https://github.com/eclipse/paho.golang) (`paho.mqtt.golang` for v3.1.1) | MQTT 5.0   | The connector's reference client.                              |
| Python                  | [`paho-mqtt`](https://pypi.org/project/paho-mqtt/) ≥ 2.1                                        | MQTT 5.0   | Callback-API v2.                                               |
| Java                    | [Eclipse Paho `mqttv5`](https://eclipse.dev/paho/)                                              | MQTT 5.0   | `org.eclipse.paho.mqttv5.client`.                              |
| JavaScript / TypeScript | [`mqtt.js`](https://github.com/mqttjs/MQTT.js) ≥ 5.15                                           | MQTT 5.0   | Works over TCP, TLS, and WebSocket.                            |
| C# / .NET               | [`MQTTnet`](https://github.com/dotnet/MQTTnet) ≥ 4.3.7                                          | MQTT 5.0   | Task-based async.                                              |
| Ruby                    | [`mqtt` gem](https://github.com/njh/ruby-mqtt) ≥ 0.6                                            | MQTT 3.1.1 | v3.1.1 subset only — no User Properties, no RPC, no WebSocket. |
| Rust                    | [`rumqttc`](https://github.com/bytebeamio/rumqtt) ≥ 0.24                                        | MQTT 5.0   | async/await on Tokio.                                          |

<Callout type="info">
  The Ruby `mqtt` gem implements **MQTT 3.1.1 only**. The MQTT 5.0-only features — RPC
  (Commands / Queries), shared-subscription queue consume, and User-Properties — are not
  available from Ruby. Use a v5 client (Go, Python, Java, JavaScript, C#, Rust) for those
  patterns. See [Topic grammar](/connectors/mqtt/reference/topic-grammar).
</Callout>

## Next steps [#next-steps]

<Cards>
  <Card title="Getting started" href="/connectors/mqtt/tutorials/getting-started" description="Connect, publish, and subscribe end-to-end through the MQTT connector in minutes." />

  <Card title="Configuration" href="/connectors/mqtt/concepts/configuration" description="The 17 connector settings, the CONNECTORSMQTT_ENABLE spelling, forced capabilities, and validation." />

  <Card title="Events" href="/connectors/mqtt/how-to/events" description="Fire-and-forget pub/sub over MQTT — topic prefixes, wildcards, and User-Properties as Tags." />

  <Card title="Topic grammar" href="/connectors/mqtt/reference/topic-grammar" description="The full topic grammar — prefixes, wildcards, $share queue consume, and $reply." />
</Cards>