Ponte de MQTT para Apache Kafka: conectores, ponte própria e ordenação

Para levar MQTT para o Apache Kafka, você executa um componente que assina topics MQTT no broker e produz cada mensagem como um registro do Kafka. Há três opções práticas: um conector MQTT source do Kafka Connect, uma pequena ponte escrita por você ou uma integração com Kafka embutida no seu broker MQTT. Seja qual for a escolha, as partes difíceis são as mesmas: mapear topics, escolher chaves de registro para que a ordem sobreviva e não perder mensagens quando algo reinicia.

Este tutorial passa pelas três opções, com uma ponte em Python funcional e um teste que você roda em cinco minutos.

Por que MQTT e Kafka juntos

MQTT e Kafka resolvem problemas diferentes. O MQTT conecta milhares de dispositivos pequenos, com conexão intermitente, a um broker com pacotes minúsculos e QoS por mensagem. O Kafka é um log durável e particionado que serviços de backend reprocessam, combinam e processam em escala. Dispositivos não deveriam rodar clientes Kafka, e o Kafka nunca foi pensado para manter abertas um milhão de sessões TCP instáveis. Uma ponte deixa cada lado fazer o que faz melhor. Para uma comparação mais profunda, veja MQTT vs Kafka.

Opção 1: Kafka Connect MQTT source

Se você já roda Kafka Connect, um conector source é o caminho com menos código. Duas escolhas comuns:

• Confluent MQTT Source Connector (io.confluent.connect.mqtt.MqttSourceConnector): distribuído sob a licença da Confluent; confira os termos para o seu deploy.
• Stream Reactor MQTT Source, da Lenses: open source, configurado com uma instrução KCQL que mapeia topics MQTT para topics Kafka.

Uma configuração do conector da Confluent fica assim:

{
  "name": "mqtt-telemetry-source",
  "config": {
    "connector.class": "io.confluent.connect.mqtt.MqttSourceConnector",
    "tasks.max": "1",
    "mqtt.server.uri": "ssl://broker.connectmq.com:8883",
    "mqtt.topics": "testmqtt/kafka/+/telemetry",
    "mqtt.qos": "1",
    "mqtt.username": "username",
    "mqtt.password": "password",
    "kafka.topic": "iot.telemetry",
    "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "confluent.topic.bootstrap.servers": "kafka:9092",
    "confluent.topic.replication.factor": "1"
  }
}

curl -X POST -H "Content-Type: application/json" \
  --data @mqtt-source.json http://localhost:8083/connectors

Mantenha ByteArrayConverter para o valor, a menos que você queira mesmo que o Connect interprete os payloads: assim JSON, Protobuf ou CBOR passam intactos. Verifique na documentação do conector como ele preenche a chave do registro; se usar o topic MQTT completo, cada dispositivo continua indo para uma partição estável, que é o que você quer para a ordenação.

O trade-off: um conector é fácil de implantar e monitorar ao lado de outros conectores, mas lógica personalizada (validação, enriquecimento, extrair o ID do dispositivo para a chave, rotear para vários topics Kafka) fica limitada ao que o conector e as Single Message Transforms suportam.

Opção 2: uma ponte própria em Python

Uma ponte própria tem cerca de 50 linhas e dá controle total. Esta usa paho-mqtt 2.x e confluent-kafka, e confirma cada mensagem MQTT somente depois que o Kafka confirmou a escrita:

import os
import paho.mqtt.client as mqtt
from confluent_kafka import Producer

MQTT_HOST = "broker.connectmq.com"  # use seu broker privado para sessões persistentes
MQTT_TOPIC = "testmqtt/kafka/+/telemetry"
KAFKA_TOPIC = "iot.telemetry"

producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "enable.idempotence": True,   # sem duplicatas nem reordenação nas novas tentativas
    "acks": "all",
    "linger.ms": 20,              # lotes pequenos, melhor throughput
})

def device_id(topic: str) -> str:
    # testmqtt/kafka/<device>/telemetry -> <device>
    return topic.split("/")[2]

def on_connect(client, userdata, flags, reason_code, properties):
    print("MQTT conectado:", reason_code)
    client.subscribe(MQTT_TOPIC, qos=1)

def on_message(client, userdata, msg):
    mid, qos = msg.mid, msg.qos

    def delivered(err, record):
        if err is None:
            client.ack(mid, qos)          # PUBACK só quando o Kafka tem o registro
        else:
            print("Falha ao gravar no Kafka, a mensagem será reenviada:", err)

    producer.produce(
        KAFKA_TOPIC,
        key=device_id(msg.topic),
        value=msg.payload,
        headers=[("mqtt_topic", msg.topic.encode())],
        on_delivery=delivered,
    )

client = mqtt.Client(
    mqtt.CallbackAPIVersion.VERSION2,
    client_id="kafka-bridge-1",
    clean_session=False,      # sessão persistente: o broker enfileira enquanto estamos fora
    manual_ack=True,
)
client.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
client.tls_set()
client.on_connect = on_connect
client.on_message = on_message
client.connect(MQTT_HOST, 8883)
client.loop_start()

