Puente de MQTT a Apache Kafka: conectores, puentes propios y orden

Para llevar MQTT a Apache Kafka ejecutas un componente que se suscribe a topics MQTT en el broker y produce cada mensaje como un registro de Kafka. Tienes tres opciones prácticas: un conector MQTT source de Kafka Connect, un pequeño puente propio o una integración con Kafka incluida en tu broker MQTT. Elijas la que elijas, las partes difíciles son las mismas: mapear topics, elegir claves de registro para que el orden sobreviva y no perder mensajes cuando algo se reinicia.

Este tutorial recorre las tres opciones, con un puente en Python que funciona y una prueba que puedes ejecutar en cinco minutos.

Por qué MQTT y Kafka juntos

MQTT y Kafka resuelven problemas distintos. MQTT conecta miles de dispositivos pequeños, con conexión intermitente, a un broker con paquetes diminutos y QoS por mensaje. Kafka es un log duradero y particionado que los servicios de backend reproducen, combinan y procesan a escala. Los dispositivos no deberían ejecutar clientes de Kafka, y Kafka nunca se diseñó para mantener abiertas un millón de sesiones TCP inestables. Un puente permite que cada lado haga lo que mejor sabe hacer. Para una comparación más profunda, consulta MQTT vs Kafka.

Opción 1: Kafka Connect MQTT source

Si ya ejecutas Kafka Connect, un conector source es lo que menos código requiere. Dos opciones habituales:

• Confluent MQTT Source Connector (io.confluent.connect.mqtt.MqttSourceConnector): se distribuye bajo la licencia de Confluent; revisa los términos para tu despliegue.
• Stream Reactor MQTT Source de Lenses: open source, se configura con una sentencia KCQL que mapea topics MQTT a topics de Kafka.

Una configuración del conector de Confluent tiene este aspecto:

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

Mantén ByteArrayConverter para el valor salvo que realmente quieras que Connect interprete los payloads: así JSON, Protobuf o CBOR pasan sin cambios. Revisa en la documentación del conector cómo rellena la clave del registro; si usa el topic MQTT completo, cada dispositivo sigue yendo a una partición estable, que es lo que quieres para el orden.

La contrapartida: un conector es fácil de desplegar y monitorizar junto a otros conectores, pero la lógica personalizada (validación, enriquecimiento, extraer un ID de dispositivo para la clave, enrutar a varios topics de Kafka) se limita a lo que soportan el conector y las Single Message Transforms.

Opción 2: un puente propio en Python

Un puente propio ocupa unas 50 líneas y te da control total. Este usa paho-mqtt 2.x y confluent-kafka, y confirma cada mensaje MQTT solo después de que Kafka haya confirmado la escritura:

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

MQTT_HOST = "broker.connectmq.com"  # usa tu broker privado para sesiones persistentes
MQTT_TOPIC = "testmqtt/kafka/+/telemetry"
KAFKA_TOPIC = "iot.telemetry"

producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "enable.idempotence": True,   # sin duplicados ni reordenación en los reintentos
    "acks": "all",
    "linger.ms": 20,              # lotes pequeños, mejor 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 solo cuando Kafka tiene el registro
        else:
            print("Fallo al escribir en Kafka, el mensaje se reenviará:", 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,      # sesión persistente: el broker encola mientras estamos caídos
    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)        # ejecuta los callbacks de entrega
finally:
    producer.flush(10)
    client.loop_stop()

Tres detalles hacen que este puente sea seguro:

1. clean_session=False con un client ID fijo: el broker conserva la suscripción y encola los mensajes QoS 1 mientras el puente se reinicia. Lee sesiones persistentes en MQTT para saber cuánto tiempo las conservan los brokers.
2. manual_ack=True: paho no envía PUBACK automáticamente. Si el proceso muere antes de que Kafka confirme, el broker reenvía el mensaje.
3. El ID del dispositivo como clave del registro: todos los mensajes del dispositivo 42 van a la misma partición, en orden.

