Kafka makes it easy to move events and hard to process them correctly. Most production incidents we are asked to fix come down to five misunderstandings: ordering is per partition, not per topic; "exactly-once" has a boundary; rebalances happen at the worst possible time; one bad message can stall a partition; and schemas evolve whether you plan for it or not. This post goes through each, with the configuration and code we use.

What guarantees does Kafka actually give?

A topic is split into partitions. Each partition is an ordered, append-only log replicated across brokers, and every record has an offset within its partition. That sentence contains most of Kafka's semantics:

  • Ordering is guaranteed only within a partition. Two records in different partitions have no defined order.
  • Consumers in a group divide the partitions between them. Each partition is read by exactly one consumer in the group at a time, so parallelism is capped by the partition count.
  • Consumers track progress by committing offsets. What "processed" means depends entirely on when you commit.
  • Retention is time- or size-based, or the topic is compacted, which keeps the latest record per key. Compacted topics make good changelogs and lookup state.

Since Kafka 4.0 the cluster runs in KRaft mode only; ZooKeeper is gone. Brokers keep metadata in an internal Raft-replicated log, which simplifies operations and speeds up controller failover.

How should you choose keys and partitions?

The producer picks a partition from the record key. The Java client hashes the key with murmur2, modulo the partition count, so all events with the same key land in the same partition, in order. Choose the key as the entity whose events must stay ordered: account_id, order_id, device_id.

