Kafka on arm64: KRaft Mode, Topic Design, and the MQTT Producer Bridge

Series: Building a Real-Time Data Platform from the Edge — Post 4 of 7. The full source is on GitHub: rpi-data-platform

---

MQTT and Kafka both move messages. That is where the similarity ends.

MQTT is a push protocol designed for constrained devices. It is connectionless from the consumer’s perspective — if your consumer is down when a message arrives, the message is gone (unless you configured persistence, and even then it is limited). Kafka is a distributed commit log. Messages sit on disk, indexed by offset, readable by any number of consumers at any time, replayable from any point. The two worlds are architecturally incompatible, and the Python bridge in this post is the translation layer between them.

But before we write the bridge, we need a Kafka cluster. On a Raspberry Pi 400. Without Zookeeper.

---

Why KRaft, and what it actually changes

Before Kafka 3.3, every Kafka deployment required a separate Zookeeper ensemble to manage cluster metadata — broker registration, topic configuration, partition leadership, and controller election. Zookeeper is a separate JVM process with its own configuration, its own failure modes, and its own memory budget. On a Pi with 4GB of RAM, that matters.

KRaft (Kafka Raft) replaces Zookeeper by embedding metadata management directly into Kafka itself. One of the brokers is elected as the controller, and cluster metadata is stored in an internal topic called __cluster_metadata using the Raft consensus protocol. There is no external dependency.

The architecture shift has two practical consequences for this setup.

First, the cluster startup sequence simplifies. With Zookeeper, you had to wait for Zookeeper to be healthy before starting brokers, and handle the case where a broker registered before Zookeeper was ready. With KRaft, there is one service.

Second, the controller knows about metadata changes synchronously. In the old architecture, the controller learned about changes via Zookeeper watches — an eventually consistent path. In KRaft, metadata changes go through the same Raft log that the controller manages. For a single-node setup this is invisible, but it is good to understand why the model is cleaner.

What is actually inside __cluster_metadata

__cluster_metadata is not a regular Kafka topic — it is an internal Raft log. You cannot consume it with kafka-console-consumer.sh. But you can inspect it with kafka-metadata-shell.sh, an interactive tool that ships with Kafka:

docker exec -it kafka kafka-metadata-shell.sh \
  --snapshot /var/lib/kafka/data/__cluster_metadata-0/00000000000000000000.checkpoint

Inside the shell, you can list brokers, topic partitions, and leader assignments:

>> ls /brokers
1
>> ls /topics
sensor-events
>> cat /topics/sensor-events/partitions/0/data
{"partitionId":0,"topicId":"...","replicas":[1],"isr":[1],"leader":1,...}

This is the ground truth for partition leadership — not Zookeeper, not a separate coordination service, just a Raft-replicated log that lives alongside the broker. When the controller processes a leader election or a topic creation, it appends a record here. The brokers receive those records and update their local metadata cache.

You will probably never need to look at this outside of debugging, but knowing it exists — and knowing it is inspectable — is part of understanding why KRaft is architecturally cleaner than the Zookeeper model.

There is one non-obvious configuration step. Each KRaft cluster needs a unique cluster ID, generated once and baked into the broker’s data directory. This is a UUID that Kafka uses to prevent a broker from accidentally joining the wrong cluster. We generate it with kafka-storage.sh before the broker starts.

---

Image choice: apache/kafka

The Apache project publishes its own official Docker image at apache/kafka. It ships multi-arch builds including linux/arm64, is actively maintained, and has no third-party packaging in the way. It is the right choice.

apache/kafka:3.9.1

Confluent’s images do not publish native arm64 builds — on a Pi they either run under emulation or not at all. Bitnami used to be the standard workaround, but they moved their images behind a paywall in late 2025. The Apache image is now the straightforward answer.

One difference from Bitnami worth knowing upfront: the Apache image uses plain Kafka property names as environment variables, with dots replaced by underscores. KAFKA_NODE_ID maps to node.id, KAFKA_PROCESS_ROLES maps to process.roles, and so on. There is no KAFKA_CFG_ prefix.

---

The docker-compose.yml

The goal is that docker compose up -d from a fresh clone produces a fully working Kafka setup with no manual steps. That means two things need to happen automatically: generating and persisting the cluster ID, and creating the topic once the broker is healthy. Both are handled by short-lived init containers that run before the main broker and producer services start.

# services/kafka/docker-compose.yml

