Producers, consumers and partitions on a real broker

#Why it matters

Streaming appeared in about a third of Software and Data Engineer job descriptions we read. Payment events, order updates, clickstreams and IoT readings flow through Kafka or a managed equivalent at most large companies. Interviewers ask how partitions, keys and consumer groups work, because those decide ordering, scale and correctness.

#The concept

• A topic is a named, append-only log of events, kept for a configured time.

• A topic is split into partitions. Order is guaranteed only within a partition.

• Producers send events with an optional key; events with the same key go to the same partition, so all events for one zone, customer or account stay in order.

• Consumers in the same consumer group share the partitions: each partition is read by one member at a time. Add members to scale out, up to the number of partitions.

• Each group tracks its position per partition as an offset. Committing an offset records "processed up to here", so a restarted consumer continues where it left off.

#Worked example: taxi trips as events

Download Kafka 4 from the Apache site, then start a single node in KRaft mode (no ZooKeeper needed):

tar xzf kafka_2.13-4.3.1.tgz && cd kafka_2.13-4.3.1
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
bin/kafka-storage.sh format --standalone -t "$KAFKA_CLUSTER_ID" -c config/server.properties
bin/kafka-server-start.sh config/server.properties            # leave running in its own terminal

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic trips --partitions 3 --replication-factor 1

Producer: replay the 8 to 9 a.m. hour of 15 January 2025, keyed by pickup zone (pip install confluent-kafka duckdb):

"""Replay one hour of real NYC taxi trips as Kafka events, keyed by pickup zone."""
import json

import duckdb
from confluent_kafka import Producer

producer = Producer({"bootstrap.servers": "localhost:9092", "enable.idempotence": True, "acks": "all"})
trips = duckdb.sql("""
    select tpep_pickup_datetime::varchar as pickup_at, PULocationID as zone, fare_amount
    from 'yellow_tripdata_2025-01.parquet'
    where tpep_pickup_datetime >= '2025-01-15 08:00' and tpep_pickup_datetime < '2025-01-15 09:00'
    order by tpep_pickup_datetime
""").fetchall()

delivered = {"ok": 0, "failed": 0}

def on_delivery(error, message):
    delivered["failed" if error else "ok"] += 1

for pickup_at, zone, fare in trips:
    event = {"pickup_at": pickup_at, "zone": zone, "fare": fare}
    producer.produce("trips", key=str(zone), value=json.dumps(event), on_delivery=on_delivery)
    producer.poll(0)                     # serve delivery callbacks while producing
producer.flush()
print(f"sent {len(trips)} events: {delivered}")

Consumer: count trips per zone, committing offsets only after processing each event:

"""Count trips per pickup zone from the stream, committing offsets only after processing."""
import json
from collections import Counter

from confluent_kafka import Consumer

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "zone-counter",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
})
consumer.subscribe(["trips"])

counts, partitions, idle_polls = Counter(), Counter(), 0
while idle_polls < 10:                                 # demo only: stop after 10 quiet seconds
    message = consumer.poll(1.0)                       # the first polls also join the consumer group
    if message is None:
        idle_polls += 1
        continue
    if message.error():
        raise RuntimeError(message.error())
    idle_polls = 0
    event = json.loads(message.value())
    counts[event["zone"]] += 1
    partitions[message.partition()] += 1
    consumer.commit(message=message, asynchronous=False)   # at-least-once: commit after processing

consumer.close()
print("events per partition:", dict(sorted(partitions.items())))
print("top pickup zones:", counts.most_common(3))

sent 7129 events: {'ok': 7129, 'failed': 0}
events per partition: {0: 2570, 1: 2866, 2: 1693}
top pickup zones: [(236, 503), (237, 342), (186, 296)]

The busiest morning pickups are the Upper East Side (zones 236 and 237) and Penn Station (186). Partitions are uneven because a few zones carry many events and every event for a zone lands in one partition: that is the price of per-key ordering.

Run the consumer again with the same group and it reads nothing: its offsets were committed. kafka-consumer-groups.sh --describe --group zone-counter shows a lag of 0 on every partition. Change group.id and a new group reads the topic from the beginning, independently.

#Common mistakes

• Expecting global order. Kafka orders within a partition only. Choose keys by what must stay in order.

• More consumers than partitions. Extra members sit idle. Plan partition counts for future scale.

• Auto-commit before processing. A crash after commit but before processing loses events. Commit after.

• Huge messages. Kafka is built for many small events; put large payloads in object storage and send a reference.

#Interview angle

"How does Kafka scale consumption, and how is ordering guaranteed?" Explain partitions, keys and consumer groups, and the trade-off between ordering and parallelism (hot keys create uneven partitions, as in the taxi example). Mention offsets and what happens when a consumer restarts.

#Practice task

1. Start two consumers in the same group at once and watch the partitions being shared between them.

2. Compute revenue per zone per 10-minute window from pickup_at.

3. Change the key to the pickup minute and compare the partition balance.

#Further reading

• Apache Kafka quickstart

• Apache Kafka documentation: design

• confluent-kafka Python client