MQTT mit Apache NiFi: ConsumeMQTT und PublishMQTT konfigurieren

Apache NiFi verbindet sich über zwei Standard-Processors mit MQTT: ConsumeMQTT abonniert einen Topic-Filter und macht aus eingehenden Nachrichten FlowFiles, PublishMQTT schickt FlowFile-Inhalte auf ein Topic. Minimal brauchst du eine Broker URI wie tcp://broker.connectmq.com:1883, einen Topic Filter bzw. ein Topic, eine QoS und – bei ConsumeMQTT – eine Max Queue Size. Alles andere (TLS, Zugangsdaten, Sessions, Records, Clustering) entscheidet, ob der Flow in Produktion zuverlässig läuft.

Dieses Tutorial geht jede wichtige Property durch und baut dann einen getesteten Flow, der JSON-Telemetrie konsumiert, mit SQL filtert und Alarme zurück nach MQTT published.

NiFi zum Testen starten

NiFi 2.x läuft auf Java 21; das offizielle Docker-Image ist der schnellste Weg zur Oberfläche:

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

Öffne https://localhost:8443/nifi (selbstsigniertes Zertifikat akzeptieren) und melde dich an. Für alles, was bleiben soll, pinne ein konkretes Image-Tag. Das Passwort muss mindestens 12 Zeichen haben, sonst startet der Container nicht.

ConsumeMQTT-Properties

Zieh einen Processor auf das Canvas, such nach ConsumeMQTT und öffne die Properties. Diese setzt du in jedem Flow:

Der Processor schreibt jede Nachricht in die Relationship Message und setzt die Attribute mqtt.broker, mqtt.topic, mqtt.qos, mqtt.isDuplicate und mqtt.isRetained. Ohne Record Reader wird jede MQTT-Nachricht zu einem FlowFile, dessen Inhalt der rohe Payload ist.

Batching mit Records

Ein FlowFile pro Nachricht ist bei wenigen Nachrichten pro Sekunde in Ordnung, darüber Verschwendung. Setz Record Reader auf einen JsonTreeReader und Record Writer auf einen JsonRecordSetWriter, dann schreibt ConsumeMQTT alle seit dem letzten Lauf empfangenen Nachrichten als Records in ein einziges FlowFile. Mit Add attributes as fields auf true bekommt jeder Record zusätzlich die Felder _topic, _qos, _isDuplicate und _isRetained – so geht das Topic nicht verloren, wenn sich Nachrichten vieler Geräte ein FlowFile teilen. Payloads, die der Reader nicht parsen kann, landen in der Relationship parse.failure; leite sie in einen Funnel oder ein Log, damit fehlerhafte Geräte sichtbar werden. Typische Ursachen findest du unter JSON-Parse-Fehler.

TLS

Ändere die URI auf ssl://broker.connectmq.com:8883 (oder wss://broker.connectmq.com:8084/mqtt hinter Proxys, die nur HTTPS durchlassen) und setz SSL Context Service auf einen StandardSSLContextService. Bei einem Broker mit öffentlich vertrauenswürdigem Zertifikat kann der Truststore auf die cacerts-Datei der JVM zeigen; bei einer privaten CA importierst du die CA in einen eigenen Truststore. Schlägt der Handshake fehl, vergleiche mit einem openssl s_client-Test – MQTT-TLS-Verbindung testen zeigt die Befehle.

PublishMQTT-Properties

PublishMQTT teilt die Verbindungs-Properties (Broker URI, Version, Client ID, Zugangsdaten, SSL) und ergänzt:

Ohne Record Reader ist der gesamte FlowFile-Inhalt ein Payload. Prüf die Größe gegen die maximale Paketgröße deines Brokers – ein gemergtes 5-MB-FlowFile als eine MQTT-Nachricht ist ein häufiger Unfall. Der Paketgrößen-Rechner hilft beim Abschätzen des Overheads.

Fehlgeschlagene Publishes gehen an failure. Leite diese Relationship mit Retry oder Penalty zurück zu PublishMQTT, damit ein kurzer Broker-Aussetzer keine Daten verliert.

Einen getesteten Flow bauen