services:

  # ── Init: generate cluster ID and write to shared volume ──────────────────
  kafka-init:
    image: apache/kafka:3.9.1
    container_name: kafka-init
    restart: "no"
    user: "0"
    volumes:
      - kafka_cluster_id:/cluster-id
    entrypoint:
      - bash
      - -c
      - |
        if [ -f /cluster-id/CLUSTER_ID ]; then
          echo 'Cluster ID already exists, skipping.'
        else
          /opt/kafka/bin/kafka-storage.sh random-uuid > /cluster-id/CLUSTER_ID
          echo "Generated cluster ID: $(cat /cluster-id/CLUSTER_ID)"
        fi

  # ── Broker ────────────────────────────────────────────────────────────────
  kafka:
    image: apache/kafka:3.9.1
    container_name: kafka
    restart: unless-stopped
    depends_on:
      kafka-init:
        condition: service_completed_successfully
    environment:
      # KRaft mode: this broker is both controller and broker
      KAFKA_NODE_ID: "1"
      KAFKA_PROCESS_ROLES: "broker,controller"
      KAFKA_CONTROLLER_QUORUM_VOTERS: "1@kafka:9093"

      # Listeners
      KAFKA_LISTENERS: "PLAINTEXT://:9092,CONTROLLER://:9093"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka:9092"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"
      KAFKA_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_INTER_BROKER_LISTENER_NAME: "PLAINTEXT"

      # Cluster ID read from the shared volume written by kafka-init
      CLUSTER_ID_FILE: "/cluster-id/CLUSTER_ID"

      # Required for single-replica operation
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: "1"
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: "1"
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: "1"

      # Log retention: keep 7 days or 500MB, whichever comes first
      KAFKA_LOG_RETENTION_HOURS: "168"
      KAFKA_LOG_RETENTION_BYTES: "524288000"
      KAFKA_LOG_SEGMENT_BYTES: "104857600"

      # Heap: 512MB is plenty for sensor-scale throughput
      KAFKA_HEAP_OPTS: "-Xmx512m -Xms256m"

    ports:
      - "9092:9092"

    volumes:
      - kafka_data:/var/lib/kafka/data
      - kafka_cluster_id:/cluster-id

    networks:
      - data-platform

    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list"]
      interval: 10s
      timeout: 5s
      retries: 10
      start_period: 30s

  # ── Init: create topic once broker is healthy ─────────────────────────────
  kafka-topic-init:
    image: apache/kafka:3.9.1
    container_name: kafka-topic-init
    restart: "no"
    depends_on:
      kafka:
        condition: service_healthy
    networks:
      - data-platform
    entrypoint:
      - bash
      - -c
      - |
        echo 'Checking for topic sensor-events...'
        if /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --list | grep -q '^sensor-events$'; then
          echo 'Topic already exists, skipping.'
        else
          /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 \
            --create \
            --topic sensor-events \
            --partitions 3 \
            --replication-factor 1 \
            --config retention.ms=604800000 \
            --config segment.bytes=104857600
          echo 'Topic sensor-events created.'
        fi

volumes:
  kafka_data:
    driver: local
  kafka_cluster_id:
    driver: local

There are a few design decisions worth explaining here.

Why a shared volume for the cluster ID? The apache/kafka image supports CLUSTER_ID_FILE — it reads the cluster ID from a file path at startup. kafka-init generates the UUID once, writes it to a named volume, and kafka reads it on every subsequent start without ever touching .env. The volume persists across restarts; the init container checks for the file and skips generation if it already exists. No manual steps, no .env editing, no stale state.

restart: "no" on init containers — init containers are meant to run once and exit. restart: "no" prevents Docker from restarting them if they exit with code 0 (which they always should). Combined with depends_on: condition: service_completed_successfully, this gives you a proper dependency chain: kafka-init completes → kafka starts → kafka-topic-init waits for healthy → creates topic.

condition: service_healthy on topic init — the broker needs time to complete leader election before it can accept client connections. Using service_healthy rather than service_started means kafka-topic-init only runs once the healthcheck passes, not the moment the container starts. Without this, the topic creation command races against broker startup and fails intermittently.

The three single-replica settings — KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR, KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR, and KAFKA_TRANSACTION_STATE_LOG_MIN_ISR all default to 3 in Kafka’s broker config, anticipating a multi-node cluster. On a single broker with replication-factor 1 you must override all three to 1, otherwise the broker starts but internal topics fail to initialise and you get confusing errors when clients connect.

PROCESS_ROLES: broker,controller — in a multi-node cluster you would separate these roles onto different nodes. In a single-node setup, the same process handles both. This is fine and documented as supported by the Kafka project.

