MQTT mit Apache Kafka verbinden: Connectoren, eigene Bridge und Reihenfolge

Um MQTT mit Apache Kafka zu verbinden, betreibst du eine Komponente, die MQTT-Topics auf dem Broker abonniert und jede Nachricht als Kafka-Record produziert. Praktisch gibt es drei Wege: einen MQTT-Source-Connector für Kafka Connect, eine kleine eigene Bridge oder eine Kafka-Integration, die dein MQTT-Broker mitbringt. Egal welchen du wählst, die schwierigen Teile sind dieselben: Topics abbilden, Record Keys so wählen, dass die Reihenfolge erhalten bleibt, und keine Nachrichten verlieren, wenn etwas neu startet.

Dieses Tutorial zeigt alle drei Wege, mit einer funktionierenden Python-Bridge und einem Test, den du in fünf Minuten ausführst.

Warum MQTT und Kafka zusammen

MQTT und Kafka lösen unterschiedliche Probleme. MQTT verbindet Tausende kleiner, zeitweise verbundener Geräte mit einem Broker, mit winzigen Paketen und QoS pro Nachricht. Kafka ist ein dauerhaftes, partitioniertes Log, das Backend-Services im großen Maßstab erneut lesen, joinen und verarbeiten. Geräte sollten keine Kafka-Clients ausführen, und Kafka wurde nie dafür gebaut, eine Million wackliger TCP-Sessions offen zu halten. Eine Bridge lässt jede Seite das tun, was sie gut kann. Einen ausführlicheren Vergleich findest du unter MQTT vs Kafka.

Option 1: Kafka Connect MQTT Source

Wenn du Kafka Connect bereits betreibst, ist ein Source-Connector der Weg mit dem wenigsten Code. Zwei verbreitete Optionen:

• Confluent MQTT Source Connector (io.confluent.connect.mqtt.MqttSourceConnector): wird unter der Confluent-Lizenz vertrieben; prüfe die Bedingungen für dein Deployment.
• Stream Reactor MQTT Source von Lenses: Open Source, konfiguriert über ein KCQL-Statement, das MQTT-Topics auf Kafka-Topics abbildet.

Eine Konfiguration für den Confluent-Connector sieht so aus:

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

Behalte den ByteArrayConverter für den Value, sofern Connect die Payloads nicht wirklich parsen soll: So laufen JSON, Protobuf oder CBOR unverändert durch. Prüfe in der Connector-Dokumentation, wie der Record Key befüllt wird; nutzt er das vollständige MQTT-Topic, landet jedes Gerät weiterhin in einer stabilen Partition, und genau das willst du für die Reihenfolge.

Der Kompromiss: Ein Connector lässt sich leicht neben anderen Connectoren deployen und überwachen, aber eigene Logik (Validierung, Anreicherung, Geräte-ID in den Key extrahieren, Routing auf mehrere Kafka-Topics) ist auf das beschränkt, was Connector und Single Message Transforms hergeben.

Option 2: eine eigene Python-Bridge

Eine eigene Bridge hat rund 50 Zeilen und gibt dir volle Kontrolle. Diese nutzt paho-mqtt 2.x und confluent-kafka und bestätigt jede MQTT-Nachricht erst, nachdem Kafka den Schreibvorgang bestätigt hat:

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

MQTT_HOST = "broker.connectmq.com"  # für persistente Sessions deinen privaten Broker nutzen
MQTT_TOPIC = "testmqtt/kafka/+/telemetry"
KAFKA_TOPIC = "iot.telemetry"

producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "enable.idempotence": True,   # keine Duplikate oder Umsortierung bei Retries
    "acks": "all",
    "linger.ms": 20,              # kleine Batches, besserer Durchsatz
})

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 verbunden:", 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 erst, wenn Kafka den Record hat
        else:
            print("Kafka-Schreibfehler, Nachricht wird erneut zugestellt:", 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,      # persistente Session: Broker puffert, solange wir weg sind
    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)        # führt die Delivery-Callbacks aus
finally:
    producer.flush(10)
    client.loop_stop()

Drei Details machen diese Bridge sicher:

1. clean_session=False mit fester Client-ID: Der Broker behält das Abonnement und puffert QoS-1-Nachrichten, während die Bridge neu startet. In MQTT Persistent Sessions steht, wie lange Broker sie aufbewahren.
2. manual_ack=True: paho sendet PUBACK nicht automatisch. Stirbt der Prozess, bevor Kafka bestätigt, stellt der Broker die Nachricht erneut zu.
3. Die Geräte-ID als Record Key: Alle Nachrichten von Gerät 42 landen in derselben Partition, in Reihenfolge.

