MQTT Integration Tests in Java with JUnit 5

An MQTT integration test in Java connects a real client to a real broker, subscribes, publishes and asserts on what arrives. With JUnit 5 the hard part is not the API — it is the asynchronous delivery: messages arrive on another thread, some time later, or never. This guide shows a test setup that handles that correctly with Eclipse Paho, the HiveMQ MQTT Client, Awaitility and Testcontainers.

Dependencies

The examples use Maven; the same coordinates work in Gradle. Versions were current at the time of writing — newer patch releases are fine.

pom.xml (test dependencies)

<dependencies>
  <dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter</artifactId>
    <version>5.11.3</version>
    <scope>test</scope>
  </dependency>
  <dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.5</version>
  </dependency>
  <dependency>
    <groupId>com.hivemq</groupId>
    <artifactId>hivemq-mqtt-client</artifactId>
    <version>1.3.3</version>
  </dependency>
  <dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <version>4.2.2</version>
    <scope>test</scope>
  </dependency>
  <dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>junit-jupiter</artifactId>
    <version>1.20.3</version>
    <scope>test</scope>
  </dependency>
</dependencies>

You only need the MQTT client your application actually uses. If you are new to the libraries, start with MQTT in Java with Paho or the HiveMQ MQTT Client guide.

Rules for MQTT tests that don't flake

Rule	Why
Unique client ID per test	Two clients with the same ID disconnect each other ("session taken over")
Unique topic prefix per test	Parallel tests and old retained messages can't leak into your assertions
Subscribe before you publish, and wait for SUBACK	A message published before the subscription is active is simply not delivered
Wait with a timeout, never `Thread.sleep`	Fast when it works, a clear failure when it doesn't
Disconnect in `@AfterEach`	Leaked connections exhaust broker limits and keep sessions alive
Broker address from configuration	The same tests run against Testcontainers locally and a hosted broker in CI

The client ID generator shows formats that are safe across brokers; in tests, a prefix plus a random suffix is enough.

A Paho test with Awaitility

The synchronous MqttClient from Paho blocks connect() and subscribe() until the broker has acknowledged them, which makes test setup simple. Delivery still happens on Paho's callback thread, so collect messages in a thread-safe queue and let Awaitility poll it.

PahoRoundTripTest.java

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;

import java.time.Duration;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.jupiter.api.*;

class PahoRoundTripTest {

  static final String BROKER =
      System.getenv().getOrDefault("MQTT_URL", "tcp://broker.connectmq.com:1883");

  MqttClient client;
  String topic;

  @BeforeEach
  void connect() throws MqttException {
    String id = UUID.randomUUID().toString().substring(0, 8);
    topic = "testmqtt/junit/" + id + "/temperature";
    client = new MqttClient(BROKER, "junit-" + id, new MemoryPersistence());

    MqttConnectOptions options = new MqttConnectOptions();
    options.setCleanSession(true);
    options.setConnectionTimeout(10);
    String user = System.getenv("MQTT_USER");
    if (user != null) {
      options.setUserName(user);
      options.setPassword(System.getenv("MQTT_PASS").toCharArray());
    }
    client.connect(options);
  }

  @AfterEach
  void disconnect() throws MqttException {
    if (client.isConnected()) client.disconnect();
    client.close();
  }

  @Test
  void publishedMessageIsReceived() throws MqttException {
    var received = new ConcurrentLinkedQueue<String>();
    client.subscribe(topic, 1, (t, msg) -> received.add(new String(msg.getPayload(), UTF_8)));

    client.publish(topic, "21.5".getBytes(UTF_8), 1, false);

    await().atMost(Duration.ofSeconds(5))
        .untilAsserted(() -> assertEquals("21.5", received.peek()));
  }
}

MemoryPersistence matters: the default file persistence writes a folder per client ID into the working directory, which litters your build directory and can collide between parallel runs.

The same test with the HiveMQ MQTT Client

The HiveMQ client supports MQTT 5. Its blocking API has a publishes() queue that is ideal for tests: register it before subscribing, then receive() with a timeout.

HiveMqRoundTripTest.java

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.jupiter.api.Assertions.*;

import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
import com.hivemq.client.mqtt.mqtt5.Mqtt5Client;
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.*;

class HiveMqRoundTripTest {

  Mqtt5BlockingClient client;
  String topic;

  @BeforeEach
  void connect() {
    String id = UUID.randomUUID().toString().substring(0, 8);
    topic = "testmqtt/junit/" + id + "/status";
    client = Mqtt5Client.builder()
        .identifier("junit-" + id)
        .serverHost("broker.connectmq.com")
        .serverPort(1883)
        .buildBlocking();
    client.connect();
  }

  @AfterEach
  void disconnect() {
    client.disconnect();
  }