Listener separation — Kafka distinguishes between internal cluster communication (CONTROLLER on port 9093) and client-facing communication (PLAINTEXT on port 9092). The ADVERTISED_LISTENERS is what brokers tell clients to connect to. Using kafka (the Docker service name) rather than localhost here matters: other containers on the same Docker network resolve kafka correctly, while localhost would point to their own container.

Heap sizing — Kafka’s default heap is 1GB. On a Pi with 4GB shared across the entire stack, that is too aggressive. 512MB max is sufficient for kilobytes-per-second sensor throughput. If you see GC overhead limit exceeded in the logs, raise it — but 512MB should be comfortable here.

---

Starting Kafka and verifying it

ssh pi@192.168.1.50
cd rpi-data-platform
docker compose up -d

That is the entire first-run process. Compose resolves the dependency chain automatically: kafka-init runs first, writes the cluster ID, then kafka starts and initialises its data directory, then kafka-topic-init waits for the broker healthcheck to pass before creating the topic. On every subsequent docker compose up -d, both init containers check their respective conditions and exit immediately without doing anything.

Watch it happen in real time:

docker compose logs -f kafka-init kafka kafka-topic-init

A clean first-run sequence looks like this:

kafka-init       | Generated cluster ID: MkU3OEVBNTcwNTJENDM2Qk
kafka            | [KafkaServer id=1] started
kafka-topic-init | Topic sensor-events created.

On a second run:

kafka-init       | Cluster ID already exists, skipping.
kafka-topic-init | Topic already exists, skipping.

If you intentionally want a clean slate, remove the volumes:

docker compose down -v
docker compose up -d

-v removes all named volumes including kafka_data and kafka_cluster_id. The init containers will regenerate everything from scratch on the next start.

---

Topic design

We have two sensors with meaningfully different characteristics:

• SNZB-04P (door/window): event-driven — emits only on state change (open/close) plus periodic availability heartbeats. Very low volume; can go minutes without a message.
• S60ZBTPF (smart plug): continuous — emits power, voltage, current, and energy readings every ~10 seconds. Higher volume, predictable cadence.

The architecturally relevant question is: one topic or two?

One topic: sensor-events

All sensor messages land in a single topic. The Spark consumer reads one stream and branches by device type. Simpler producer, simpler consumer, easier offset management.

The downside: you cannot apply different retention policies per sensor type, and you cannot independently scale consumers per sensor. Neither matters at this scale.

Two topics: sensor-door and sensor-power

Cleaner separation of concerns. Independent retention. Independent partition counts. But two consumers to manage, two sets of offsets to track.

For this stack, one topic wins. The complexity of two topics is not justified by sensor-scale throughput. This is a decision worth revisiting if you add a dozen sensors with wildly different data rates.

Partition count: what the number actually controls

# Create the topic on the Pi
docker exec kafka kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --create \
  --topic sensor-events \
  --partitions 3 \
  --replication-factor 1 \
  --config retention.ms=604800000 \
  --config segment.bytes=104857600

# Verify
docker exec kafka kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --topic sensor-events

Why 3 partitions? Not for throughput — a single sensor produces a few kilobytes per second at most. Partitions here serve two other purposes: they determine Spark’s task parallelism at read time (one task per partition), and they let you scale without recreating the topic. Changing partition count on a live topic is disruptive because Kafka guarantees ordering within a partition, not across partitions. Start with 3.

The deeper point is how Kafka decides which partition a message lands in. When a producer sends a message with a key — as our bridge does, keying by device_id — Kafka runs the key bytes through the murmur2 hash function and takes the result modulo the partition count:

partition = murmur2(key_bytes) % num_partitions

This is deterministic. "door-sensor-01" always hashes to the same partition, regardless of which broker you connect to or which version of the producer library you use (the murmur2 variant Kafka uses is fixed in the protocol). All messages from the same device land in the same partition, in order of arrival.

This matters downstream. With two devices keyed by device_id and 3 partitions, messages distribute across 2 of the 3 partitions — whichever two the hashes map to. The third stays empty. That is fine; you are not paying for it in any meaningful way. What you are buying is per-device ordering, which becomes important in Post 5: if you need to detect that the door opened three times within 60 seconds, you need those events ordered within a partition, not globally scattered across the topic.

Why replication-factor 1? Single broker, single replica. There is no second broker to replicate to. This is expected and correct.

---

The Python producer bridge

The bridge does three things:

1. Subscribes to Zigbee2MQTT MQTT topics on Mosquitto
2. Enriches each payload with a device_id and ingested_at timestamp
3. Produces to the sensor-events Kafka topic

