MQTT in Java with Eclipse Paho

Eclipse Paho Java is the classic MQTT client for the JVM, and it comes in two flavours: org.eclipse.paho.client.mqttv3 for MQTT 3.1.1 and org.eclipse.paho.mqttv5.client for MQTT 5. This tutorial connects both to a broker over TLS, explains the callback and threading rules that cause most production bugs, and shows how Paho's persistence and offline buffer behave.

Examples use broker.connectmq.com on TLS port 8883, topics under testmqtt/java/... and credentials from MQTT_USER / MQTT_PASS. Java 11 or later works; the code was compiled with Java 21.

Pick the right Paho artifact

	MQTT 3.1.1	MQTT 5
Maven artifact	`org.eclipse.paho:org.eclipse.paho.client.mqttv3`	`org.eclipse.paho:org.eclipse.paho.mqttv5.client`
Package	`org.eclipse.paho.client.mqttv3`	`org.eclipse.paho.mqttv5.client` (+ `…mqttv5.common`)
Options class	`MqttConnectOptions`	`MqttConnectionOptions`
Callback	`MqttCallback` / `MqttCallbackExtended`	`MqttCallback` (six methods)
Session flag	`setCleanSession`	`setCleanStart` + `setSessionExpiryInterval`
Password type	`char[]`	`byte[]`

Both are at version 1.2.5 and are mature but receive few changes. If you want a reactive or more actively developed API, compare it with the HiveMQ MQTT client in Java MQTT libraries compared.

pom.xml

<dependency>
  <groupId>org.eclipse.paho</groupId>
  <artifactId>org.eclipse.paho.mqttv5.client</artifactId>
  <version>1.2.5</version>
</dependency>
<!-- or, for MQTT 3.1.1 -->
<dependency>
  <groupId>org.eclipse.paho</groupId>
  <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
  <version>1.2.5</version>
</dependency>

build.gradle.kts

implementation("org.eclipse.paho:org.eclipse.paho.mqttv5.client:1.2.5")
// implementation("org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.5")

MQTT 3.1.1 client (mqttv3)

MqttClient is the blocking API; every call waits for the broker's answer. MqttCallbackExtended adds connectComplete, which fires after the first connect and after every automatic reconnect — the right place to subscribe:

TelemetryV3.java

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

public class TelemetryV3 {
    static final String BROKER = "ssl://broker.connectmq.com:8883";
    static final String TOPIC = "testmqtt/java/v3/telemetry";

    public static void main(String[] args) throws Exception {
        String clientId = "java-v3-" + UUID.randomUUID().toString().substring(0, 8);
        MqttClient client = new MqttClient(BROKER, clientId, new MemoryPersistence());

        MqttConnectOptions opts = new MqttConnectOptions();
        String user = System.getenv("MQTT_USER");
        if (user != null) {
            opts.setUserName(user);
            opts.setPassword(System.getenv("MQTT_PASS").toCharArray());
        }
        opts.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
        opts.setCleanSession(true);
        opts.setAutomaticReconnect(true);
        opts.setKeepAliveInterval(30);
        opts.setConnectionTimeout(10);

        client.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                System.out.println((reconnect ? "reconnected to " : "connected to ") + serverURI);
                try {
                    client.subscribe(TOPIC, 1); // clean session: subscribe again after every reconnect
                } catch (MqttException e) {
                    System.err.println("subscribe failed: " + e);
                }
            }

            @Override
            public void connectionLost(Throwable cause) {
                System.err.println("connection lost: " + cause.getMessage());
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
                System.out.println(topic + " -> " + new String(message.getPayload(), StandardCharsets.UTF_8));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) { }
        });

        client.connect(opts);

        MqttMessage msg = new MqttMessage("{\"temp\":21.4}".getBytes(StandardCharsets.UTF_8));
        msg.setQos(1);
        client.publish(TOPIC, msg);

        Thread.sleep(3000);
        client.disconnect();
        client.close();
    }
}

Details that matter:

