MQTT'yi Apache Kafka'ya Bağlayın: Connector'lar, Özel Köprü ve Sıralama

MQTT'yi Apache Kafka'ya bağlamak için broker'daki MQTT topic'lerine subscribe olan ve her mesajı bir Kafka record'u olarak üreten bir bileşen çalıştırırsınız. Üç pratik seçeneğiniz var: bir Kafka Connect MQTT source connector'ı, kendi yazdığınız küçük bir köprü ya da MQTT broker'ınızda yerleşik bir Kafka entegrasyonu. Hangisini seçerseniz seçin zor kısımlar aynıdır: topic'leri eşlemek, sıralamanın korunacağı record key'leri seçmek ve bir şey yeniden başladığında mesaj kaybetmemek.

Bu eğitim üç seçeneği de, çalışan bir Python köprüsü ve beş dakikada koşturabileceğiniz bir testle birlikte anlatıyor.

Neden MQTT ve Kafka birlikte

MQTT ve Kafka farklı problemleri çözer. MQTT; binlerce küçük, aralıklı bağlanan cihazı minik paketler ve mesaj bazında QoS ile bir broker'a bağlar. Kafka ise backend servislerinin büyük ölçekte yeniden okuduğu, birleştirdiği ve işlediği kalıcı, partition'lara bölünmüş bir log'dur. Cihazlar Kafka istemcisi çalıştırmamalı; Kafka da bir milyon kararsız TCP session'ını açık tutmak için tasarlanmadı. Köprü, her iki tarafın da en iyi yaptığı işi yapmasını sağlar. Daha ayrıntılı karşılaştırma için MQTT vs Kafka yazısına bakın.

Seçenek 1: Kafka Connect MQTT source

Zaten Kafka Connect çalıştırıyorsanız source connector en az kod gerektiren yoldur. Yaygın iki seçenek:

• Confluent MQTT Source Connector (io.confluent.connect.mqtt.MqttSourceConnector): Confluent lisansıyla dağıtılır; kendi kurulumunuz için koşulları kontrol edin.
• Lenses'in Stream Reactor MQTT Source connector'ı: açık kaynaktır, MQTT topic'lerini Kafka topic'lerine eşleyen bir KCQL ifadesiyle yapılandırılır.

Confluent connector yapılandırması şöyle görünür:

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

Connect'in payload'ları gerçekten parse etmesini istemiyorsanız value için ByteArrayConverter kullanmaya devam edin; böylece JSON, Protobuf veya CBOR olduğu gibi geçer. Connector'ın record key'i nasıl doldurduğunu dokümantasyonundan kontrol edin; tam MQTT topic'ini kullanıyorsa her cihaz yine sabit bir partition'a gider ve sıralama için istediğiniz de budur.

Ödünleşim: connector'ı diğer connector'ların yanında kurmak ve izlemek kolaydır, ancak özel mantık (doğrulama, zenginleştirme, cihaz ID'sini key'e çıkarma, birden çok Kafka topic'ine yönlendirme) connector'ın ve Single Message Transform'ların desteklediğiyle sınırlıdır.

Seçenek 2: özel bir Python köprüsü

Özel bir köprü yaklaşık 50 satırdır ve size tam kontrol verir. Bu örnek paho-mqtt 2.x ve confluent-kafka kullanıyor ve her MQTT mesajını ancak Kafka yazmayı onayladıktan sonra onaylıyor:

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

MQTT_HOST = "broker.connectmq.com"  # kalıcı session'lar için kendi özel broker'ınızı kullanın
MQTT_TOPIC = "testmqtt/kafka/+/telemetry"
KAFKA_TOPIC = "iot.telemetry"

producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "enable.idempotence": True,   # retry'larda kopya veya sıra bozulması yok
    "acks": "all",
    "linger.ms": 20,              # küçük batch'ler, daha iyi 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 bağlandı:", 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 yalnızca record Kafka'dayken
        else:
            print("Kafka'ya yazılamadı, mesaj yeniden iletilecek:", 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,      # kalıcı session: biz kapalıyken broker kuyruğa alır
    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)        # teslimat callback'lerini çalıştırır
finally:
    producer.flush(10)
    client.loop_stop()

Bu köprüyü güvenli yapan üç ayrıntı var:

1. Sabit bir client ID ile clean_session=False: broker, köprü yeniden başlarken aboneliği korur ve QoS 1 mesajlarını kuyruğa alır. Broker'ların bunları ne kadar sakladığını MQTT kalıcı session'lar yazısında bulabilirsiniz.
2. manual_ack=True: paho PUBACK'i otomatik göndermez. Süreç Kafka onaylamadan ölürse broker mesajı yeniden iletir.
3. Record key olarak cihaz ID'si: 42 numaralı cihazın tüm mesajları aynı partition'a, sırayla gider.

Sonuç at-least-once teslimattır. Kafka yazımı ile PUBACK arasında bir çökme kopya üretir; bu yüzden sonraki consumer'lar idempotent olmalıdır (örneğin payload içindeki bir mesaj ID'siyle tekilleştirme).