El resultado es una entrega at-least-once. Una caída entre la escritura en Kafka y el PUBACK produce un duplicado, así que los consumidores posteriores deben ser idempotentes (por ejemplo, deduplicar por un ID de mensaje incluido en el payload).

Orden y claves

El orden es donde fallan la mayoría de los pipelines de MQTT a Kafka. Las reglas:

• MQTT ordena los mensajes por topic dentro de una sesión de cliente con el mismo QoS. No promete nada entre topics distintos ni entre publishers distintos.
• Kafka ordena los registros por partición. Con el particionador por defecto, la misma clave siempre cae en la misma partición.
• Por tanto, usa una clave que coincida con la unidad que necesitas ordenar: casi siempre el ID del dispositivo, no el topic completo y nunca un valor aleatorio. Si un dispositivo publica telemetry y status en dos topics y necesitas que estén ordenados entre sí, usa el ID del dispositivo como clave en ambos.

Si escalas el puente horizontalmente con una suscripción compartida ($share/kafka-bridge/testmqtt/kafka/+/telemetry), el broker reparte los mensajes entre las instancias del puente. Los mensajes de un mismo dispositivo pueden entonces estar en vuelo en dos instancias a la vez y llegar a Kafka desordenados. Opciones: aceptarlo y reordenar después por un timestamp o número de secuencia del payload, o particionar el lado MQTT (un puente por rango de topics, por ejemplo +/telemetry dividido por prefijo de región). Puedes verificar el comportamiento con el enfoque de probar el orden de los mensajes MQTT.

Opción 3: integración con Kafka nativa del broker

Algunos brokers MQTT incluyen una integración con Kafka, a menudo como extensión enterprise o como acción de su motor de reglas, que produce los mensajes directamente desde el proceso del broker. Eso elimina un salto y un servicio que operar. Las contrapartidas: ata tu pipeline a ese broker, la configuración vive en el broker en lugar de en tu código y la contrapresión de Kafka afecta al propio broker. Compara opciones en la página de comparación de brokers antes de comprometerte.

Prueba el puente de extremo a extremo

Arranca un Kafka de un solo nodo (modo KRaft) y un consumidor de consola que imprima las claves:

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

Ejecuta el puente en una segunda terminal y publica algunos mensajes desde dos 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

El consumidor debería mostrar las claves dev-a y dev-b con seq creciente por clave. También puedes publicar desde el cliente MQTT online y ver cómo llegan los mismos mensajes a Kafka. Después ejecuta la prueba de fallos que de verdad importa. Depende de una sesión persistente, así que apunta el puente a tu propia instancia privada de broker (por ejemplo, un broker dedicado de ConnectMQ): el endpoint de pruebas compartido no guarda sesiones offline y tiene límite de tasa, lo que vale para smoke tests pero no para probar colas.

1. Detén el puente con Ctrl+C.
2. Publica cinco mensajes más.
3. Vuelve a arrancar el puente: los mensajes encolados deben aparecer en Kafka.
4. Detén Kafka (docker stop kafka), publica y vuelve a arrancar Kafka: el puente no debe enviar PUBACK mientras Kafka está caído, así que los mensajes llegan tras la recuperación.

Si en el paso 3 se pierden mensajes, tu sesión no es persistente (client ID incorrecto, clean_session=True o el broker expiró la sesión). Si se pierden en el paso 4, estás confirmando antes de que Kafka lo haga.

Observa los topics de entrada del puente en el navegador

Un puente de MQTT a Kafka es pequeño, pero los detalles deciden si es fiable: sesiones persistentes, confirmaciones ligadas a las de Kafka y claves de registro que coinciden con el orden que necesitas. Empieza con un conector si cubre tu mapeo, pasa a un puente propio cuando necesites lógica y mantén la prueba de reinicio en tu CI para que siga siendo cierta.

Preguntas frecuentes