MQTT con Apache NiFi: configuración de ConsumeMQTT y PublishMQTT

Apache NiFi se conecta a MQTT con dos procesadores estándar: ConsumeMQTT se suscribe a un filtro de topic y convierte los mensajes entrantes en FlowFiles, y PublishMQTT envía el contenido de un FlowFile a un topic. La configuración mínima es un Broker URI como tcp://broker.connectmq.com:1883, un Topic Filter o Topic, una QoS y —en ConsumeMQTT— un Max Queue Size. Todo lo demás (TLS, credenciales, sesiones, records, clúster) decide si el flujo es fiable en producción.

Este tutorial repasa cada propiedad importante y luego construye y prueba un flujo que consume telemetría JSON, la filtra con SQL y publica alertas de vuelta en MQTT.

Arranca NiFi para pruebas

NiFi 2.x funciona sobre Java 21; la imagen Docker oficial es la forma más rápida de tener la interfaz:

docker run --name nifi -p 8443:8443 \
  -e SINGLE_USER_CREDENTIALS_USERNAME=admin \
  -e SINGLE_USER_CREDENTIALS_PASSWORD=change-me-to-12-chars-min \
  -d apache/nifi:latest

Abre https://localhost:8443/nifi (acepta el certificado autofirmado) e inicia sesión. Fija un tag de imagen concreto para todo lo que vayas a conservar. La contraseña debe tener al menos 12 caracteres o el contenedor no arrancará.

Propiedades de ConsumeMQTT

Arrastra un procesador al canvas, busca ConsumeMQTT y abre sus propiedades. Estas son las que configurarás en cada flujo:

El procesador escribe cada mensaje en la relación Message y añade estos atributos: mqtt.broker, mqtt.topic, mqtt.qos, mqtt.isDuplicate y mqtt.isRetained. Sin record reader, cada mensaje MQTT se convierte en un FlowFile y su contenido es el payload en bruto.

Agrupar con records

Un FlowFile por mensaje está bien para unos pocos mensajes por segundo y es un derroche por encima de eso. Pon Record Reader en un JsonTreeReader y Record Writer en un JsonRecordSetWriter, y ConsumeMQTT escribirá todos los mensajes recibidos desde la última ejecución como registros en un único FlowFile. Con Add attributes as fields en true, cada registro recibe además los campos _topic, _qos, _isDuplicate e _isRetained, así no pierdes el topic cuando mensajes de muchos dispositivos comparten un FlowFile. Los payloads que el reader no puede parsear van a la relación parse.failure: envíala a un funnel o a un log para que los dispositivos defectuosos sean visibles; consulta errores de parseo JSON para ver las causas típicas.

TLS

Cambia el URI a ssl://broker.connectmq.com:8883 (o wss://broker.connectmq.com:8084/mqtt detrás de proxies que solo dejan pasar HTTPS) y define SSL Context Service con un StandardSSLContextService. Para un broker con un certificado de confianza pública, el truststore puede apuntar al archivo cacerts de la JVM; para una CA privada, importa la CA en un truststore dedicado. Si el handshake falla, compara con una prueba de openssl s_client: probar una conexión MQTT con TLS muestra los comandos.

Propiedades de PublishMQTT

PublishMQTT comparte las propiedades de conexión (Broker URI, versión, Client ID, credenciales, SSL) y añade:

Sin record reader, todo el contenido del FlowFile es un único payload. Compara el tamaño con el tamaño máximo de paquete de tu broker: un FlowFile fusionado de 5 MB enviado como un solo mensaje MQTT es un accidente habitual. La calculadora de tamaño de paquetes ayuda a estimar el overhead.

Las publicaciones fallidas van a failure. Devuelve esa relación a PublishMQTT con reintento o penalización para que un corte breve del broker no pierda datos.

Construye un flujo probado

El objetivo: consumir telemetría de testmqtt/nifi/<device>/telemetry, guardarlo todo y publicar una alerta cuando la temperatura supere 30.

ConsumeMQTT ──Message──▶ QueryRecord ──hot──▶ PublishMQTT (testmqtt/nifi/alerts)
     │                        └──original──▶ PutDatabaseRecord / PublishKafka / PutFile
     └──parse.failure──▶ LogAttribute