Das Ergebnis ist At-least-once-Zustellung. Ein Absturz zwischen Kafka-Schreibvorgang und PUBACK erzeugt ein Duplikat, deshalb sollten nachgelagerte Consumer idempotent sein (zum Beispiel per Nachrichten-ID im Payload deduplizieren).

Reihenfolge und Keys

Bei der Reihenfolge gehen die meisten MQTT-zu-Kafka-Pipelines schief. Die Regeln:

• MQTT ordnet Nachrichten pro Topic innerhalb einer Client-Session bei gleichem QoS. Über Topics oder Publisher hinweg gibt es kein Versprechen.
• Kafka ordnet Records pro Partition. Mit dem Standard-Partitioner landet derselbe Key immer in derselben Partition.
• Nimm also einen Key, der der Einheit entspricht, die geordnet sein muss: fast immer die Geräte-ID, nicht das ganze Topic und niemals einen Zufallswert. Publisht ein Gerät telemetry und status auf zwei Topics und brauchst du sie relativ zueinander geordnet, nutze für beide die Geräte-ID als Key.

Skalierst du die Bridge horizontal mit einer Shared Subscription ($share/kafka-bridge/testmqtt/kafka/+/telemetry), verteilt der Broker die Nachrichten auf die Bridge-Instanzen. Nachrichten eines Geräts können dann gleichzeitig auf zwei Instanzen unterwegs sein und in falscher Reihenfolge in Kafka ankommen. Optionen: das akzeptieren und nachgelagert per Timestamp oder Sequenznummer im Payload umsortieren, oder die MQTT-Seite partitionieren (eine Bridge pro Topic-Bereich, etwa +/telemetry nach Regionspräfix aufgeteilt). Das Verhalten kannst du mit dem Ansatz aus MQTT-Nachrichtenreihenfolge testen prüfen.

Option 3: native Kafka-Integration im Broker

Manche MQTT-Broker bringen eine Kafka-Integration mit, oft als Enterprise-Erweiterung oder Aktion einer Rule Engine, die Nachrichten direkt aus dem Broker-Prozess produziert. Das spart einen Hop und einen zu betreibenden Service. Die Kompromisse: Deine Pipeline hängt an diesem Broker, die Konfiguration liegt im Broker statt in deinem Code, und Backpressure aus Kafka trifft den Broker selbst. Vergleiche die Optionen auf der Broker-Vergleichsseite, bevor du dich festlegst.

Die Bridge Ende-zu-Ende testen

Starte ein Single-Node-Kafka (KRaft-Modus) und einen Console-Consumer, der die Keys ausgibt:

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

Starte die Bridge in einem zweiten Terminal und publishe dann ein paar Nachrichten von zwei Geräten:

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

Der Consumer sollte die Keys dev-a und dev-b ausgeben, mit pro Key steigendem seq. Du kannst auch aus dem Online-MQTT-Client publishen und zusehen, wie dieselben Nachrichten in Kafka ankommen. Danach kommt der Fehlertest, der wirklich zählt. Er braucht eine persistente Session, also richte die Bridge auf deine eigene private Broker-Instanz aus (zum Beispiel einen dedizierten ConnectMQ-Broker): Der geteilte Test-Endpoint hält keine Offline-Sessions und ist rate-limitiert, was für Smoke-Tests reicht, aber nicht für Queueing-Tests.

1. Stoppe die Bridge mit Ctrl+C.
2. Publishe fünf weitere Nachrichten.
3. Starte die Bridge wieder: Die gepufferten Nachrichten müssen in Kafka erscheinen.
4. Stoppe Kafka (docker stop kafka), publishe, starte Kafka wieder: Die Bridge darf kein PUBACK senden, solange Kafka weg ist, die Nachrichten kommen also nach der Wiederherstellung an.

Gehen in Schritt 3 Nachrichten verloren, ist deine Session nicht persistent (falsche Client-ID, clean_session=True oder der Broker hat die Session ablaufen lassen). Gehen sie in Schritt 4 verloren, bestätigst du, bevor Kafka bestätigt.

Die Eingangs-Topics der Bridge im Browser beobachten

Eine MQTT-zu-Kafka-Bridge ist klein, aber die Details entscheiden über ihre Zuverlässigkeit: persistente Sessions, Bestätigungen, die an Kafka-Bestätigungen gekoppelt sind, und Record Keys, die zur benötigten Reihenfolge passen. Fang mit einem Connector an, wenn er dein Mapping abdeckt, wechsle zu einer eigenen Bridge, sobald du Logik brauchst, und behalte den Neustart-Test in deiner CI, damit all das auch so bleibt.

Häufig gestellte Fragen