Things that bite:

  • Changing the partition count remaps keys. After adding partitions, a key's new events can land in a different partition from its old ones, breaking per-key ordering during the transition. Size partitions up front for peak throughput and consumer parallelism. Partition counts can be increased but never decreased.
  • Hot keys. One huge tenant or one null-ish key ("unknown") overloads one partition while others idle. Detect skew from per-partition throughput, and fix it at the key design, for example by salting keys for events that do not need strict order.
  • Mixed client libraries partition differently. librdkafka-based clients (Python's confluent-kafka, Go, .NET, Node) default to a CRC32-based partitioner, not Java's murmur2. If services in different languages write the same keys to the same topic, set partitioner=murmur2_random on the librdkafka side, or the same key goes to different partitions.
  • Null keys are spread across partitions in batches (the "sticky" partitioner). That is good for throughput, but there is no ordering at all.

How do we configure producers so data is not lost?

acks=all                                 # wait for all in-sync replicas
enable.idempotence=true                  # broker de-duplicates producer retries
max.in.flight.requests.per.connection=5  # <= 5 keeps ordering with idempotence
retries=2147483647                       # retry until delivery.timeout.ms
delivery.timeout.ms=120000               # total time budget for a send
linger.ms=10                             # small wait to build bigger batches
compression.type=zstd                    # or lz4; better throughput and storage

Pair it with topic and broker settings: replication factor 3 and min.insync.replicas=2. A write is then acknowledged only when at least two replicas have it, and the cluster tolerates the loss of one broker without losing acknowledged data. Do not enable unclean leader election on topics you care about.

Current versions of the Java client default to acks=all with idempotence enabled. librdkafka-based clients do not enable idempotence by default, so set it explicitly. Idempotence gives each producer an ID and per-partition sequence numbers, and the broker discards duplicates caused by retries. It removes duplicates from retries within one producer session, not duplicates your application creates by sending twice.

At-least-once, at-most-once and exactly-once: where are the boundaries?

A consumer's delivery semantics follow from when it commits:

  • Commit before processing: at-most-once. A crash loses the in-flight records.
  • Commit after processing: at-least-once. A crash replays records processed since the last commit. This is the right default, if processing is idempotent.
  • Exactly-once: Kafka transactions make a read–process–write cycle within Kafka atomic. Consume from topic A, produce to topic B and commit the consumer offsets, all in one transaction.
from confluent_kafka import Consumer, KafkaException, Producer

consumer = Consumer({
    "bootstrap.servers": BROKERS,
    "group.id": "order-enricher",
    "enable.auto.commit": False,
    "isolation.level": "read_committed",   # never read aborted transactional data
    "auto.offset.reset": "earliest",
})
producer = Producer({
    "bootstrap.servers": BROKERS,
    "transactional.id": f"order-enricher-{INSTANCE_ID}",  # stable per instance across restarts
    "enable.idempotence": True,
    "acks": "all",
})

producer.init_transactions()          # also fences off any zombie with the same transactional.id
consumer.subscribe(["orders"])

while True:
    batch = consumer.consume(num_messages=500, timeout=1.0)
    if not batch:
        continue
    producer.begin_transaction()
    try:
        for msg in batch:
            if msg.error():
                raise KafkaException(msg.error())
            producer.produce("orders-enriched", key=msg.key(), value=enrich(msg.value()))
        # offsets are committed as part of the same transaction as the output records
        producer.send_offsets_to_transaction(
            consumer.position(consumer.assignment()),
            consumer.consumer_group_metadata(),
        )
        producer.commit_transaction()
    except Exception:                    # simplified: fatal errors (e.g. fenced producer) need a restart
        producer.abort_transaction()
        rewind_to_committed(consumer)    # seek each partition back to its last committed offset

Downstream consumers must use isolation.level=read_committed. They then see only committed transactions, and never read past the last stable offset of an open transaction.

The boundary. Kafka's exactly-once covers Kafka-to-Kafka. The moment processing writes to a database, calls an API or sends an email, those side effects are outside the transaction. Exactly-once effects then require one of:

  1. An idempotent sink. Upsert by a natural key or event ID, so replays overwrite instead of duplicating. Our CDC post shows sequence-guarded MERGEs for exactly this.
  2. Deduplication by event ID, against a store that is updated in the same transaction as the side effect (the inbox pattern).
  3. A transactional or batch-idempotent sink that ties output to the consumed offsets. Spark's checkpoints with Delta sinks achieve this effect: offsets are recorded per batch and the Delta write for a batch is idempotent.

How do you publish events from a database without dual writes?

Writing to the database and then producing to Kafka is a dual write: a crash between the two loses or orphans an event. The transactional outbox pattern fixes it. The service writes the business change and an outbox row in the same database transaction, and a CDC connector such as Debezium streams the outbox table to Kafka. The event is published only if the business transaction committed. Delivery is at-least-once, so consumers deduplicate by event ID.

Why do rebalances hurt, and how do you tame them?

When consumers join, leave or miss heartbeats, the group rebalances its partitions. Under the classic eager protocol, every consumer gives up all its partitions and processing stops across the group until assignment completes. Long processing loops make it worse: if poll() is not called within max.poll.interval.ms, the consumer is evicted, which triggers yet another rebalance.

Mitigations, roughly in order of impact:

  • Use the new consumer group protocol (KIP-848), generally available since Kafka 4.0. Set group.protocol=consumer on clients that support it. Assignment moves to the broker and becomes incremental, so there is no global stop-the-world.
  • On the classic protocol, use the cooperative sticky assignor, which revokes only the partitions that actually move.
  • Enable static membership (group.instance.id) so a restarting pod reclaims its partitions without a rebalance, within session.timeout.ms.
  • Keep per-poll work bounded (max.poll.records), or hand slow work to a worker pool and commit only completed offsets.
  • On partition revocation, flush in-progress work and commit, so the next owner does not redo it.

How should consumers handle failures?

Classify every error:

  • Transient (timeouts, throttling, a dependency briefly down): retry with exponential backoff and jitter, within a bounded budget.
  • Permanent (unparseable payload, schema violation, business-rule rejection): do not retry. Send it to a dead-letter topic with the original payload, and headers recording the error, source topic, partition, offset and timestamp.

Blocking a partition on one poison message stalls every key behind it. Tiered retry topics (for example orders.retry.1m, orders.retry.10m, then orders.dlq) keep the main partition flowing. The catch is that retry topics break per-key ordering: a retried event can be overtaken by a newer one for the same key. When ordering matters, either make the consumer order-tolerant with sequence guards in the sink, or park the key: once one event for a key is in retry, route that key's later events behind it until it clears.

A DLQ is a process, not a bin. Alert on DLQ volume, give it an owner, and build a replay tool that re-publishes fixed messages to the main topic.

How do you evolve event schemas safely?

Use a schema registry with Avro, Protobuf or JSON Schema, and enforce a compatibility mode per subject:

Mode Guarantee Deploy order
BACKWARD (common default) New schema can read data written with the previous one Upgrade consumers first
FORWARD Previous schema can read data written with the new one Upgrade producers first
FULL Both directions Any order
*_TRANSITIVE Checked against all previous versions, not just the last As above

Safe changes: add optional fields with defaults, and remove fields that have defaults. Breaking changes (renaming, changing types, reusing Protobuf field numbers) need a new topic or a versioned event type, plus a migration period.

Design events to be self-describing: an event_id (UUID) for deduplication, event_type and schema_version, the entity key, and both event time (when it happened) and ingest time. Prefer events that state facts ("PaymentCaptured", with the amount) over commands or partial deltas that consumers cannot interpret alone.

Processing Kafka streams with Spark Structured Streaming

For lakehouse pipelines we usually consume Kafka with Spark Structured Streaming on Databricks, landing raw events in a bronze Delta table and building silver and gold from there.

from pyspark.sql import functions as F, types as T

schema = T.StructType([
    T.StructField("event_id", T.StringType()),
    T.StructField("merchant_id", T.StringType()),
    T.StructField("amount", T.DecimalType(18, 2)),
    T.StructField("event_time", T.TimestampType()),
])

raw = (spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", BROKERS)
    .option("subscribe", "payments")
    .option("startingOffsets", "earliest")
    .option("maxOffsetsPerTrigger", 2_000_000)      # bound each micro-batch
    .load())

events = (raw
    .select(F.from_json(F.col("value").cast("string"), schema).alias("e"),
            F.col("partition"), F.col("offset"))
    .select("e.*", "partition", "offset")
    .withWatermark("event_time", "30 minutes")      # accept events up to 30 min late
    .dropDuplicatesWithinWatermark(["event_id"]))   # Spark 3.5+: bounded-state de-duplication

per_merchant = (events
    .groupBy(F.window("event_time", "5 minutes"), "merchant_id")
    .agg(F.count("*").alias("txn_count"), F.sum("amount").alias("amount")))

(per_merchant.writeStream
    .outputMode("append")      # emit each window once, after the watermark passes it
    .option("checkpointLocation", "/Volumes/prod/checkpoints/merchant_5min")
    .toTable("gold.merchant_activity_5min"))

How it behaves:

  • Offsets live in the checkpoint, not in a Kafka consumer group. Spark records the offsets of each micro-batch and commits them with the sink's output. With Delta sinks, a restart neither loses nor double-writes data. Do not share or delete checkpoint directories casually: they are your consumer position.
  • Watermarks bound state. Without a watermark, aggregations and deduplication keep state forever. The watermark is a trade-off between completeness (later events accepted) and latency and state size.
  • Large state belongs in RocksDB. Set spark.sql.streaming.stateStore.providerClass to the RocksDB state store provider (on Databricks, the Databricks RocksDB provider), so state is not held on the JVM heap.
  • The Kafka sink is at-least-once. When writing back to Kafka from Spark, downstream consumers must deduplicate.

When do we use something else? Kafka Streams suits JVM microservices doing per-event, low-latency transformations with local state. Apache Flink suits sub-second, large-state, event-time-heavy processing. Spark fits best when the stream feeds the lakehouse and shares code and governance with batch.

What should you monitor?

  • Consumer lag per partition, in records and in time. Lag growing on one partition usually means a hot key or a poison message.
  • Under-replicated and offline partitions, ISR shrink and expand rates, and controller health.
  • Producer error and retry rates, request latency, and batch size.
  • DLQ volume and age, plus rebalance frequency and duration per consumer group.
  • Disk and retention. Tiered storage (production-ready since Kafka 3.9) moves older segments to object storage, so long retention no longer means huge broker disks.

A production checklist

  • Keys chosen for the ordering you need; partitions sized for peak, with headroom
  • RF = 3, min.insync.replicas = 2, acks=all, idempotence on in every client
  • Offsets committed after processing; processing and sinks idempotent
  • Transactions with read_committed where Kafka-to-Kafka exactly-once is required
  • Outbox pattern instead of dual writes
  • KIP-848 or cooperative-sticky rebalancing, plus static membership
  • Retry topics, a DLQ with an owner and a replay tool
  • Schema registry with enforced compatibility; events carry ID, key and event time
  • Lag, replication and DLQ alerts wired to on-call