Relier MQTT à Apache Kafka : connecteurs, bridge maison et ordre des messages

Pour relier MQTT à Apache Kafka, vous exécutez un composant qui s'abonne à des topics MQTT sur le broker et produit chaque message sous forme de record Kafka. Trois options pratiques s'offrent à vous : un connecteur MQTT source pour Kafka Connect, un petit bridge écrit par vos soins ou une intégration Kafka intégrée à votre broker MQTT. Quel que soit votre choix, les difficultés sont les mêmes : faire correspondre les topics, choisir des clés de record qui préservent l'ordre et ne pas perdre de messages lors d'un redémarrage.

Ce tutoriel couvre les trois options, avec un bridge Python fonctionnel et un test que vous pouvez lancer en cinq minutes.

Pourquoi associer MQTT et Kafka

MQTT et Kafka résolvent des problèmes différents. MQTT relie des milliers de petits appareils, connectés par intermittence, à un broker avec des paquets minuscules et une QoS par message. Kafka est un log durable et partitionné que les services backend relisent, joignent et traitent à grande échelle. Les appareils ne devraient pas exécuter de clients Kafka, et Kafka n'a jamais été prévu pour maintenir ouvertes un million de sessions TCP instables. Un bridge laisse chaque côté faire ce qu'il fait de mieux. Pour une comparaison plus approfondie, lisez MQTT vs Kafka.

Option 1 : Kafka Connect MQTT source

Si vous exploitez déjà Kafka Connect, un connecteur source demande le moins de code. Deux choix courants :

• Confluent MQTT Source Connector (io.confluent.connect.mqtt.MqttSourceConnector) : distribué sous la licence Confluent ; vérifiez les conditions pour votre déploiement.
• Stream Reactor MQTT Source de Lenses : open source, configuré par une instruction KCQL qui associe topics MQTT et topics Kafka.

Une configuration du connecteur Confluent ressemble à ceci :

{
  "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

Gardez ByteArrayConverter pour la valeur, sauf si vous voulez vraiment que Connect analyse les payloads : JSON, Protobuf ou CBOR passent ainsi sans modification. Vérifiez dans la documentation du connecteur comment il remplit la clé du record ; s'il utilise le topic MQTT complet, chaque appareil reste associé à une partition stable, ce qui est exactement ce qu'il faut pour l'ordre.

Le compromis : un connecteur se déploie et se surveille facilement à côté des autres connecteurs, mais la logique spécifique (validation, enrichissement, extraction d'un ID d'appareil pour la clé, routage vers plusieurs topics Kafka) se limite à ce que permettent le connecteur et les Single Message Transforms.

Option 2 : un bridge Python maison

Un bridge maison tient en une cinquantaine de lignes et vous donne un contrôle total. Celui-ci utilise paho-mqtt 2.x et confluent-kafka, et n'acquitte chaque message MQTT qu'une fois l'écriture confirmée par Kafka :

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

MQTT_HOST = "broker.connectmq.com"  # utilisez votre broker privé pour les sessions persistantes
MQTT_TOPIC = "testmqtt/kafka/+/telemetry"
KAFKA_TOPIC = "iot.telemetry"

producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "enable.idempotence": True,   # ni doublons ni réordonnancement lors des retries
    "acks": "all",
    "linger.ms": 20,              # petits lots, meilleur débit
})

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 connecté :", 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 seulement quand Kafka a le record
        else:
            print("Échec d'écriture Kafka, le message sera renvoyé :", 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,      # session persistante : le broker met en file pendant l'arrêt
    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)        # exécute les callbacks de livraison
finally:
    producer.flush(10)
    client.loop_stop()

Trois détails rendent ce bridge sûr :

1. clean_session=False avec un client ID fixe : le broker conserve l'abonnement et met en file les messages QoS 1 pendant le redémarrage du bridge. Lisez les sessions persistantes MQTT pour savoir combien de temps les brokers les gardent.
2. manual_ack=True : paho n'envoie pas le PUBACK automatiquement. Si le processus meurt avant la confirmation de Kafka, le broker renvoie le message.
3. L'ID de l'appareil comme clé de record : tous les messages de l'appareil 42 vont dans la même partition, dans l'ordre.

