Testes de integração MQTT em Java com JUnit 5

Um teste de integração MQTT em Java conecta um cliente real a um broker real, se inscreve, publica e verifica o que chega. Com JUnit 5, a parte difícil não é a API, e sim a entrega assíncrona: as mensagens chegam em outra thread, um pouco depois, ou nunca. Este guia mostra uma configuração de testes que lida com isso corretamente usando Eclipse Paho, o HiveMQ MQTT Client, Awaitility e Testcontainers.

Dependências

Os exemplos usam Maven; as mesmas coordenadas funcionam no Gradle. As versões eram as atuais no momento da escrita; releases de patch mais novos também servem.

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>

Você só precisa do cliente MQTT que sua aplicação realmente usa. Se as bibliotecas são novidade para você, comece por MQTT em Java com Paho ou pelo guia do HiveMQ MQTT Client.

Regras para testes MQTT que não ficam instáveis

Regra	Por quê
Client ID único por teste	Dois clientes com o mesmo ID se desconectam mutuamente ("session taken over")
Prefixo de topic único por teste	Testes paralelos e mensagens retidas antigas não vazam para suas asserções
Inscreva-se antes de publicar e espere o SUBACK	Uma mensagem publicada antes de a inscrição estar ativa simplesmente não é entregue
Espere com timeout, nunca com `Thread.sleep`	Rápido quando funciona, uma falha clara quando não funciona
Desconecte no `@AfterEach`	Conexões esquecidas esgotam os limites do broker e mantêm sessões vivas
Endereço do broker vindo da configuração	Os mesmos testes rodam contra Testcontainers localmente e um broker hospedado na CI

O gerador de client IDs mostra formatos seguros em qualquer broker; em testes, um prefixo mais um sufixo aleatório basta.

Um teste com Paho e Awaitility

O MqttClient síncrono do Paho bloqueia connect() e subscribe() até o broker confirmá-los, o que simplifica o setup do teste. A entrega continua acontecendo na thread de callbacks do Paho, então colete as mensagens em uma fila thread-safe e deixe o Awaitility consultá-la.

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 faz diferença: a persistência padrão em arquivo grava uma pasta por client ID no diretório de trabalho, o que suja seu diretório de build e pode colidir entre execuções paralelas.

O mesmo teste com o HiveMQ MQTT Client

O cliente HiveMQ suporta MQTT 5. Sua API bloqueante tem uma fila publishes() ideal para testes: registre-a antes de se inscrever e depois chame receive() com 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));
    }
  }
}

Para TLS, adicione .sslWithDefaultConfig() e a porta 8883; para credenciais, adicione .simpleAuth().username(...).password(...).applySimpleAuth() à chamada de conexão.

Um broker descartável com Testcontainers

Para testes que precisam rodar offline ou exigem uma configuração específica do broker, suba um container de broker por classe de teste. O Mosquitto 2.x só escuta em localhost se não receber configuração, então use a configuração sem autenticação que vem na imagem:

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();
  }
}

Um container static sobe uma vez por classe; um campo de instância subiria um por método de teste, o que é lento. O Testcontainers também tem um módulo HiveMQ (org.testcontainers:hivemq) se você quiser testar contra o HiveMQ CE. Os prós e contras estão em testes MQTT com Testcontainers.

Container local ou broker hospedado?

Um container prova que seu código funciona contra um broker. Ele não prova que seu código funciona contra o seu broker — aquele com TLS, autenticação, ACLs e o caminho de rede que seus dispositivos usam. Uma boa divisão:

Testes unitários e de componentes: Testcontainers, sem dependência de rede, feedback rápido.
Testes de integração na CI: um broker de testes hospedado com a mesma configuração de TLS e credenciais da produção, selecionado por MQTT_URL, MQTT_USER e MQTT_PASS.
Smoke tests após o deploy: o broker de produção, um topic de teste dedicado, asserções somente leitura.

Testando casos negativos

Testes de integração também devem provar o que não pode acontecer. Dois padrões:

Falha de conexão esperada. Com Paho, assertThrows(MqttSecurityException.class, () -> client.connect(badOptions)) verifica que credenciais erradas são rejeitadas; getReasonCode() retorna 4 ou 5 para "bad username or password" e "not authorized" no MQTT 3.1.1. Consulte qualquer outro código na ferramenta de reason codes.
Silêncio esperado. Para provar que uma ACL bloqueia um topic, inscreva-se, espere um tempo limitado e verifique que a fila continua vazia: await().during(Duration.ofSeconds(2)).atMost(Duration.ofSeconds(3)).until(received::isEmpty). during faz o Awaitility verificar que a condição permanece verdadeira.

Rodando a suíte

Marque os testes que dependem de rede com @Tag("integration") e rode-os com o Maven Failsafe (*IT.java) ou uma task separada do Gradle, para que mvn test continue rápido e offline. Na CI, exporte as variáveis do broker e rode mvn verify; testes MQTT no GitHub Actions mostra um workflow completo.

Enquanto um teste roda, abra o cliente MQTT online e inscreva-se em testmqtt/junit/# para ver o tráfego que sua suíte gera — o jeito mais rápido de descobrir se um teste que falhou nunca publicou ou publicou no topic errado.

Veja ao vivo o tráfego dos seus testes JUnit

Bons testes MQTT com JUnit 5 se resumem a isolamento e paciência: um client ID e um topic únicos por teste, uma inscrição confirmada antes de publicar e esperas limitadas em vez de sleeps. Mantenha o endereço do broker configurável e a mesma suíte vai cobrir um container no seu notebook, um broker de CI e um smoke test pós-deploy. Para uma estratégia mais ampla, veja testes de integração MQTT.

Perguntas frequentes

Devo usar Thread.sleep em testes MQTT?

Não. Um sleep fixo é curto demais em um runner de CI lento ou desperdiça tempo em um rápido. Use Awaitility ou um CompletableFuture com timeout: o teste continua assim que a mensagem chega e falha com uma mensagem clara quando ela não chega.

Paho ou cliente HiveMQ para testes com JUnit?

Teste com o cliente que sua aplicação usa. Eclipse Paho (mqttv3) é o mais comum para MQTT 3.1.1; o HiveMQ MQTT Client suporta MQTT 5 e oferece APIs bloqueante, assíncrona e reativa. Os padrões deste guia funcionam com os dois.

Como evito que testes MQTT em paralelo interfiram entre si?

Dê a cada teste um client ID único e um prefixo de topic único, por exemplo testmqtt/junit/<id-aleatorio>/. Assim os testes rodam em paralelo no mesmo broker sem receber mensagens uns dos outros nem se derrubarem por client IDs duplicados.