1. ConsumeMQTT con las propiedades anteriores, Record Reader JsonTreeReader, Record Writer JsonRecordSetWriter, Add attributes as fields true.
2. QueryRecord con el mismo reader/writer y una propiedad dinámica llamada hot:

SELECT _topic, device, temperature, ts
FROM FLOWFILE
WHERE temperature > 30

1. PublishMQTT en la relación hot, Topic testmqtt/nifi/alerts, QoS 1, con Record Reader/Writer para que cada lectura caliente se publique como un mensaje propio.
2. Envía la relación original a tu almacenamiento: PostgreSQL con PutDatabaseRecord, Kafka con PublishKafka o simplemente PutFile mientras pruebas.

Por defecto, QueryRecord también envía un FlowFile vacío a hot cuando nada coincide. Pon Include Zero Record FlowFiles en false para que solo los lotes con al menos una lectura caliente lleguen a PublishMQTT.

Pruébalo de extremo a extremo

Arranca todos los procesadores y abre un suscriptor en el topic de alertas:

mosquitto_sub -h broker.connectmq.com -p 1883 -t 'testmqtt/nifi/alerts' -v

Publica una lectura normal y una caliente:

mosquitto_pub -h broker.connectmq.com -p 1883 -q 1 \
  -t testmqtt/nifi/press-7/telemetry \
  -m '{"device":"press-7","temperature":24.1,"ts":1760700000}'

mosquitto_pub -h broker.connectmq.com -p 1883 -q 1 \
  -t testmqtt/nifi/press-7/telemetry \
  -m '{"device":"press-7","temperature":33.8,"ts":1760700005}'

La terminal 1 debería mostrar exactamente una alerta con "temperature":33.8 y "_topic":"testmqtt/nifi/press-7/telemetry". En NiFi, haz clic derecho en la cola después de ConsumeMQTT y elige List queue (o abre Data Provenance) para ver ambos FlowFiles y sus atributos mqtt.*. Luego publica not json en el mismo topic y confirma que llega a parse.failure en lugar de romper el flujo. Puedes hacer lo mismo desde el cliente MQTT online si prefieres el navegador.

Observa en directo los topics de prueba de NiFi

ConsumeMQTT en un clúster

En un clúster NiFi cada nodo ejecuta cada procesador. Para ConsumeMQTT eso significa que cada nodo se suscribe, y cada nodo recibe cada mensaje. Elige uno de estos dos patrones:

• Solo nodo primario. En la pestaña Scheduling del procesador pon Execution en Primary node. Sencillo y ordenado, pero un solo nodo hace todo el trabajo MQTT y hay un breve hueco cuando cambia el primario.
• Shared subscription. Define Group ID (por ejemplo nifi-ingest) y deja Client ID vacío para que cada nodo reciba un ID único. El broker reparte los mensajes entre los nodos; consulta shared subscriptions en MQTT. El orden solo se mantiene por nodo, no de forma global.

Nunca pongas el mismo Client ID fijo en todos los nodos: el broker permite una conexión por client ID, así que los nodos se quitan la sesión unos a otros en bucle.

Sesiones, QoS y reinicios

Con Clean Session y QoS 0, todo lo publicado mientras NiFi está parado se pierde. Para una ingesta que no puedes permitirte perder, usa QoS 1, un Client ID fijo por consumidor, Resume Session y —en MQTT 5— un Session Expiry Interval lo bastante largo para cubrir las ventanas de mantenimiento. El broker encola entonces los mensajes durante la caída y los entrega cuando NiFi se reconecta. Esto requiere un broker que conserve sesiones persistentes; el endpoint de pruebas compartido broker.connectmq.com no almacena sesiones offline, así que verifica este comportamiento contra tu propia instancia privada de broker (consulta sesiones persistentes).

Errores comunes

Consulta los reason codes numéricos con la herramienta de reason codes MQTT.

Los dos procesadores MQTT de NiFi son sencillos, pero los flujos de producción dependen de los detalles: TLS y credenciales parametrizadas, records en lugar de un FlowFile por mensaje, una estrategia de clúster clara y una configuración de sesión acorde con cuántos datos puedes permitirte perder. Prueba cada flujo con un segundo cliente MQTT antes de confiarle dispositivos reales.

Preguntas frecuentes