MQTT over Kafka

The mqtt-kafka binding turns Kafka into the storage layer for an MQTT broker. Sessions, messages, and retained messages are each mapped onto their own Kafka topic, so there is no separate MQTT broker to run or scale independently of Kafka.

Prerequisites

Docker Compose

Set up the MQTT broker

Create zilla.yaml and docker-compose.yaml in the same directory.

zilla.yaml

name: MQTT-intro
bindings:
  # Proxy service entrypoint
  north_tcp_server:
    type: tcp
    kind: server
    options:
      host: 0.0.0.0
      port: 7183
    exit: north_mqtt_server

  # MQTT Broker With an exit to Kafka
  north_mqtt_server:
    type: mqtt
    kind: server
    exit: north_mqtt_kafka_mapping

  # Proxy MQTT messages to Kafka
  north_mqtt_kafka_mapping:
    type: mqtt-kafka
    kind: proxy
    options:
      topics:
        sessions: mqtt-sessions
        messages: mqtt-messages
        retained: mqtt-retained
    exit: north_kafka_cache_client

  # Kafka sync layer
  north_kafka_cache_client:
    type: kafka
    kind: cache_client
    exit: south_kafka_cache_server
  south_kafka_cache_server:
    type: kafka
    kind: cache_server
    options:
      bootstrap:
        - mqtt-sessions
        - mqtt-retained
    exit: south_kafka_client

  # Connect to Kafka
  south_kafka_client:
    type: kafka
    kind: client
    options:
      servers:
        - ${{env.KAFKA_BOOTSTRAP_SERVER}}
    exit: south_kafka_tcp_client
  south_kafka_tcp_client:
    type: tcp
    kind: client

telemetry:
  exporters:
    stdout_logs_exporter:
      type: stdout

docker-compose.yaml

version: '3'
services:

  zilla:
    image: ghcr.io/aklivity/zilla:latest
    pull_policy: always
    depends_on:
      - kafka
    ports:
      - 7183:7183
    environment:
      KAFKA_BOOTSTRAP_SERVER: "kafka:29092"
    volumes:
      - ./zilla.yaml:/etc/zilla/zilla.yaml
    command: start -v -e

  kafka:
    image: bitnami/kafka:3.5
    hostname: kafka
    ports:
      - 9092:9092
      - 29092:9092
    environment:
      ALLOW_PLAINTEXT_LISTENER: "yes"
      KAFKA_CFG_NODE_ID: "1"
      KAFKA_CFG_BROKER_ID: "1"
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: "1@127.0.0.1:9093"
      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: "CLIENT:PLAINTEXT,INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT"
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_CFG_LOG_DIRS: "/tmp/logs"
      KAFKA_CFG_PROCESS_ROLES: "broker,controller"
      KAFKA_CFG_LISTENERS: "CLIENT://:9092,INTERNAL://:29092,CONTROLLER://:9093"
      KAFKA_CFG_INTER_BROKER_LISTENER_NAME: "INTERNAL"
      KAFKA_CFG_ADVERTISED_LISTENERS: "CLIENT://localhost:9092,INTERNAL://kafka:29092"
      KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "true"

  kafka-init:
    image: bitnami/kafka:3.5
    command:
      - "/bin/bash"
      - "-c"
      -  |
        /opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server kafka:29092 --create --if-not-exists --topic mqtt-messages
        /opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server kafka:29092 --create --if-not-exists --topic mqtt-sessions --config cleanup.policy=compact
        /opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server kafka:29092 --create --if-not-exists --topic mqtt-retained --config cleanup.policy=compact
    depends_on:
      - kafka
    init: true

The north_mqtt_kafka_mapping route maps the MQTT broker onto three Kafka topics: mqtt-sessions, mqtt-messages, and mqtt-retained. The docker-compose.yaml file's kafka-init container creates all three, with cleanup.policy=compact set on the sessions and retained topics so the latest state for each key is kept.

Run Zilla and Kafka

docker-compose up --detach

Using eclipse-mosquitto, subscribe to the zilla topic:

docker run -it --rm eclipse-mosquitto \
mosquitto_sub --url mqtt://host.docker.internal:7183/zilla

In a separate session, publish a message on the zilla topic:

docker run -it --rm eclipse-mosquitto \
mosquitto_pub --url mqtt://host.docker.internal:7183/zilla --message 'Hello, world'

The subscriber receives the message immediately, relayed through the mqtt-messages Kafka topic.

Retained messages

Publish with the retained flag set:

docker run -it --rm eclipse-mosquitto \
mosquitto_pub --url mqtt://host.docker.internal:7183/zilla --message 'Hello, retained' --retain

Restart the mosquitto_sub command above. The latest retained message is delivered to the new subscriber, and the earlier, non-retained messages are not, because retained messages are read back from the compacted mqtt-retained topic.

Remove the running containers

docker-compose down

Next Steps

Walk through the MQTT to Kafka use case for a deeper look at the broker-on-Kafka pattern.
See the mqtt-kafka binding reference for every option.
Try the Running an MQTT Kafka broker how-to and the mqtt.kafka.proxy example.