  @Test
  void statusMessageIsDelivered() throws InterruptedException {
    try (var publishes = client.publishes(MqttGlobalPublishFilter.ALL)) {
      client.subscribeWith().topicFilter(topic).qos(MqttQos.AT_LEAST_ONCE).send();
      client.publishWith().topic(topic).qos(MqttQos.AT_LEAST_ONCE)
          .payload("online".getBytes(UTF_8)).send();

      Optional<Mqtt5Publish> msg = publishes.receive(5, TimeUnit.SECONDS);
      assertTrue(msg.isPresent(), "no message within 5 s on " + topic);
      assertEquals("online", new String(msg.get().getPayloadAsBytes(), UTF_8));
    }
  }
}

For TLS, add .sslWithDefaultConfig() and port 8883; for credentials, add .simpleAuth().username(...).password(...).applySimpleAuth() to the connect call.

A throwaway broker with Testcontainers

For tests that must run offline or need a specific broker configuration, start a broker container per test class. Mosquitto 2.x only listens on localhost unless it gets a configuration, so use the bundled no-auth config:

LocalBrokerTest.java

import org.junit.jupiter.api.Test;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;

@Testcontainers
class LocalBrokerTest {

  @Container
  static GenericContainer<?> mosquitto =
      new GenericContainer<>(DockerImageName.parse("eclipse-mosquitto:2.0"))
          .withExposedPorts(1883)
          .withCommand("mosquitto", "-c", "/mosquitto-no-auth.conf")
          .waitingFor(Wait.forLogMessage(".*mosquitto version.*running.*", 1));

  static String brokerUrl() {
    return "tcp://" + mosquitto.getHost() + ":" + mosquitto.getMappedPort(1883);
  }

  @Test
  void brokerAcceptsConnections() throws Exception {
    var client = new org.eclipse.paho.client.mqttv3.MqttClient(
        brokerUrl(), "junit-local",
        new org.eclipse.paho.client.mqttv3.persist.MemoryPersistence());
    client.connect();
    client.disconnect();
    client.close();
  }
}

A static container starts once per class; an instance field would start one per test method, which is slow. Testcontainers also ships a HiveMQ module (org.testcontainers:hivemq) if you want to test against HiveMQ CE. The trade-offs are covered in MQTT tests with Testcontainers.

Local container or hosted broker?

A container proves your code works against a broker. It does not prove your code works against your broker — the one with TLS, authentication, ACLs and the network path your devices use. A good split:

Unit and component tests: Testcontainers, no network dependency, fast feedback.
Integration tests in CI: a hosted test broker with the same TLS and credential setup as production, selected through MQTT_URL, MQTT_USER and MQTT_PASS.
Smoke tests after deploy: the production broker, a dedicated test topic, read-only assertions.

Testing negative cases

Integration tests should also prove what must not happen. Two patterns:

Expected connection failure. With Paho, assertThrows(MqttSecurityException.class, () -> client.connect(badOptions)) checks that wrong credentials are rejected; getReasonCode() returns 4 or 5 for MQTT 3.1.1 "bad username or password" and "not authorized". Look up any other code with the reason code tool.
Expected silence. To assert that an ACL blocks a topic, subscribe and wait a bounded time, then assert the queue is still empty: await().during(Duration.ofSeconds(2)).atMost(Duration.ofSeconds(3)).until(received::isEmpty). during makes Awaitility check that the condition stays true.

Running the suite

Tag network-dependent tests with @Tag("integration") and run them with Maven Failsafe (*IT.java) or a separate Gradle task, so mvn test stays fast and offline. In CI, export the broker variables and run mvn verify; MQTT testing in GitHub Actions shows a complete workflow.

While a test is running, open the online MQTT client and subscribe to testmqtt/junit/# to watch the traffic your suite generates — the fastest way to see whether a failing test never published or published to the wrong topic.

Watch your JUnit test traffic live

Good MQTT tests in JUnit 5 come down to isolation and patience: a unique client ID and topic per test, a subscription that is acknowledged before you publish, and bounded waits instead of sleeps. Keep the broker address configurable and the same suite will cover a laptop container, a CI broker and a post-deploy smoke test. For broader strategy, see MQTT integration testing.

Frequently asked questions

Should I use Thread.sleep in MQTT tests?

No. A fixed sleep is either too short on a slow CI runner or wastes time on a fast one. Use Awaitility or a CompletableFuture with a timeout, so the test continues as soon as the message arrives and fails with a clear message when it does not.

Paho or HiveMQ client for JUnit tests?

Test with the client your application uses. Eclipse Paho (mqttv3) is the most common for MQTT 3.1.1; the HiveMQ MQTT Client supports MQTT 5 and offers blocking, async and reactive APIs. The testing patterns in this guide work with both.

How do I keep parallel MQTT tests from interfering?

Give every test a unique client ID and a unique topic prefix, for example testmqtt/junit/<random-id>/. Then tests can run in parallel against the same broker without receiving each other's messages or kicking each other off with duplicate client IDs.