Le résultat est une livraison at-least-once. Un crash entre l'écriture Kafka et le PUBACK produit un doublon ; les consommateurs en aval doivent donc être idempotents (par exemple en dédupliquant sur un ID de message présent dans le payload).

Ordre et clés

C'est sur l'ordre que la plupart des pipelines MQTT vers Kafka se trompent. Les règles :

• MQTT ordonne les messages par topic au sein d'une session client, à QoS égale. Rien n'est promis entre topics différents ni entre publishers différents.
• Kafka ordonne les records par partition. Avec le partitionneur par défaut, une même clé tombe toujours dans la même partition.
• Utilisez donc une clé qui correspond à l'unité à ordonner : presque toujours l'ID de l'appareil, pas le topic complet et jamais une valeur aléatoire. Si un appareil publie telemetry et status sur deux topics et que vous avez besoin de leur ordre relatif, utilisez l'ID de l'appareil comme clé pour les deux.

Si vous faites évoluer le bridge horizontalement avec un abonnement partagé ($share/kafka-bridge/testmqtt/kafka/+/telemetry), le broker répartit les messages entre les instances du bridge. Les messages d'un même appareil peuvent alors être en transit sur deux instances à la fois et arriver dans Kafka dans le désordre. Options : l'accepter et réordonner en aval grâce à un timestamp ou un numéro de séquence dans le payload, ou partitionner côté MQTT (un bridge par plage de topics, par exemple +/telemetry découpé par préfixe de région). Vous pouvez vérifier le comportement avec l'approche décrite dans tester l'ordre des messages MQTT.

Option 3 : intégration Kafka native du broker

Certains brokers MQTT proposent une intégration Kafka, souvent sous forme d'extension entreprise ou d'action de moteur de règles, qui produit les messages directement depuis le processus du broker. Cela supprime un saut et un service à exploiter. Les compromis : votre pipeline dépend de ce broker, la configuration vit dans le broker plutôt que dans votre code, et la contre-pression de Kafka touche le broker lui-même. Comparez les options sur la page de comparaison des brokers avant de vous engager.

Tester le bridge de bout en bout

Démarrez un Kafka mononœud (mode KRaft) et un consommateur console qui affiche les clés :

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

Lancez le bridge dans un deuxième terminal, puis publiez quelques messages depuis deux appareils :

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

Le consommateur doit afficher les clés dev-a et dev-b avec un seq croissant par clé. Vous pouvez aussi publier depuis le client MQTT en ligne et voir les mêmes messages arriver dans Kafka. Lancez ensuite le test de panne qui compte vraiment. Il repose sur une session persistante : pointez donc le bridge vers votre propre instance de broker privée (par exemple un broker ConnectMQ dédié), car l'endpoint de test partagé ne conserve pas de sessions hors ligne et limite le débit, ce qui convient aux smoke tests mais pas aux tests de mise en file.

1. Arrêtez le bridge avec Ctrl+C.
2. Publiez cinq messages de plus.
3. Relancez le bridge : les messages en file doivent apparaître dans Kafka.
4. Arrêtez Kafka (docker stop kafka), publiez, relancez Kafka : le bridge ne doit pas envoyer de PUBACK tant que Kafka est arrêté, les messages arrivent donc après la reprise.

Si l'étape 3 perd des messages, votre session n'est pas persistante (mauvais client ID, clean_session=True ou session expirée par le broker). Si l'étape 4 en perd, vous acquittez avant la confirmation de Kafka.

Observer les topics d'entrée du bridge dans le navigateur

Un bridge MQTT vers Kafka est petit, mais ce sont les détails qui décident de sa fiabilité : sessions persistantes, acquittements liés aux confirmations de Kafka et clés de record adaptées à l'ordre dont vous avez besoin. Commencez par un connecteur s'il couvre votre mapping, passez à un bridge maison quand il vous faut de la logique, et gardez le test de redémarrage dans votre CI pour qu'il reste vrai.

Questions fréquentes