ssl:// means TLS. Paho uses the JVM's default trust store, so a broker with a publicly trusted certificate needs no extra code. For a private CA, build an SSLSocketFactory from your trust store and pass it with opts.setSocketFactory(...). WebSockets use wss://broker.connectmq.com:8084/mqtt.
Set the protocol version explicitly. By default Paho tries 3.1.1 and silently falls back to 3.1 if the first attempt fails, so a real error can show up as Invalid protocol version (1).
Unique client IDs. Two clients with the same ID take turns disconnecting each other. The client ID generator produces safe ones.

MQTT 5 client (mqttv5)

The MQTT 5 client looks similar but lives in other packages, and its callback has six methods. connectWithResult returns the CONNACK token, so you can read the session-present flag; disconnected tells you the broker's reason code instead of a generic exception:

TelemetryV5.java

import org.eclipse.paho.mqttv5.client.*;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

public class TelemetryV5 {
    static final String BROKER = "ssl://broker.connectmq.com:8883";
    static final String TOPIC = "testmqtt/java/v5/telemetry";

    public static void main(String[] args) throws Exception {
        String clientId = "java-v5-" + UUID.randomUUID().toString().substring(0, 8);
        MqttClient client = new MqttClient(BROKER, clientId, new MemoryPersistence());

        MqttConnectionOptions opts = new MqttConnectionOptions();
        String user = System.getenv("MQTT_USER");
        if (user != null) {
            opts.setUserName(user);
            opts.setPassword(System.getenv("MQTT_PASS").getBytes(StandardCharsets.UTF_8));
        }
        opts.setCleanStart(true);
        opts.setAutomaticReconnect(true);
        opts.setKeepAliveInterval(30);
        opts.setConnectionTimeout(10);

        client.setCallback(new MqttCallback() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                System.out.println((reconnect ? "reconnected to " : "connected to ") + serverURI);
            }

            @Override
            public void disconnected(MqttDisconnectResponse response) {
                System.err.println("disconnected: " + response.getReasonString()
                        + " (reason code " + response.getReturnCode() + ")");
            }

            @Override
            public void mqttErrorOccurred(MqttException e) {
                System.err.println("MQTT error: " + e.getMessage());
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
                System.out.println(topic + " -> " + new String(message.getPayload(), StandardCharsets.UTF_8));
            }

            @Override
            public void deliveryComplete(IMqttToken token) { }

            @Override
            public void authPacketArrived(int reasonCode, MqttProperties properties) { }
        });

        IMqttToken connack = client.connectWithResult(opts);
        System.out.println("session present: " + connack.getSessionPresent());

        client.subscribe(TOPIC, 1);

        MqttMessage msg = new MqttMessage("{\"temp\":21.4}".getBytes(StandardCharsets.UTF_8));
        msg.setQos(1);
        MqttProperties props = new MqttProperties();
        props.setContentType("application/json");
        props.setMessageExpiryInterval(60L);
        msg.setProperties(props);
        client.publish(TOPIC, msg);

        Thread.sleep(3000);
        client.disconnect();
        client.close();
    }
}

The numbers in disconnected and in exceptions are MQTT 5 reason codes; look them up in the reason code reference. authPacketArrived is only used for enhanced authentication and can stay empty.

Callbacks and threading

Paho delivers every message for a client on one callback thread. That single fact explains most "Paho is slow" and "Paho randomly disconnects" reports:

Keep messageArrived short. While it runs, further messages queue up in memory and then back up on the network. Slow database writes belong on another thread.
Never let it throw. An exception thrown from messageArrived shuts the connection down; QoS 1 and 2 messages are redelivered after the reconnect, so a poison message can loop forever.
Do not disconnect from inside a callback. Paho cannot acknowledge the message being processed and the call deadlocks. Publishing a reply from the callback is allowed.

A small executor fixes all three:

Hand work to a worker thread

ExecutorService worker = Executors.newSingleThreadExecutor(); // single thread keeps message order

@Override
public void messageArrived(String topic, MqttMessage message) {
    byte[] payload = message.getPayload();
    worker.submit(() -> {
        try {
            handle(topic, payload);
        } catch (Exception e) {
            System.err.println("handler failed for " + topic + ": " + e);
        }
    });
}

Use a pool instead of a single thread when order does not matter. The same rules apply in frameworks built on Paho, such as Spring Boot with Spring Integration MQTT.

Persistence, sessions and the offline buffer

