MQTT with Apache NiFi: ConsumeMQTT and PublishMQTT Configuration

Apache NiFi connects to MQTT with two standard processors: ConsumeMQTT subscribes to a topic filter and turns incoming messages into FlowFiles, and PublishMQTT sends FlowFile content to a topic. The minimum configuration is a Broker URI such as tcp://broker.connectmq.com:1883, a Topic Filter or Topic, a QoS and — on ConsumeMQTT — a Max Queue Size. Everything else (TLS, credentials, sessions, records, clustering) decides whether the flow is reliable in production.

This tutorial walks through each property that matters, then builds and tests a flow that consumes JSON telemetry, filters it with SQL and publishes alerts back to MQTT.

Start NiFi for testing

NiFi 2.x runs on Java 21; the official Docker image is the quickest way to get a UI:

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

Open https://localhost:8443/nifi (accept the self-signed certificate) and log in. Pin a specific image tag for anything you keep. The password must be at least 12 characters, or the container refuses to start.

ConsumeMQTT properties

Drag a processor onto the canvas, search for ConsumeMQTT and open its properties. These are the ones you set on every flow:

The processor writes each message to the Message relationship and adds these attributes: mqtt.broker, mqtt.topic, mqtt.qos, mqtt.isDuplicate and mqtt.isRetained. Without a record reader, each MQTT message becomes one FlowFile, and the FlowFile content is the raw payload.

Batching with records

One FlowFile per message is fine for a few messages per second and wasteful beyond that. Set Record Reader to a JsonTreeReader and Record Writer to a JsonRecordSetWriter, and ConsumeMQTT writes all messages received since the last run as records in a single FlowFile. With Add attributes as fields set to true, every record also gets _topic, _qos, _isDuplicate and _isRetained fields, so you do not lose the topic when messages from many devices share one FlowFile. Payloads that the reader cannot parse go to the parse.failure relationship — route it to a funnel or a log so bad devices are visible; see JSON parse errors for typical causes.

TLS

Change the URI to ssl://broker.connectmq.com:8883 (or wss://broker.connectmq.com:8084/mqtt through proxies that only allow HTTPS) and set SSL Context Service to a StandardSSLContextService. For a broker with a publicly trusted certificate, the truststore can point at the JVM's cacerts file; for a private CA, import the CA into a dedicated truststore. If the handshake fails, compare with an openssl s_client test — testing an MQTT TLS connection shows the commands.

PublishMQTT properties

PublishMQTT shares the connection properties (Broker URI, version, Client ID, credentials, SSL) and adds:

Without a record reader, the whole FlowFile content is one payload. Check the size against your broker's maximum packet size — a 5 MB merged FlowFile sent as one MQTT message is a common accident. The packet size calculator helps estimate overhead.

Failed publishes go to failure. Route that relationship back to PublishMQTT with a retry or penalty so a broker blip does not drop data.

Build a tested flow

The goal: consume telemetry from testmqtt/nifi/<device>/telemetry, keep everything for storage, and publish an alert when the temperature exceeds 30.

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

1. ConsumeMQTT with the properties above, Record Reader JsonTreeReader, Record Writer JsonRecordSetWriter, Add attributes as fields true.
2. QueryRecord with the same reader/writer and a dynamic property named hot:

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

1. PublishMQTT on the hot relationship, Topic testmqtt/nifi/alerts, QoS 1, with a Record Reader/Writer so every hot reading is published as its own message.
2. Send the original relationship to your storage: PostgreSQL with PutDatabaseRecord, Kafka with PublishKafka, or simply PutFile while you test.

By default QueryRecord also sends an empty FlowFile to hot when nothing matches. Set Include Zero Record FlowFiles to false so only batches with at least one hot reading reach PublishMQTT.

Test it end to end

Start all processors, then open a subscriber on the alert topic:

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

Publish one normal and one hot reading:

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 should print exactly one alert containing "temperature":33.8 and "_topic":"testmqtt/nifi/press-7/telemetry". In NiFi, right-click the queue after ConsumeMQTT and choose List queue (or open Data Provenance) to see both FlowFiles and their mqtt.* attributes. Then publish not json to the same topic and confirm it lands on parse.failure instead of crashing the flow. You can do the same from the online MQTT client if you prefer a browser.

Watch the NiFi test topics live

Running ConsumeMQTT in a cluster

In a NiFi cluster every node runs every processor. For ConsumeMQTT that means every node subscribes — and every node receives every message. Pick one of two patterns:

• Primary node only. In the processor's Scheduling tab set Execution to Primary node. Simple and ordered, but one node does all MQTT work and there is a short gap when the primary changes.
• Shared subscription. Set Group ID (for example nifi-ingest) and leave Client ID empty so each node gets a unique ID. The broker distributes messages across nodes; see MQTT shared subscriptions. Ordering is then only per node, not global.

Never set the same fixed Client ID on all nodes: the broker allows one connection per client ID, so the nodes take the session away from each other in a loop.

Sessions, QoS and restarts

With Clean Session and QoS 0, anything published while NiFi is stopped is gone. For ingestion you cannot afford to lose, use QoS 1, a fixed Client ID per consumer, Resume Session and — on MQTT 5 — a Session Expiry Interval long enough to cover maintenance windows. The broker then queues messages during the outage and delivers them when NiFi reconnects. This requires a broker that keeps persistent sessions; the shared test endpoint broker.connectmq.com does not store offline sessions, so verify this behaviour against your own private broker instance (see persistent sessions).

Common errors

Look up numeric reason codes with the MQTT reason codes tool.

NiFi's two MQTT processors are simple, but production flows depend on the details: TLS and parameterised credentials, records instead of one FlowFile per message, a clear clustering strategy and a session setup that matches how much data you can afford to lose. Test every flow with a second MQTT client before trusting it with real devices.

Frequently asked questions