Streaming with Kafka

The log

Kafka is not a queue that deletes messages once read. It is a log: an append-only sequence of records, each with an offset, retained for a configured time or size regardless of who has read it. Producers append; consumers read from any offset and track their own position. Several consumers can read the same data independently, a new consumer can start from the beginning and replay history, and a crashed consumer resumes from where it left off. This decoupling — producers do not know or wait for consumers — is why a log works as the backbone between systems that evolve separately.

A topic is a named log, split into partitions for parallelism and scale. Within a partition, records are strictly ordered; across partitions there is no ordering guarantee. Records have an optional key; records with the same key always go to the same partition, so all events for one customer, one order, one device arrive in order to whoever reads that partition. Choosing the key is choosing what must be ordered together, and it decides how evenly load spreads: a key with a few hot values (country) skews partitions; a key with many values (customer id) balances them.

kafka-topics.sh --bootstrap-server broker:9092 --create --topic orders \
  --partitions 12 --replication-factor 3 \
  --config retention.ms=604800000 --config min.insync.replicas=2

# produce keyed records (key:value with a separator)
printf 'cust-1042:{"order_id":9001,"amount":42.5}\ncust-77:{"order_id":9002,"amount":9.99}\n' \
  | kafka-console-producer.sh --bootstrap-server broker:9092 --topic orders \
      --property parse.key=true --property key.separator=:

# consume from the beginning, showing partition and offset
kafka-console-consumer.sh --bootstrap-server broker:9092 --topic orders --from-beginning \
  --property print.partition=true --property print.offset=true --property print.key=true --max-messages 2

Consumers, groups and offsets

A consumer group is a set of consumers sharing a group id; Kafka assigns each partition of the subscribed topics to exactly one consumer in the group, so the group processes the topic in parallel with each partition read once. Add a consumer and partitions rebalance; a consumer dies and its partitions move to survivors. Each group tracks its committed offset per partition: the position it will resume from after a restart. Independent groups (the fraud service and the analytics loader) each maintain their own offsets over the same topic, reading at their own speed.

from confluent_kafka import Consumer, KafkaException

consumer = Consumer({
    "bootstrap.servers": "broker:9092",
    "group.id": "orders-loader",
    "enable.auto.commit": False,          # we commit explicitly, after work is done
    "auto.offset.reset": "earliest",      # a brand-new group starts from the beginning
    "max.poll.interval.ms": 300000,
})
consumer.subscribe(["orders"])

try:
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            raise KafkaException(msg.error())
        process(msg.key(), msg.value())        # idempotent: upsert by order_id
        consumer.commit(message=msg, asynchronous=False)   # commit this offset + 1
finally:
    consumer.close()

Consumer lag — committed offset versus the latest offset per partition — is the health metric of streaming: growing lag means consumers cannot keep up, are stuck, or have died. Monitor it per group and alert on it. Slow processing inside poll loops causes the broker to consider the consumer dead and rebalance, which makes lag worse; keep per-record work fast, batch the slow parts, and raise max.poll.interval.ms only with a reason.

Delivery guarantees, honestly

There are three achievable semantics. At-most-once: commit before processing; a crash loses the in-flight record; acceptable for metrics where a gap is tolerable. At-least-once: commit after processing; a crash reprocesses the record; the default for anything that matters, and it requires idempotent processing (upsert by key, deduplicate by event id) so replays are harmless. Exactly-once semantics (EOS): Kafka's idempotent producer prevents duplicate appends from producer retries, and transactions let a consume-transform-produce loop commit its output records and its input offsets atomically, so a Kafka-to-Kafka pipeline neither loses nor duplicates. EOS stops at Kafka's edge: writing to an external database exactly once still needs idempotent writes or a transactional sink.

from confluent_kafka import Producer

producer = Producer({
    "bootstrap.servers": "broker:9092",
    "acks": "all",                 # leader + in-sync replicas must persist it
    "enable.idempotence": True,    # retries cannot create duplicates
    "linger.ms": 20,               # small batching for throughput
    "compression.type": "zstd",
})