try:
    while True:
        producer.poll(0.1)        # executa os callbacks de entrega
finally:
    producer.flush(10)
    client.loop_stop()

Três detalhes tornam esta ponte segura:

1. clean_session=False com um client ID fixo: o broker mantém a assinatura e enfileira mensagens QoS 1 enquanto a ponte reinicia. Leia sessões persistentes no MQTT para saber por quanto tempo os brokers as guardam.
2. manual_ack=True: o paho não envia PUBACK automaticamente. Se o processo morrer antes de o Kafka confirmar, o broker reenvia a mensagem.
3. O ID do dispositivo como chave do registro: todas as mensagens do dispositivo 42 vão para a mesma partição, em ordem.

O resultado é entrega at-least-once. Uma queda entre a escrita no Kafka e o PUBACK gera uma duplicata, então os consumidores seguintes devem ser idempotentes (por exemplo, deduplicar por um ID de mensagem no payload).

Ordenação e chaves

A ordenação é onde a maioria dos pipelines de MQTT para Kafka erra. As regras:

• O MQTT ordena mensagens por topic dentro de uma sessão de cliente com o mesmo QoS. Não promete nada entre topics diferentes nem entre publishers diferentes.
• O Kafka ordena registros por partição. Com o particionador padrão, a mesma chave sempre cai na mesma partição.
• Então use uma chave que corresponda à unidade que precisa ficar ordenada: quase sempre o ID do dispositivo, não o topic completo e nunca um valor aleatório. Se um dispositivo publica telemetry e status em dois topics e você precisa deles ordenados entre si, use o ID do dispositivo como chave nos dois.

Se você escalar a ponte horizontalmente com uma assinatura compartilhada ($share/kafka-bridge/testmqtt/kafka/+/telemetry), o broker distribui as mensagens entre as instâncias da ponte. Mensagens de um mesmo dispositivo podem então estar em trânsito em duas instâncias ao mesmo tempo e chegar ao Kafka fora de ordem. Opções: aceitar e reordenar depois por um timestamp ou número de sequência no payload, ou particionar o lado MQTT (uma ponte por faixa de topics, como +/telemetry dividido por prefixo de região). Você pode verificar o comportamento com a abordagem de testar a ordem das mensagens MQTT.

Opção 3: integração com Kafka nativa do broker

Alguns brokers MQTT trazem uma integração com Kafka, muitas vezes como extensão enterprise ou ação do motor de regras, que produz mensagens direto do processo do broker. Isso elimina um salto e um serviço para operar. Os trade-offs: amarra seu pipeline àquele broker, a configuração fica no broker e não no seu código, e a contrapressão do Kafka afeta o próprio broker. Compare as opções na página de comparação de brokers antes de se comprometer.

Teste a ponte de ponta a ponta

Suba um Kafka de nó único (modo KRaft) e um consumidor de console que imprima as chaves:

docker run -d --name kafka -p 9092:9092 apache/kafka:3.8.0

docker exec -it kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic iot.telemetry \
  --from-beginning --property print.key=true --property print.headers=true

Rode a ponte em um segundo terminal e publique algumas mensagens de dois dispositivos:

for i in 1 2 3 4 5; do
  mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
    -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
    -t testmqtt/kafka/dev-a/telemetry -m "{\"seq\":$i,\"temp\":21.$i}"
  mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
    -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
    -t testmqtt/kafka/dev-b/telemetry -m "{\"seq\":$i,\"temp\":19.$i}"
done

O consumidor deve mostrar as chaves dev-a e dev-b com seq crescente por chave. Você também pode publicar pelo cliente MQTT online e ver as mesmas mensagens chegarem ao Kafka. Depois rode o teste de falha que realmente importa. Ele depende de uma sessão persistente, então aponte a ponte para a sua própria instância privada de broker (por exemplo, um broker dedicado da ConnectMQ): o endpoint de teste compartilhado não guarda sessões offline e tem limite de taxa, o que serve para smoke tests, mas não para testar filas.

1. Pare a ponte com Ctrl+C.
2. Publique mais cinco mensagens.
3. Inicie a ponte de novo: as mensagens enfileiradas precisam aparecer no Kafka.
4. Pare o Kafka (docker stop kafka), publique e inicie o Kafka de novo: a ponte não pode enviar PUBACK enquanto o Kafka está fora, então as mensagens chegam depois da recuperação.

Se o passo 3 perder mensagens, sua sessão não é persistente (client ID errado, clean_session=True ou o broker expirou a sessão). Se o passo 4 perder, você está confirmando antes do Kafka.

Acompanhe os topics de entrada da ponte no navegador

Uma ponte de MQTT para Kafka é pequena, mas os detalhes decidem se ela é confiável: sessões persistentes, confirmações atreladas às do Kafka e chaves de registro que correspondem à ordem de que você precisa. Comece com um conector se ele cobrir seu mapeamento, passe para uma ponte própria quando precisar de lógica e mantenha o teste de reinício no seu CI para que ele continue valendo.

Perguntas frequentes