MQTT avec Apache NiFi : configurer ConsumeMQTT et PublishMQTT

Apache NiFi se connecte à MQTT avec deux processeurs standard : ConsumeMQTT s'abonne à un filtre de topic et transforme les messages reçus en FlowFiles, et PublishMQTT envoie le contenu d'un FlowFile sur un topic. La configuration minimale tient en un Broker URI comme tcp://broker.connectmq.com:1883, un Topic Filter ou Topic, une QoS et — pour ConsumeMQTT — un Max Queue Size. Tout le reste (TLS, identifiants, sessions, records, cluster) détermine si le flux est fiable en production.

Ce tutoriel passe en revue chaque propriété importante, puis construit et teste un flux qui consomme de la télémétrie JSON, la filtre en SQL et republie des alertes sur MQTT.

Lancer NiFi pour les tests

NiFi 2.x tourne sur Java 21 ; l'image Docker officielle est le moyen le plus rapide d'obtenir l'interface :

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

Ouvrez https://localhost:8443/nifi (acceptez le certificat auto-signé) et connectez-vous. Figez un tag d'image précis pour tout ce que vous gardez. Le mot de passe doit comporter au moins 12 caractères, sinon le conteneur refuse de démarrer.

Propriétés de ConsumeMQTT

Glissez un processeur sur le canvas, cherchez ConsumeMQTT et ouvrez ses propriétés. Voici celles à régler dans chaque flux :

Le processeur écrit chaque message sur la relation Message et ajoute les attributs mqtt.broker, mqtt.topic, mqtt.qos, mqtt.isDuplicate et mqtt.isRetained. Sans record reader, chaque message MQTT devient un FlowFile dont le contenu est le payload brut.

Regrouper avec des records

Un FlowFile par message convient pour quelques messages par seconde et devient du gaspillage au-delà. Réglez Record Reader sur un JsonTreeReader et Record Writer sur un JsonRecordSetWriter : ConsumeMQTT écrit alors tous les messages reçus depuis la dernière exécution sous forme de records dans un seul FlowFile. Avec Add attributes as fields à true, chaque record reçoit aussi les champs _topic, _qos, _isDuplicate et _isRetained, ce qui évite de perdre le topic quand des messages de nombreux appareils partagent un FlowFile. Les payloads que le reader ne sait pas parser partent sur la relation parse.failure — dirigez-la vers un funnel ou un log pour rendre visibles les appareils défaillants ; voir erreurs de parsing JSON pour les causes typiques.

TLS

Passez l'URI à ssl://broker.connectmq.com:8883 (ou wss://broker.connectmq.com:8084/mqtt derrière des proxys qui n'autorisent que HTTPS) et réglez SSL Context Service sur un StandardSSLContextService. Pour un broker doté d'un certificat publiquement reconnu, le truststore peut pointer vers le fichier cacerts de la JVM ; pour une CA privée, importez la CA dans un truststore dédié. Si le handshake échoue, comparez avec un test openssl s_client — tester une connexion MQTT TLS donne les commandes.

Propriétés de PublishMQTT

PublishMQTT partage les propriétés de connexion (Broker URI, version, Client ID, identifiants, SSL) et ajoute :

Sans record reader, tout le contenu du FlowFile constitue un seul payload. Vérifiez sa taille par rapport à la taille maximale de paquet de votre broker — un FlowFile fusionné de 5 Mo envoyé comme un seul message MQTT est un accident fréquent. Le calculateur de taille de paquet aide à estimer l'overhead.

Les publications en échec partent sur failure. Rebouclez cette relation vers PublishMQTT avec retry ou pénalité pour qu'un incident bref du broker ne perde pas de données.

Construire un flux testé

Objectif : consommer la télémétrie de testmqtt/nifi/<device>/telemetry, tout conserver et publier une alerte quand la température dépasse 30.

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

1. ConsumeMQTT avec les propriétés ci-dessus, Record Reader JsonTreeReader, Record Writer JsonRecordSetWriter, Add attributes as fields true.
2. QueryRecord avec le même reader/writer et une propriété dynamique nommée hot :

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

1. PublishMQTT sur la relation hot, Topic testmqtt/nifi/alerts, QoS 1, avec Record Reader/Writer pour que chaque mesure chaude soit publiée comme un message à part.
2. Envoyez la relation original vers votre stockage : PostgreSQL avec PutDatabaseRecord, Kafka avec PublishKafka, ou simplement PutFile pendant les tests.

Par défaut, QueryRecord envoie aussi un FlowFile vide sur hot quand rien ne correspond. Mettez Include Zero Record FlowFiles à false pour que seuls les lots contenant au moins une mesure chaude atteignent PublishMQTT.

Tester de bout en bout

Démarrez tous les processeurs, puis ouvrez un subscriber sur le topic d'alerte :

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

Publiez une mesure normale et une mesure chaude :

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}'

Le terminal 1 doit afficher exactement une alerte contenant "temperature":33.8 et "_topic":"testmqtt/nifi/press-7/telemetry". Dans NiFi, faites un clic droit sur la file après ConsumeMQTT et choisissez List queue (ou ouvrez Data Provenance) pour voir les deux FlowFiles et leurs attributs mqtt.*. Publiez ensuite not json sur le même topic et vérifiez qu'il arrive sur parse.failure au lieu de casser le flux. Vous pouvez faire de même depuis le client MQTT en ligne si vous préférez le navigateur.

Suivre en direct les topics de test NiFi

ConsumeMQTT en cluster

Dans un cluster NiFi, chaque nœud exécute chaque processeur. Pour ConsumeMQTT, cela veut dire que chaque nœud s'abonne — et que chaque nœud reçoit chaque message. Choisissez l'un de ces deux schémas :

• Nœud primaire uniquement. Dans l'onglet Scheduling du processeur, réglez Execution sur Primary node. Simple et ordonné, mais un seul nœud fait tout le travail MQTT et il existe une courte interruption lors d'un changement de primaire.
• Shared subscription. Définissez un Group ID (par exemple nifi-ingest) et laissez le Client ID vide pour que chaque nœud obtienne un ID unique. Le broker répartit les messages entre les nœuds ; voir les shared subscriptions MQTT. L'ordre n'est alors garanti que par nœud, pas globalement.

Ne mettez jamais le même Client ID fixe sur tous les nœuds : le broker n'autorise qu'une connexion par client ID, et les nœuds se volent la session en boucle.

Sessions, QoS et redémarrages

Avec Clean Session et QoS 0, tout ce qui est publié pendant l'arrêt de NiFi est perdu. Pour une ingestion qui ne tolère pas de perte, utilisez QoS 1, un Client ID fixe par consommateur, Resume Session et — en MQTT 5 — un Session Expiry Interval assez long pour couvrir les fenêtres de maintenance. Le broker met alors les messages en file pendant l'interruption et les livre à la reconnexion de NiFi. Cela suppose un broker qui conserve les sessions persistantes ; l'endpoint de test partagé broker.connectmq.com ne stocke pas de sessions hors ligne, vérifiez donc ce comportement sur votre propre instance privée de broker (voir sessions persistantes).

Erreurs courantes

Retrouvez les reason codes numériques dans l'outil des reason codes MQTT.

Les deux processeurs MQTT de NiFi sont simples, mais les flux de production tiennent aux détails : TLS et identifiants paramétrés, records plutôt qu'un FlowFile par message, une stratégie de cluster claire et une configuration de session adaptée à la quantité de données que vous pouvez vous permettre de perdre. Testez chaque flux avec un second client MQTT avant de lui confier de vrais appareils.

Questions fréquentes