def on_delivery(err, msg):
    if err is not None:
        log_failed_event(msg.key(), err)   # dead-letter or alert; never ignore


producer.produce("orders", key="cust-1042", value=payload_bytes, on_delivery=on_delivery)
producer.flush(timeout=10)

Schemas and evolution

Bytes on a topic have no schema unless you give them one, and a producer that changes its JSON shape breaks every consumer at once. A schema registry stores versioned schemas (Avro, Protobuf or JSON Schema) per topic; producers serialise with a registered schema id embedded in the record, consumers fetch the schema by id and deserialise correctly even across versions. The registry enforces a compatibility mode: backward compatibility (new schema can read old data: add optional fields, remove fields with defaults) is the usual choice so consumers can upgrade after producers. This is the data contract from the quality module, enforced at the moment of writing.

{
  "type": "record",
  "name": "OrderPlaced",
  "namespace": "acme.shop",
  "fields": [
    {"name": "order_id", "type": "long"},
    {"name": "customer_id", "type": "long"},
    {"name": "ordered_at", "type": {"type": "long", "logicalType": "timestamp-millis"}},
    {"name": "amount", "type": {"type": "bytes", "logicalType": "decimal", "precision": 12, "scale": 2}},
    {"name": "status", "type": {"type": "enum", "name": "Status", "symbols": ["PLACED", "PAID", "SHIPPED", "REFUNDED"]}},
    {"name": "channel", "type": ["null", "string"], "default": null}
  ]
}

Design events as facts about what happened (OrderPlaced, PaymentCaptured), carrying the identifiers and values a consumer needs, with a unique event id and the event time as a field. Include a version or rely on the registry's; never put a secret or a full card number in an event, because a log retains it for the retention period across every replica and every consumer's copy.

Stream processing: windows, time and state

Reading one record at a time is easy; most useful computations involve many: revenue per minute, distinct users per hour, joining a click stream with a purchase stream, detecting three failed logins in five minutes. Stream processors (Kafka Streams, Apache Flink, Spark Structured Streaming) manage the state these need and the hard problem behind them: time. Events carry an event time (when it happened) and arrive at a processing time (when you saw it), and they arrive late and out of order. Windows must be computed on event time, which means deciding how long to wait for stragglers: a watermark declares "I assume no events older than T will still arrive", closes windows before T, and routes anything later to a side output or a correction.

from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StructType, StructField, LongType, StringType, DoubleType, TimestampType

spark = SparkSession.builder.appName("revenue_per_minute").getOrCreate()

schema = StructType([
    StructField("order_id", LongType()), StructField("country", StringType()),
    StructField("amount", DoubleType()), StructField("ordered_at", TimestampType()),
])

orders = (
    spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "orders")
    .option("startingOffsets", "latest")
    .load()
    .select(F.from_json(F.col("value").cast("string"), schema).alias("o")).select("o.*")
)

per_minute = (
    orders
    .withWatermark("ordered_at", "10 minutes")                       # tolerate 10 min of lateness
    .groupBy(F.window("ordered_at", "1 minute"), "country")
    .agg(F.sum("amount").alias("revenue"), F.count("*").alias("orders"))
)

query = (
    per_minute.writeStream
    .outputMode("append")                                           # emit a window once it is final
    .format("parquet")
    .option("path", "s3://acme-lake/gold/revenue_per_minute/")
    .option("checkpointLocation", "s3://acme-lake/checkpoints/revenue_per_minute/")
    .trigger(processingTime="30 seconds")
    .start()
)
query.awaitTermination()

Window types: tumbling (fixed, non-overlapping: per minute), sliding (overlapping: last five minutes, every minute) and session (gap-based: a user's visit). Stream-stream joins need state bounded by time (join clicks to purchases within an hour); stream-table joins enrich events with a slowly changing reference. All of it needs checkpointing so state and offsets survive restarts, and it needs the same idempotent sinks as batch, because a restart replays from the last checkpoint.