Bridge MQTT to Apache Kafka: Connectors, Custom Bridges and Ordering

To bridge MQTT to Apache Kafka, you run a component that subscribes to MQTT topics on the broker and produces each message as a Kafka record. You have three practical options: a Kafka Connect MQTT source connector, a small custom bridge you write yourself, or a Kafka integration built into your MQTT broker. Whichever you pick, the hard parts are the same: mapping topics, choosing record keys so ordering survives, and not losing messages when something restarts.

This tutorial walks through all three, with a working Python bridge and a test you can run in five minutes.

Why MQTT and Kafka together

MQTT and Kafka solve different problems. MQTT connects thousands of small, intermittently connected devices to a broker with tiny packets and per-message QoS. Kafka is a durable, partitioned log that backend services replay, join and process at scale. Devices should not run Kafka clients, and Kafka was never meant to keep a million flaky TCP sessions open. A bridge lets each side do what it is good at. For a deeper comparison see MQTT vs Kafka.

Option 1: Kafka Connect MQTT source

If you already run Kafka Connect, a source connector is the least code. Two common choices:

• Confluent MQTT Source Connector (io.confluent.connect.mqtt.MqttSourceConnector) — distributed under Confluent's licence; check the terms for your deployment.
• Stream Reactor MQTT Source from Lenses — open source, configured with a KCQL statement that maps MQTT topics to Kafka topics.

A Confluent connector configuration looks like this:

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

Keep ByteArrayConverter for the value unless you really want Connect to parse payloads — it passes JSON, Protobuf or CBOR through untouched. Check the connector documentation for how it fills the record key; if it uses the full MQTT topic, every device still maps to a stable partition, which is what you want for ordering.

The trade-off: a connector is easy to deploy and monitor next to other connectors, but custom logic (validation, enrichment, extracting a device ID into the key, routing to several Kafka topics) is limited to what the connector and Single Message Transforms support.

Option 2: a custom Python bridge

A custom bridge is about 50 lines and gives you full control. This one uses paho-mqtt 2.x and confluent-kafka, and it acknowledges each MQTT message only after Kafka has confirmed the write:

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

MQTT_HOST = "broker.connectmq.com"  # use your private broker for persistent sessions
MQTT_TOPIC = "testmqtt/kafka/+/telemetry"
KAFKA_TOPIC = "iot.telemetry"

producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "enable.idempotence": True,   # no duplicates or reordering on producer retries
    "acks": "all",
    "linger.ms": 20,              # small batches, better 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 connected:", 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 only after Kafka has the record
        else:
            print("Kafka write failed, message will be redelivered:", 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,      # persistent session: broker queues while we are down
    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)        # runs delivery callbacks
finally:
    producer.flush(10)
    client.loop_stop()

Three details make this bridge safe:

1. clean_session=False with a fixed client ID — the broker keeps the subscription and queues QoS 1 messages while the bridge restarts. Read MQTT persistent sessions for how long brokers keep them.
2. manual_ack=True — paho does not send PUBACK automatically. If the process dies before Kafka confirms, the broker redelivers the message.
3. The device ID as the record key — every message from device 42 goes to the same partition, in order.

The result is at-least-once delivery. A crash between the Kafka write and the PUBACK produces a duplicate, so downstream consumers should be idempotent (for example, deduplicate on a message ID in the payload).

Ordering and keys

Ordering is where most MQTT-to-Kafka pipelines go wrong. The rules:

• MQTT orders messages per topic within one client session at the same QoS. It makes no promise across topics or across publishers.
• Kafka orders records per partition. With the default partitioner, the same key always lands in the same partition.
• So use a key that matches the unit you need ordered — almost always the device ID, not the full topic and never a random value. If a device publishes telemetry and status on two topics and you need them ordered relative to each other, key both by device ID.

If you scale the bridge horizontally with a shared subscription ($share/kafka-bridge/testmqtt/kafka/+/telemetry), the broker spreads messages across bridge instances. Messages from one device can then be in flight on two instances at once and reach Kafka out of order. Options: accept it and reorder downstream by a timestamp or sequence number in the payload, or partition the MQTT side instead (one bridge per topic range, such as +/telemetry split by region prefix). You can verify the behaviour with the approach in testing MQTT message ordering.

Option 3: broker-native Kafka integration

Some MQTT brokers ship a Kafka integration — often as an enterprise extension or rule-engine action — that produces messages directly from the broker process. That removes one hop and one service to operate. The trade-offs: it ties your pipeline to that broker, configuration lives in the broker rather than in your codebase, and backpressure from Kafka affects the broker itself. Compare options on the broker comparison page before you lock in.

Test the bridge end to end

Start a single-node Kafka (KRaft mode) and a console consumer that prints keys:

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

Run the bridge in a second terminal, then publish a few messages from two devices:

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

The consumer should print dev-a and dev-b keys with seq increasing per key. You can also publish from the online MQTT client to watch the same messages arrive in Kafka. Then run the failure test that actually matters. It relies on a persistent session, so point the bridge at your own private broker instance (for example a dedicated ConnectMQ broker): the shared test endpoint keeps no offline sessions and is rate-limited, which is fine for smoke tests but not for queueing tests.

1. Stop the bridge with Ctrl+C.
2. Publish five more messages.
3. Start the bridge again — the queued messages must appear in Kafka.
4. Stop Kafka (docker stop kafka), publish, start Kafka again — the bridge must not PUBACK while Kafka is down, so the messages arrive after recovery.

If step 3 loses messages, your session is not persistent (wrong client ID, clean_session=True, or the broker expired the session). If step 4 loses them, you are acknowledging before Kafka confirms.

Watch the bridge's input topics in your browser

An MQTT-to-Kafka bridge is small, but the details decide whether it is reliable: persistent sessions, acknowledgements tied to Kafka confirmations, and record keys that match the order you need. Start with a connector if it covers your mapping, switch to a custom bridge when you need logic, and keep the restart test in your CI so it stays true.

Frequently asked questions