Sıralama ve key'ler

MQTT'den Kafka'ya pipeline'ların çoğu sıralamada hata yapar. Kurallar şunlar:

• MQTT, aynı QoS ile tek bir istemci session'ı içinde mesajları topic bazında sıralar. Farklı topic'ler veya farklı publisher'lar arasında bir garanti vermez.
• Kafka, record'ları partition bazında sıralar. Varsayılan partitioner ile aynı key her zaman aynı partition'a düşer.
• O hâlde sıralı olması gereken birimle eşleşen bir key kullanın: neredeyse her zaman cihaz ID'si; tam topic değil ve asla rastgele bir değer değil. Bir cihaz telemetry ve status mesajlarını iki ayrı topic'e publish ediyorsa ve bunların birbirine göre sıralı olması gerekiyorsa, ikisinde de cihaz ID'sini key yapın.

Köprüyü bir shared subscription ($share/kafka-bridge/testmqtt/kafka/+/telemetry) ile yatayda ölçeklerseniz broker mesajları köprü instance'ları arasında dağıtır. Bu durumda aynı cihazın mesajları aynı anda iki instance'ta işleniyor olabilir ve Kafka'ya sırasız ulaşabilir. Seçenekler: bunu kabul edip sonradan payload'daki timestamp veya sıra numarasıyla yeniden sıralamak ya da MQTT tarafını bölmek (her topic aralığı için bir köprü; örneğin +/telemetry bölge önekine göre ayrılmış). Davranışı MQTT mesaj sıralamasını test etme yazısındaki yaklaşımla doğrulayabilirsiniz.

Seçenek 3: broker'a yerleşik Kafka entegrasyonu

Bazı MQTT broker'ları, mesajları doğrudan broker sürecinden üreten bir Kafka entegrasyonu sunar; genellikle enterprise bir eklenti ya da kural motoru aksiyonu olarak. Bu, bir ara adımı ve işletilecek bir servisi ortadan kaldırır. Ödünleşimler: pipeline'ınız o broker'a bağlanır, yapılandırma kod tabanınızda değil broker'da yaşar ve Kafka'dan gelen backpressure doğrudan broker'ı etkiler. Karar vermeden önce seçenekleri broker karşılaştırma sayfasında karşılaştırın.

Köprüyü uçtan uca test edin

Tek node'lu bir Kafka (KRaft modu) ve key'leri yazdıran bir console consumer başlatın:

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

Köprüyü ikinci bir terminalde çalıştırın, ardından iki cihazdan birkaç mesaj publish edin:

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

Consumer, dev-a ve dev-b key'lerini her key için artan seq değerleriyle göstermelidir. Aynı mesajların Kafka'ya ulaştığını görmek için online MQTT istemcisinden de publish edebilirsiniz. Ardından asıl önemli olan hata testini çalıştırın. Bu test kalıcı bir session'a dayanır; bu yüzden köprüyü kendi özel broker instance'ınıza (örneğin dedicated bir ConnectMQ broker'ına) yönlendirin. Paylaşımlı test endpoint'i çevrimdışı session saklamaz ve hız sınırlıdır; smoke test için yeterlidir ama kuyruk testleri için uygun değildir.

1. Köprüyü Ctrl+C ile durdurun.
2. Beş mesaj daha publish edin.
3. Köprüyü yeniden başlatın: kuyruktaki mesajlar Kafka'da görünmelidir.
4. Kafka'yı durdurun (docker stop kafka), publish edin, Kafka'yı yeniden başlatın: Kafka kapalıyken köprü PUBACK göndermemelidir; mesajlar toparlanmadan sonra gelir.

1. adımda mesaj kaybolursa session'ınız kalıcı değildir (yanlış client ID, clean_session=True ya da broker session'ın süresini doldurmuştur). 4. adımda kaybolursa Kafka onaylamadan önce ack gönderiyorsunuz demektir.

Köprünün giriş topic'lerini tarayıcıda izleyin

MQTT'den Kafka'ya bir köprü küçüktür, ama güvenilir olup olmadığını ayrıntılar belirler: kalıcı session'lar, Kafka onaylarına bağlı ack'ler ve ihtiyaç duyduğunuz sıralamayla eşleşen record key'ler. Eşlemenizi karşılıyorsa bir connector ile başlayın, mantık gerektiğinde özel köprüye geçin ve yeniden başlatma testini CI'ınızda tutun ki bu garantiler geçerliliğini korusun.

Sıkça sorulan sorular