Pruebas de integración MQTT en Java con JUnit 5

Una prueba de integración MQTT en Java conecta un cliente real a un broker real, se suscribe, publica y comprueba lo que llega. Con JUnit 5 lo difícil no es la API, sino la entrega asíncrona: los mensajes llegan en otro hilo, algo más tarde o nunca. Esta guía muestra una configuración de pruebas que lo resuelve correctamente con Eclipse Paho, el HiveMQ MQTT Client, Awaitility y Testcontainers.

Dependencias

Los ejemplos usan Maven; las mismas coordenadas sirven en Gradle. Las versiones eran las actuales al momento de escribir; las versiones de parche más nuevas también valen.

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>

Solo necesitas el cliente MQTT que usa realmente tu aplicación. Si eres nuevo en estas librerías, empieza por MQTT en Java con Paho o la guía del HiveMQ MQTT Client.

Reglas para pruebas MQTT que no fallen al azar

Regla	Por qué
Client ID único por prueba	Dos clientes con el mismo ID se desconectan entre sí ("session taken over")
Prefijo de topic único por prueba	Ni las pruebas en paralelo ni los mensajes retenidos antiguos se cuelan en tus aserciones
Suscríbete antes de publicar y espera el SUBACK	Un mensaje publicado antes de que la suscripción esté activa simplemente no se entrega
Espera con timeout, nunca con `Thread.sleep`	Rápido cuando funciona, un fallo claro cuando no
Desconecta en `@AfterEach`	Las conexiones olvidadas agotan los límites del broker y mantienen sesiones vivas
Dirección del broker desde la configuración	Las mismas pruebas corren contra Testcontainers en local y un broker alojado en CI

El generador de client IDs muestra formatos seguros en cualquier broker; en pruebas basta con un prefijo más un sufijo aleatorio.

Una prueba con Paho y Awaitility

El MqttClient síncrono de Paho bloquea connect() y subscribe() hasta que el broker los confirma, lo que simplifica la preparación de la prueba. La entrega sigue ocurriendo en el hilo de callbacks de Paho, así que recoge los mensajes en una cola thread-safe y deja que Awaitility la consulte.

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 importa: la persistencia por defecto en ficheros escribe una carpeta por client ID en el directorio de trabajo, lo que ensucia tu directorio de build y puede chocar entre ejecuciones en paralelo.

La misma prueba con el HiveMQ MQTT Client

El cliente HiveMQ soporta MQTT 5. Su API bloqueante tiene una cola publishes() ideal para pruebas: regístrala antes de suscribirte y luego llama a receive() con un 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, añade .sslWithDefaultConfig() y el puerto 8883; para credenciales, añade .simpleAuth().username(...).password(...).applySimpleAuth() a la llamada de conexión.

Un broker desechable con Testcontainers

Para pruebas que deben correr sin red o que necesitan una configuración concreta del broker, arranca un contenedor de broker por clase de prueba. Mosquitto 2.x solo escucha en localhost si no recibe configuración, así que usa la configuración sin autenticación incluida en la imagen:

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

Un contenedor static arranca una vez por clase; un campo de instancia arrancaría uno por método de prueba, lo cual es lento. Testcontainers también incluye un módulo HiveMQ (org.testcontainers:hivemq) si quieres probar contra HiveMQ CE. Las ventajas e inconvenientes se explican en pruebas MQTT con Testcontainers.

¿Contenedor local o broker alojado?

Un contenedor demuestra que tu código funciona contra un broker. No demuestra que funcione contra tu broker: el que tiene TLS, autenticación, ACLs y la ruta de red que usan tus dispositivos. Un buen reparto:

Pruebas unitarias y de componentes: Testcontainers, sin dependencia de red, feedback rápido.
Pruebas de integración en CI: un broker de pruebas alojado con la misma configuración de TLS y credenciales que producción, seleccionado mediante MQTT_URL, MQTT_USER y MQTT_PASS.
Smoke tests tras el despliegue: el broker de producción, un topic de prueba dedicado, aserciones de solo lectura.

Probar los casos negativos

Las pruebas de integración también deben demostrar lo que no debe ocurrir. Dos patrones:

Fallo de conexión esperado. Con Paho, assertThrows(MqttSecurityException.class, () -> client.connect(badOptions)) comprueba que se rechazan credenciales incorrectas; getReasonCode() devuelve 4 o 5 para "bad username or password" y "not authorized" en MQTT 3.1.1. Consulta cualquier otro código con la herramienta de reason codes.
Silencio esperado. Para comprobar que una ACL bloquea un topic, suscríbete, espera un tiempo acotado y verifica que la cola sigue vacía: await().during(Duration.ofSeconds(2)).atMost(Duration.ofSeconds(3)).until(received::isEmpty). during hace que Awaitility compruebe que la condición se mantiene verdadera.

Ejecutar la suite

Etiqueta las pruebas que dependen de la red con @Tag("integration") y ejecútalas con Maven Failsafe (*IT.java) o una tarea de Gradle aparte, para que mvn test siga siendo rápido y sin red. En CI, exporta las variables del broker y ejecuta mvn verify; pruebas MQTT en GitHub Actions muestra un workflow completo.

Mientras corre una prueba, abre el cliente MQTT online y suscríbete a testmqtt/junit/# para ver el tráfico que genera tu suite: es la forma más rápida de saber si una prueba fallida nunca publicó o publicó en el topic equivocado.

Mira en vivo el tráfico de tus pruebas JUnit

Las buenas pruebas MQTT con JUnit 5 se reducen a aislamiento y paciencia: un client ID y un topic únicos por prueba, una suscripción confirmada antes de publicar y esperas acotadas en lugar de sleeps. Mantén configurable la dirección del broker y la misma suite cubrirá un contenedor en tu portátil, un broker de CI y un smoke test tras el despliegue. Para una estrategia más amplia, consulta pruebas de integración MQTT.

Preguntas frecuentes

¿Debo usar Thread.sleep en las pruebas MQTT?

No. Un sleep fijo es demasiado corto en un runner de CI lento o desperdicia tiempo en uno rápido. Usa Awaitility o un CompletableFuture con timeout: la prueba continúa en cuanto llega el mensaje y falla con un mensaje claro cuando no llega.

¿Paho o el cliente HiveMQ para pruebas con JUnit?

Prueba con el cliente que usa tu aplicación. Eclipse Paho (mqttv3) es el más habitual para MQTT 3.1.1; el HiveMQ MQTT Client soporta MQTT 5 y ofrece APIs bloqueante, asíncrona y reactiva. Los patrones de esta guía funcionan con ambos.

¿Cómo evito que las pruebas MQTT en paralelo interfieran entre sí?

Da a cada prueba un client ID único y un prefijo de topic único, por ejemplo testmqtt/junit/<id-aleatorio>/. Así las pruebas pueden ejecutarse en paralelo contra el mismo broker sin recibir mensajes ajenos ni expulsarse mutuamente por client IDs duplicados.