Paho's persistence stores in-flight QoS 1 and 2 messages until the broker acknowledges them. It is not a message database:

Option	What it does	When to use
`MemoryPersistence`	In-flight messages in RAM, lost on restart	Most apps, tests, containers
`MqttDefaultFilePersistence(dir)`	In-flight messages on disk, survive restart	Gateways that must not lose QoS 1/2 messages
Default constructor `new MqttClient(uri, id)`	File persistence in the working directory	Avoid: surprising files and lock clashes

File persistence only pays off together with a persistent session (setCleanSession(false) in v3, setCleanStart(false) plus setSessionExpiryInterval(...) in v5) and a stable client ID. The async client can also buffer publishes while offline:

Gateway with file persistence and offline buffer (v3)

MqttAsyncClient client = new MqttAsyncClient(
        "ssl://broker.connectmq.com:8883",
        "gateway-07",                                   // stable ID: the session belongs to it
        new MqttDefaultFilePersistence("/var/lib/myapp/mqtt"));

DisconnectedBufferOptions buffer = new DisconnectedBufferOptions();
buffer.setBufferEnabled(true);
buffer.setBufferSize(5000);
buffer.setPersistBuffer(true);
buffer.setDeleteOldestMessages(true);
client.setBufferOpts(buffer);

MqttConnectOptions opts = new MqttConnectOptions();
opts.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
opts.setCleanSession(false);
opts.setAutomaticReconnect(true);
opts.setMaxInflight(100);
client.connect(opts).waitForCompletion(10_000);

The broker side of the session is explained in MQTT persistent sessions.

Common Paho errors

Message (code)	Usual cause
`Unable to connect to server (32103)`	Wrong host or port, firewall, or TLS on a plain port
`Bad user name or password (4)`	Wrong credentials, or none sent
`Not authorized to connect (5)`	Credentials accepted but the user may not connect
`Timed out waiting for a response from the server (32000)`	No CONNACK/SUBACK in time: network or overloaded broker
`Connection lost (32109)`	Socket dropped: keep-alive timeout, duplicate client ID, exception in `messageArrived`
`Persistence already in use (32200)`	Two instances share an ID and a persistence directory
`Too many publishes in progress (32202)`	More in-flight QoS 1/2 messages than `setMaxInflight` allows

Each one is covered step by step in Paho Java errors.

Test it

Open the online MQTT client, connect to broker.connectmq.com and subscribe to testmqtt/java/#.
Run TelemetryV3 or TelemetryV5. You should see connected to ssl://broker.connectmq.com:8883 in the console and {"temp":21.4} in the browser.
Publish a message from the browser to testmqtt/java/v5/telemetry while the program runs; it is printed by messageArrived.
Disable your network for longer than the keep-alive interval, then restore it. Expect connection lost (v3) or disconnected (v5) followed by reconnected to …, and messages flowing again thanks to the resubscribe in connectComplete.

For automated checks, the same flow fits into a JUnit test with a real broker, as shown in MQTT testing with JUnit.

Watch the Java test topics

Paho Java is dependable once you respect its rules: choose mqttv3 or mqttv5 deliberately, use unique client IDs and explicit persistence, subscribe in connectComplete, and keep the callback thread free. Test reconnects and sessions against a real broker before trusting them in production.

Frequently asked questions

Should I use the Paho mqttv3 or mqttv5 client?

Use org.eclipse.paho.mqttv5.client if your broker supports MQTT 5 and you want reason codes, session expiry, message expiry or user properties. Use org.eclipse.paho.client.mqttv3 for MQTT 3.1.1 brokers or existing code. They are separate artifacts with separate packages, so a project can depend on both.

What does "Persistence already in use (32200)" mean?

Two client instances with the same client ID and server URI are using the same file persistence directory, usually because the default MqttClient constructor writes to the working directory. Give each client a unique ID, pass MemoryPersistence, or point MqttDefaultFilePersistence at a directory per instance.

Why did my Paho client disconnect after an exception in messageArrived?

Paho treats an exception thrown from messageArrived as fatal and shuts the connection down; QoS 1 and 2 messages are redelivered after reconnecting. Catch exceptions inside the callback, or hand the payload to an executor and handle errors there.