# services/producer/mqtt_kafka_bridge.py

import json
import logging
import os
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
from kafka import KafkaProducer
from kafka.errors import KafkaError

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
log = logging.getLogger("mqtt-kafka-bridge")

KAFKA_BOOTSTRAP = os.getenv("KAFKA_BOOTSTRAP", "kafka:9092")
MQTT_HOST = os.getenv("MQTT_HOST", "mosquitto")
MQTT_PORT = int(os.getenv("MQTT_PORT", "1883"))
MQTT_TOPIC_PREFIX = os.getenv("MQTT_TOPIC_PREFIX", "zigbee2mqtt")
KAFKA_TOPIC = os.getenv("KAFKA_TOPIC", "sensor-events")


def make_producer() -> KafkaProducer:
    return KafkaProducer(
        bootstrap_servers=KAFKA_BOOTSTRAP,
        value_serializer=lambda v: json.dumps(v).encode("utf-8"),
        key_serializer=lambda k: k.encode("utf-8") if k else None,
        acks="all",
        retries=5,
        retry_backoff_ms=500,
        request_timeout_ms=10_000,
    )


producer = make_producer()


def on_connect(client, userdata, flags, rc):
    if rc == 0:
        log.info("Connected to Mosquitto")
        client.subscribe(f"{MQTT_TOPIC_PREFIX}/#")
        log.info("Subscribed to %s/#", MQTT_TOPIC_PREFIX)
    else:
        log.error("MQTT connection failed, rc=%d", rc)


def on_message(client, userdata, msg):
    parts = msg.topic.split("/")
    # zigbee2mqtt/bridge/# topics are internal status — not sensor data
    if len(parts) < 2 or parts[1] == "bridge":
        return

    device_id = parts[1]

    try:
        payload = json.loads(msg.payload.decode("utf-8"))
    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
        log.warning("Malformed payload from %s: %s", device_id, exc)
        return

    if not isinstance(payload, dict):
        return

    event = {
        "device_id": device_id,
        "ingested_at": datetime.now(timezone.utc).isoformat(),
        "payload": payload,
    }

    try:
        future = producer.send(topic=KAFKA_TOPIC, key=device_id, value=event)
        future.add_errback(lambda exc: log.error("Kafka produce error: %s", exc))
    except KafkaError as exc:
        log.error("Failed to produce message: %s", exc)


def main():
    client = mqtt.Client()
    client.on_connect = on_connect
    client.on_message = on_message

    log.info("Connecting to Mosquitto at %s:%d", MQTT_HOST, MQTT_PORT)
    client.connect(MQTT_HOST, MQTT_PORT, keepalive=60)

    try:
        client.loop_forever()
    except KeyboardInterrupt:
        log.info("Shutting down")
    finally:
        producer.flush(timeout=10)
        producer.close()


if __name__ == "__main__":
    main()

The bridge runs in a Docker container. services/producer/Dockerfile builds the image:

# services/producer/Dockerfile

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY mqtt_kafka_bridge.py .

CMD ["python", "-u", "mqtt_kafka_bridge.py"]

requirements.txt pins the two dependencies:

paho-mqtt==1.6.1
kafka-python-ng==2.2.3

services/producer/docker-compose.yml wires the bridge into the stack and waits for both Mosquitto and Kafka to be healthy before starting:

# services/producer/docker-compose.yml

services:

  producer:
    build: .
    container_name: producer
    restart: unless-stopped
    depends_on:
      mosquitto:
        condition: service_healthy
      kafka:
        condition: service_healthy
    environment:
      - KAFKA_BOOTSTRAP=kafka:9092
      - MQTT_HOST=mosquitto
      - MQTT_PORT=1883
      - MQTT_TOPIC_PREFIX=zigbee2mqtt
      - KAFKA_TOPIC=sensor-events
    networks:
      - data-platform

All three service files — MQTT, Kafka, and producer — are composed together by the root docker-compose.yml at the repo root using Docker Compose’s include: directive:

# docker-compose.yml

include:
  - services/kafka/docker-compose.yml
  - services/mqtt/docker-compose.yml
  - services/producer/docker-compose.yml

networks:
  data-platform:
    driver: bridge

This is what docker compose up -d at the repo root actually runs. Each service file owns its own containers and volumes; the root file owns the shared network.

A few design decisions worth calling out.

acks="all" — this tells the producer to wait until all in-sync replicas have acknowledged the write. With replication-factor 1, the in-sync replica set contains only the leader, so acks="all" and acks=1 are operationally identical — there is nobody else to wait for. The reason to set it anyway: the day you move to a three-broker cluster with replication-factor 3 and min.insync.replicas 2, your producer configuration is already correct without a code change.

