MQTT com Apache NiFi: configurando ConsumeMQTT e PublishMQTT

O Apache NiFi se conecta ao MQTT com dois processadores padrão: o ConsumeMQTT assina um filtro de topic e transforma as mensagens recebidas em FlowFiles, e o PublishMQTT envia o conteúdo de um FlowFile para um topic. A configuração mínima é um Broker URI como tcp://broker.connectmq.com:1883, um Topic Filter ou Topic, uma QoS e — no ConsumeMQTT — um Max Queue Size. Todo o resto (TLS, credenciais, sessões, records, cluster) decide se o fluxo é confiável em produção.

Este tutorial passa por cada propriedade importante e depois monta e testa um fluxo que consome telemetria JSON, filtra com SQL e publica alertas de volta no MQTT.

Suba o NiFi para testes

O NiFi 2.x roda em Java 21; a imagem Docker oficial é o jeito mais rápido de ter a 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

Abra https://localhost:8443/nifi (aceite o certificado autoassinado) e faça login. Fixe uma tag de imagem específica para tudo o que for manter. A senha precisa ter pelo menos 12 caracteres, senão o container não sobe.

Propriedades do ConsumeMQTT

Arraste um processador para o canvas, procure ConsumeMQTT e abra as propriedades. Estas são as que você define em todo fluxo:

O processador grava cada mensagem no relacionamento Message e adiciona os atributos mqtt.broker, mqtt.topic, mqtt.qos, mqtt.isDuplicate e mqtt.isRetained. Sem record reader, cada mensagem MQTT vira um FlowFile e o conteúdo dele é o payload bruto.

Agrupando com records

Um FlowFile por mensagem funciona para poucas mensagens por segundo e é desperdício acima disso. Defina Record Reader como JsonTreeReader e Record Writer como JsonRecordSetWriter, e o ConsumeMQTT grava todas as mensagens recebidas desde a última execução como registros em um único FlowFile. Com Add attributes as fields em true, cada registro ganha também os campos _topic, _qos, _isDuplicate e _isRetained, então você não perde o topic quando mensagens de vários dispositivos dividem o mesmo FlowFile. Payloads que o reader não consegue interpretar vão para o relacionamento parse.failure — mande-o para um funnel ou um log para que dispositivos com problema fiquem visíveis; veja erros de parse de JSON para as causas típicas.

TLS

Troque o URI para ssl://broker.connectmq.com:8883 (ou wss://broker.connectmq.com:8084/mqtt atrás de proxies que só liberam HTTPS) e defina SSL Context Service com um StandardSSLContextService. Para um broker com certificado de confiança pública, o truststore pode apontar para o arquivo cacerts da JVM; para uma CA privada, importe a CA em um truststore dedicado. Se o handshake falhar, compare com um teste de openssl s_client — testar uma conexão MQTT com TLS mostra os comandos.

Propriedades do PublishMQTT

O PublishMQTT compartilha as propriedades de conexão (Broker URI, versão, Client ID, credenciais, SSL) e acrescenta:

Sem record reader, todo o conteúdo do FlowFile é um único payload. Confira o tamanho em relação ao tamanho máximo de pacote do broker — um FlowFile mesclado de 5 MB enviado como uma única mensagem MQTT é um acidente comum. A calculadora de tamanho de pacote ajuda a estimar o overhead.

Publicações com falha vão para failure. Ligue esse relacionamento de volta ao PublishMQTT com retry ou penalidade para que uma oscilação do broker não descarte dados.

Monte um fluxo testado

O objetivo: consumir telemetria de testmqtt/nifi/<device>/telemetry, guardar tudo e publicar um alerta quando a temperatura passar de 30.

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

1. ConsumeMQTT com as propriedades acima, Record Reader JsonTreeReader, Record Writer JsonRecordSetWriter, Add attributes as fields true.
2. QueryRecord com o mesmo reader/writer e uma propriedade dinâmica chamada hot:

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

1. PublishMQTT no relacionamento hot, Topic testmqtt/nifi/alerts, QoS 1, com Record Reader/Writer para que cada leitura quente seja publicada como mensagem própria.
2. Mande o relacionamento original para o armazenamento: PostgreSQL com PutDatabaseRecord, Kafka com PublishKafka, ou simplesmente PutFile enquanto testa.

Por padrão, o QueryRecord também envia um FlowFile vazio para hot quando nada combina. Defina Include Zero Record FlowFiles como false para que só lotes com pelo menos uma leitura quente cheguem ao PublishMQTT.

Teste de ponta a ponta

Inicie todos os processadores e abra um subscriber no topic de alertas:

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

Publique uma leitura normal e uma quente:

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

O terminal 1 deve mostrar exatamente um alerta com "temperature":33.8 e "_topic":"testmqtt/nifi/press-7/telemetry". No NiFi, clique com o botão direito na fila depois do ConsumeMQTT e escolha List queue (ou abra Data Provenance) para ver os dois FlowFiles e seus atributos mqtt.*. Depois publique not json no mesmo topic e confirme que ele cai em parse.failure em vez de quebrar o fluxo. Você pode fazer o mesmo pelo cliente MQTT online se preferir o navegador.

Acompanhe ao vivo os topics de teste do NiFi

ConsumeMQTT em cluster

Em um cluster NiFi, todo nó executa todos os processadores. Para o ConsumeMQTT, isso significa que todo nó assina — e todo nó recebe todas as mensagens. Escolha um destes dois padrões:

• Somente nó primário. Na aba Scheduling do processador, defina Execution como Primary node. Simples e ordenado, mas um único nó faz todo o trabalho MQTT e há uma pequena lacuna quando o primário muda.
• Shared subscription. Defina Group ID (por exemplo nifi-ingest) e deixe o Client ID vazio para que cada nó tenha um ID único. O broker distribui as mensagens entre os nós; veja shared subscriptions no MQTT. A ordem passa a valer só por nó, não globalmente.

Nunca use o mesmo Client ID fixo em todos os nós: o broker permite uma conexão por client ID, então os nós ficam tomando a sessão uns dos outros em loop.

Sessões, QoS e reinícios

Com Clean Session e QoS 0, tudo o que for publicado enquanto o NiFi estiver parado se perde. Para uma ingestão que não pode perder dados, use QoS 1, um Client ID fixo por consumidor, Resume Session e — no MQTT 5 — um Session Expiry Interval longo o bastante para cobrir janelas de manutenção. Assim o broker enfileira as mensagens durante a parada e as entrega quando o NiFi reconecta. Isso exige um broker que mantenha sessões persistentes; o endpoint de testes compartilhado broker.connectmq.com não armazena sessões offline, então verifique esse comportamento na sua própria instância privada de broker (veja sessões persistentes).

Erros comuns

Consulte reason codes numéricos na ferramenta de reason codes MQTT.

Os dois processadores MQTT do NiFi são simples, mas fluxos de produção dependem dos detalhes: TLS e credenciais parametrizadas, records em vez de um FlowFile por mensagem, uma estratégia de cluster clara e uma configuração de sessão compatível com quanto dado você pode se dar ao luxo de perder. Teste cada fluxo com um segundo cliente MQTT antes de confiar dispositivos reais a ele.

Perguntas frequentes