Ziel: Telemetrie von testmqtt/nifi/<device>/telemetry konsumieren, alles speichern und einen Alarm publishen, wenn die Temperatur 30 übersteigt.

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

1. ConsumeMQTT mit den Properties von oben, Record Reader JsonTreeReader, Record Writer JsonRecordSetWriter, Add attributes as fields true.
2. QueryRecord mit demselben Reader/Writer und einer dynamischen Property namens hot:

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

1. PublishMQTT an der Relationship hot, Topic testmqtt/nifi/alerts, QoS 1, mit Record Reader/Writer, damit jeder heiße Messwert als eigene Nachricht published wird.
2. Leite die Relationship original in deinen Speicher: PostgreSQL mit PutDatabaseRecord, Kafka mit PublishKafka oder beim Testen einfach PutFile.

Standardmäßig schickt QueryRecord auch dann ein leeres FlowFile an hot, wenn nichts passt. Setz Include Zero Record FlowFiles auf false, damit nur Batches mit mindestens einem heißen Messwert bei PublishMQTT ankommen.

End-to-End testen

Starte alle Processors und öffne einen Subscriber auf dem Alarm-Topic:

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

Publish einen normalen und einen heißen Messwert:

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

Terminal 1 sollte genau einen Alarm mit "temperature":33.8 und "_topic":"testmqtt/nifi/press-7/telemetry" ausgeben. Rechtsklick in NiFi auf die Queue nach ConsumeMQTT und List queue wählen (oder Data Provenance öffnen), um beide FlowFiles samt mqtt.*-Attributen zu sehen. Publish dann not json auf dasselbe Topic und prüf, dass es in parse.failure landet, statt den Flow zu stören. Im Browser geht das genauso mit dem Online-MQTT-Client.

NiFi-Test-Topics live beobachten

ConsumeMQTT im Cluster

In einem NiFi-Cluster führt jeder Knoten jeden Processor aus. Für ConsumeMQTT heißt das: Jeder Knoten abonniert – und jeder Knoten bekommt jede Nachricht. Wähle eines von zwei Mustern:

• Nur Primary Node. Setz im Scheduling-Tab des Processors Execution auf Primary node. Einfach und geordnet, aber ein Knoten erledigt die ganze MQTT-Arbeit, und beim Wechsel des Primary entsteht eine kurze Lücke.
• Shared Subscription. Setz eine Group ID (zum Beispiel nifi-ingest) und lass die Client ID leer, damit jeder Knoten eine eindeutige ID bekommt. Der Broker verteilt die Nachrichten auf die Knoten; siehe MQTT Shared Subscriptions. Die Reihenfolge gilt dann nur pro Knoten, nicht global.

Setz niemals dieselbe feste Client ID auf allen Knoten: Der Broker erlaubt eine Verbindung pro Client ID, die Knoten nehmen sich die Session also in einer Schleife gegenseitig weg.

Sessions, QoS und Neustarts

Mit Clean Session und QoS 0 ist alles weg, was published wird, während NiFi steht. Für Ingestion, bei der nichts verloren gehen darf, nutze QoS 1, eine feste Client ID pro Consumer, Resume Session und – bei MQTT 5 – ein Session Expiry Interval, das Wartungsfenster abdeckt. Der Broker puffert die Nachrichten dann während des Ausfalls und stellt sie beim Reconnect zu. Das setzt einen Broker voraus, der persistente Sessions hält; der geteilte Test-Endpoint broker.connectmq.com speichert keine Offline-Sessions, prüf dieses Verhalten also gegen deine eigene private Broker-Instanz (siehe persistente Sessions).

Häufige Fehler

Numerische Reason Codes schlägst du im MQTT-Reason-Code-Tool nach.

NiFis zwei MQTT-Processors sind einfach, aber produktive Flows hängen an den Details: TLS und parametrisierte Zugangsdaten, Records statt einem FlowFile pro Nachricht, eine klare Cluster-Strategie und ein Session-Setup, das dazu passt, wie viele Daten du verlieren darfst. Teste jeden Flow mit einem zweiten MQTT-Client, bevor du ihm echte Geräte anvertraust.

Häufig gestellte Fragen