Filtering bridge topics — Zigbee2MQTT publishes internal status messages under zigbee2mqtt/bridge/# (coordinator state, device list, log output). These are not sensor events. The parts[1] == "bridge" check drops them before they reach Kafka. Without this, the bridge would attempt to parse and forward JSON blobs that have nothing to do with sensor data.

Keying by device_id — via the murmur2 hash described above, all messages from door-sensor-01 land in the same partition, in arrival order. This is a prerequisite for the watermarking logic in Post 5: per-device ordering lets Spark detect sequences and apply event-time windowing correctly.

Non-blocking produce — producer.send() is asynchronous. The add_errback handles delivery failures without blocking the MQTT message loop. If the bridge blocked waiting for Kafka acknowledgement, it would fall behind on MQTT messages during any Kafka hiccup.

---

Verifying end-to-end

Run these from the Pi with the bridge running and both sensors active.

Check messages are landing in the topic:

docker exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic sensor-events \
  --from-beginning \
  --max-messages 5

You will see two distinct event shapes arriving. The door sensor emits only on state change:

{"device_id": "door-sensor-01", "ingested_at": "2026-06-08T11:23:45.123456+00:00", "payload": {"contact": false, "battery": 97, "tamper": false, "linkquality": 231}}

The smart plug emits continuously every ~10 seconds:

{"device_id": "smart-plug-01", "ingested_at": "2026-06-08T11:23:47.891011+00:00", "payload": {"power": 84.3, "voltage": 230.1, "current": 0.366, "energy": 1.204, "state": "ON", "linkquality": 248}}

Check partition distribution:

docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --topic sensor-events

With two devices keyed by device_id, messages distribute across 2 of the 3 partitions. Which two depends on the murmur2 hash of each key — run it and see. The third partition stays empty. That is expected.

Confirm the key-to-partition mapping:

docker exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic sensor-events \
  --property "print.key=true" \
  --property "print.partition=true" \
  --from-beginning

You will see output like Partition:0 door-sensor-01 {...} — the same key always lands on the same partition.

Check consumer group lag (relevant once Spark is consuming in Post 5):

docker exec kafka /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --group spark-sensor-consumer

---

What __consumer_offsets is actually doing

When Spark Structured Streaming reads from Kafka, it commits its progress — the last offset read per partition — back to Kafka in an internal topic called __consumer_offsets. This is Kafka’s native offset storage mechanism, introduced in Kafka 0.9 to replace offset storage in Zookeeper.

Each committed offset is a key-value record: (group_id, topic, partition) → offset. Kafka compacts this topic aggressively so only the latest offset per key is retained.

This matters for understanding exactly-once semantics in Post 5. Spark Structured Streaming checkpoints its offsets independently of Kafka’s __consumer_offsets topic — it writes them to the checkpoint directory on Delta Lake storage. The Kafka consumer group offset is a secondary record used by kafka-consumer-groups.sh to show lag. The authoritative offset from Spark’s perspective is in the checkpoint. If the two disagree after a crash, Spark wins.

That is worth sitting with for a moment: two offset stores, two sources of truth, and Spark ignores Kafka’s version. The checkpoint directory is what matters. Post 5 will open that directory and show you exactly what is inside it.

---

Memory budget check

Before Post 5 adds Spark to the stack, it is worth tracking where the Pi’s 4GB is going:

Service	Heap / RAM	Notes
Kafka	512MB JVM heap	Set via KAFKA_HEAP_OPTS
Mosquitto	~10MB	Negligible
MinIO	~256MB	Object storage
OS + Docker overhead	~400MB
Available for Spark	~2.8GB	Enough for 512MB executor heap + overhead

Spark’s JVM heap will be capped at 512MB in Post 5. The total footprint of the full stack should stay under 3.5GB, leaving the Pi comfortable.

---

What comes next

Kafka is running, the topic exists, and real sensor events are flowing — sparse contact events from the SNZB-04P, steady power readings from the S60ZBTPF. Post 5 connects Spark Structured Streaming to this topic, writes to Delta Lake on MinIO, and gets into the parts of this series that justify the whole exercise: checkpoint internals, exactly-once semantics, and what Delta’s transaction log actually does when Spark crashes mid-batch.

---

Previous: Post 3 — Zigbee2MQTT + Mosquitto: pairing sensors, understanding MQTT payloads Next: Post 5 — Spark Structured Streaming + Delta Lake: checkpoints, exactly-once, late data