Apache Kafka Internals

Storage internals, replication protocol, and the operational playbook for running Kafka for real — segments and indexes, the producer/consumer wire paths, ISR replication, compaction, exactly-once semantics, and the lag/rebalance debugging patterns you reach for during an incident. See kafka-field-guide.md for the concept-first tour this guide sits underneath.

Track how many of the knowledge checks below you've cleared as you go:

Storage: Log Segments

Kafka stores each partition as an append-only log on disk, divided into segment files.

graph TD
    subgraph "Partition 0 on disk"
        SEG0["00000000000000000000.log<br/>messages offset 0-999"]
        IDX0["00000000000000000000.index<br/>offset --> file position"]
        SEG1["00000000000000001000.log<br/>messages offset 1000-1999"]
        IDX1["00000000000000001000.index<br/>sparse index"]
        ACTIVE["00000000000000002000.log<br/>ACTIVE segment<br/>new writes go here"]
    end
    PROD["Producer<br/>append to active"] --> ACTIVE
    CONS["Consumer<br/>seek to offset, read sequentially"] --> SEG0 & SEG1 & ACTIVE

Segment rolling: When active segment reaches log.segment.bytes (default 1GB) or log.roll.ms (default 7 days), it's closed and a new one starts.

The index file: Sparse index mapping offsets to byte positions. Consumer seeks to an offset → binary search in index → seek to file position → read forward. O(log n) seek, then O(1) sequential read.

Step through what a seek to an arbitrary offset actually does:

Producer Write Path

sequenceDiagram
    participant PROD as Producer
    participant LEADER as Partition Leader (broker-1)
    participant ISR1 as ISR Replica (broker-2)
    participant ISR2 as ISR Replica (broker-3)

    PROD->>LEADER: ProduceRequest (acks=all, messages)
    LEADER->>LEADER: Append to local log segment
    LEADER->>ISR1: Replicate (async)
    LEADER->>ISR2: Replicate (async)
    ISR1-->>LEADER: Fetch offset acknowledged
    ISR2-->>LEADER: Fetch offset acknowledged
    Note over LEADER: All ISR replicas caught up
    LEADER-->>PROD: ProduceResponse (offset=1234)
    Note over PROD: Write committed (acks=all)

acks settings:

Consumer Groups and Offset Management

graph TD
    TOPIC["Topic: orders<br/>6 partitions"] --> CG["Consumer Group: payments"]
    subgraph CG["Consumer Group: payments (3 consumers)"]
        C1["Consumer-1<br/>assigned: P0, P1"]
        C2["Consumer-2<br/>assigned: P2, P3"]
        C3["Consumer-3<br/>assigned: P4, P5"]
    end

    C1 & C2 & C3 -->|"commit offsets"| OFFSET_TOPIC["__consumer_offsets topic<br/>stores: group+topic+partition --> offset"]

Offset commit strategies:

// Auto commit (default, at-least-once risk)
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "5000");

// Manual commit after processing (at-least-once, safer)
consumer.poll(Duration.ofMillis(100));
// ... process records ...
consumer.commitSync();   // block until broker confirms

// Exactly-once: commit offset in same DB transaction as business logic
// (transactional outbox pattern)

Offset reset policy:

Replication — ISR Deep Dive

graph TD
    LEADER2["Partition Leader<br/>HW = 1005 (High Watermark)"] --> ISR_A["ISR: broker-2<br/>LEO = 1007 (Log End Offset)"]
    LEADER2 --> ISR_B["ISR: broker-3<br/>LEO = 1005"]
    LEADER2 --> OUT_ISR["OUT of ISR: broker-4<br/>LEO = 950 (too far behind)<br/>replica.lag.time.max.ms exceeded"]

    CONS2["Consumer<br/>can only read up to HW=1005<br/>not uncommitted messages 1006-1007"]

High Watermark (HW): The offset up to which ALL ISR replicas have the data. Consumers can only read up to HW. Messages above HW are uncommitted — might be lost if leader crashes.

Leo (Log End Offset): Latest offset written to the log, may be ahead of HW.

ISR shrink/expand:

Try It Yourself: Live ISR Simulator

Same idea as the diagram above, but live: one partition, 1 leader + 2 followers. Produce records, stall a follower to simulate it falling behind, and watch exactly when it gets dropped from the ISR — and what acks=all is actually blocked on at each step, versus acks=1. Non-stalled followers catch up to the leader instantly here (a simplification of real async replication lag) — the part worth watching closely is what happens to a stalled one.

Cluster Metadata and the Controller: KRaft (KIP-500)

Everything above this point — ISR membership, which broker leads which partition, who's allowed to do what — is a decision someone has to track cluster-wide, not per-partition. That someone is the controller: exactly one broker (pre-KRaft) or a small elected leader (KRaft) responsible for topic/partition metadata, leader assignment, broker liveness, and ACLs. If it goes away, the remaining nodes elect a new one and the cluster keeps serving data in the meantime. The interesting question this section answers: where does that metadata actually live, and what does "elect a new one" actually mean under the hood — because as of Kafka 3.x (production-ready) and 4.0 (default, ZooKeeper support removed entirely), the answer changed completely.

Why ZooKeeper got removed. For most of Kafka's life, none of that metadata lived inside Kafka at all — it lived in Apache ZooKeeper, a general-purpose coordination service Kafka delegated to. That meant every Kafka deployment was actually two distributed systems stacked on top of each other: the Kafka cluster you wanted, and a ZooKeeper ensemble underneath it whose sole job was being Kafka's filing cabinet for topics, partitions, ACLs, and controller election. ZooKeeper brought its own cluster to provision and patch, its own quorum-sizing rules, its own session-timeout tuning, and its own failure modes — a lost ZK quorum could stall Kafka's entire control plane (no new leader elections, no topic changes) even while every broker was up and healthy serving reads and writes. Running Kafka well meant a team had to also run ZooKeeper well, as a second, unrelated skill set. KIP-500 removed that second system.

KRaft's approach. Instead of delegating metadata to an outside system, Kafka now stores it the same way it stores everything else it's good at storing: as a log. Every metadata change — a topic created, a partition's leader changing, an ACL granted — is appended as a record to an internal topic, __cluster_metadata, and that topic is replicated using an actual Raft implementation among a small set of nodes running the controller role. Which physical nodes play that role is a config choice, not a fixed topology — the process.roles setting on each node is broker, controller, or both:

graph TD
    subgraph Small["Small cluster -- roles combined"]
        S1["Node 1<br/>process.roles=broker,controller"]
        S2["Node 2<br/>process.roles=broker,controller"]
        S3["Node 3<br/>process.roles=broker,controller"]
        S1 --> S2
        S2 --> S3
        S3 --> S1
    end

    subgraph Large["Larger cluster -- roles split"]
        C1["Controller 1 -- current Raft leader<br/>process.roles=controller"]
        C2["Controller 2<br/>process.roles=controller"]
        C3["Controller 3<br/>process.roles=controller"]
        B1["Broker 1<br/>process.roles=broker"]
        B2["Broker N<br/>process.roles=broker"]
        C1 --> C2
        C1 --> C3
        B1 -->|"fetch __cluster_metadata"| C1
        B2 -->|"fetch __cluster_metadata"| C1
    end

Small clusters typically combine both roles on the same handful of nodes (cheaper, fewer processes to run). Larger clusters split them — a dedicated 3- or 5-node controller quorum handling only metadata, separate from the brokers serving produce/consume traffic, so metadata load never competes with data load on the same process.

Leader election for that metadata quorum is no longer "ZooKeeper handles it as a black box" — it's genuine Raft: terms, log offsets, majority votes, the same mechanics already covered in replication.md § 9, Consensus Algorithms (including a live leader-election demo). That's the identical algorithm, not a loose analogy — it's just electing a leader for the __cluster_metadata log instead of a general-purpose replicated log, so there's no need to re-derive term numbers or vote-counting here.

What's operationally different for a cluster admin:

• One system to run and monitor, not two. No separate ZooKeeper ensemble to size, patch, upgrade, and page on — controller state is just another Kafka log, observed with the same tooling as everything else in the cluster.
• Faster controller failover. ZooKeeper-based failover was bottlenecked by ZK session timeouts before a dead controller's session even expired and a new election could start. Raft's own election timeout drives failover directly instead — no second system's timeout sitting in the critical path.
• Metadata durability rides on Kafka's own replication. __cluster_metadata is replicated and made durable the same way any other Kafka log is, instead of depending on a separate consensus implementation (ZooKeeper's) with its own semantics and its own bugs to reason about.

Log Compaction

graph LR
    subgraph Before["Before compaction (key:value log)"]
        M1["offset=0: user:1 --> {name:Alice}"]
        M2["offset=1: user:2 --> {name:Bob}"]
        M3["offset=2: user:1 --> {name:ALICE}"]
        M4["offset=3: user:3 --> {name:Charlie}"]
        M5["offset=4: user:2 --> null (tombstone = delete)"]
    end

    subgraph After["After compaction"]
        K1["offset=2: user:1 --> {name:ALICE} (latest)"]
        K2["offset=3: user:3 --> {name:Charlie}"]
        Note["user:2 deleted (tombstone + old value removed)"]
    end

Log compaction retains the latest value per key — turns Kafka into a changelog/event store for materialized views. Used by Kafka Streams and ksqlDB.

log.cleanup.policy=compact          # enable compaction
log.cleanup.policy=compact,delete   # compact AND delete old segments
min.cleanable.dirty.ratio=0.5       # compact when 50% of log is dirty

Step through a compaction pass on the log shown above:

Try It Yourself: Live Log Compaction

The stepper above narrates one fixed, scripted pass. This one is live: append your own keyed records (the same key can land at many offsets — that's the whole point), tombstone a key to mark it for deletion, then run Compact and watch only the highest offset per key survive.

Exactly-Once Semantics

graph LR
    PROD2["Producer<br/>enable.idempotence=true<br/>transactional.id=tx-1"] -->|"ProducerID + sequence number<br/>broker deduplicates"| BROKER["Kafka Broker<br/>dedup by ProducerID+Seq"]
    BROKER -->|"transaction: atomic multi-partition write"| P1["Partition A"]
    BROKER --> P2["Partition B"]
    P1 & P2 -->|"consumer reads isolation.level=read_committed"| CONS3["Consumer<br/>only sees committed transactions"]

Idempotent producer: Each message tagged with ProducerID + sequence number. Broker rejects duplicates (retry after network failure = same message, not duplicate).

Transactions: Write to multiple partitions atomically. Either all committed or none visible.

Key Metrics

# Consumer lag (most important — alert > 10000)
kafka_consumer_group_lag > 10000

# Under-replicated partitions (alert > 0)
kafka_server_replication_under_replicated_partitions > 0

# ISR shrink rate (alert if frequent)
rate(kafka_server_replication_isr_shrinks_total[5m]) > 0

# Producer request latency p99
histogram_quantile(0.99, kafka_network_request_total_time_ms_bucket{request="Produce"}) > 100

Schema Registry

Avro/Protobuf schemas are stored in the Schema Registry. Producers serialize with schema ID; consumers look up the schema to deserialize. Prevents incompatible schema changes breaking consumers.

graph LR
    PROD2["Producer"] -->|"register/lookup schema"| SR["Schema Registry"]
    PROD2 -->|"[magic:1B][schema_id:4B][avro_bytes]"| BROKER2["Kafka Broker"]
    CONS2["Consumer"] -->|"lookup schema by ID"| SR
    BROKER2 -->|"raw bytes"| CONS2
    CONS2 -->|"deserialize with schema"| DATA2["Typed object"]

# Producer with schema registry
from confluent_kafka.avro import AvroProducer
producer = AvroProducer(
    {'bootstrap.servers': 'kafka:9092', 'schema.registry.url': 'http://registry:8081'},
    default_value_schema=avro.loads(value_schema_str)
)
producer.produce(topic='orders', value={'id': '123', 'amount': 99.99})

Schema compatibility modes:

Kafka Transactions (Exactly-Once)

sequenceDiagram
    participant APP as Application
    participant BROKER as Kafka Broker
    participant OFFSET_TOPIC as __consumer_offsets

    APP->>BROKER: initTransactions()
    APP->>BROKER: beginTransaction()
    APP->>BROKER: produce(orders, message1)
    APP->>BROKER: produce(analytics, message2)
    APP->>OFFSET_TOPIC: sendOffsetsToTransaction(group, offsets)
    APP->>BROKER: commitTransaction()
    Note over BROKER: All messages + offset commit atomic
    Note over BROKER: Consumers with isolation.level=read_committed<br/>only see committed messages

producer.initTransactions();
producer.beginTransaction();
try {
    producer.send(new ProducerRecord<>("orders", key, value));
    producer.sendOffsetsToTransaction(offsets, groupMetadata);
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

Walk through the happy path plus the abort branch:

Kafka Streams

Kafka Streams is a Java library for stream processing — stateless transformations, aggregations, joins — all backed by Kafka topics.

StreamsBuilder builder = new StreamsBuilder();

// Read from topic
KStream<String, Order> orders = builder.stream("orders");

// Stateless: filter + transform
KStream<String, Order> paidOrders = orders
    .filter((key, order) -> order.getStatus().equals("paid"))
    .mapValues(order -> enrichOrder(order));

// Stateful: count per user (stored in RocksDB state store)
KTable<String, Long> orderCounts = orders
    .groupByKey()
    .count(Materialized.as("order-counts-store"));

// Write to output topic
paidOrders.to("paid-orders");
orderCounts.toStream().to("order-counts");

State stores (RocksDB) are backed by changelog topics — on restart, the state is rebuilt from the changelog without reprocessing all input.

Consumer Lag Alerting

# Alert: consumer group is falling behind (lag > 10K messages)
kafka_consumer_group_lag{group="payments", topic="orders"} > 10000

# Calculate processing rate needed to catch up
# current_lag / (consume_rate - produce_rate) = time to catch up

# Alert: no consumer is running for a group (lag growing without consumption)
increase(kafka_consumer_group_lag[5m]) > 0
AND
kafka_consumer_group_members{group="payments"} == 0

# Real-time lag monitoring
kafka-consumer-groups.sh \
  --bootstrap-server kafka:9092 \
  --describe --group payments
# Watch lag column — should trend toward 0 for healthy consumer

# Kafka UI tools: Kafdrop, Redpanda Console, Conduktor

Topic Sizing and Retention

# View topic configuration
kafka-configs.sh --bootstrap-server kafka:9092 \
  --describe --entity-type topics --entity-name orders

# Override retention for a specific topic
kafka-configs.sh --bootstrap-server kafka:9092 \
  --alter --entity-type topics --entity-name orders \
  --add-config retention.ms=604800000  # 7 days
            # retention.bytes=10737418240  # 10GB

# Estimate disk usage
# disk_per_partition = (produce_rate_bytes/s × retention_seconds) / num_partitions
# Total disk = disk_per_partition × total_partitions × replication_factor

Debugging

# Describe a topic (partitions, replicas, ISR)
kafka-topics.sh --bootstrap-server kafka:9092 --describe --topic orders
# Partition: 0  Leader: 1  Replicas: 1,2,3  Isr: 1,2,3
# If Isr != Replicas: a replica is behind → investigate

# Read messages from beginning
kafka-console-consumer.sh \
  --bootstrap-server kafka:9092 \
  --topic orders --from-beginning --max-messages 10

# Check broker log dirs (find large partitions)
kafka-log-dirs.sh --bootstrap-server kafka:9092 \
  --broker-list 1,2,3 --topic-list orders

# Preferred replica election (rebalance leaders back after failure)
kafka-leader-election.sh --bootstrap-server kafka:9092 \
  --election-type PREFERRED --all-topic-partitions

Consumer Lag Deep-Dive

Consumer lag = log-end-offset - committed-offset. It tells you how far behind a consumer group is from the head of the partition.

Reading lag correctly

# View lag per partition for a consumer group
kafka-consumer-groups.sh \
  --bootstrap-server kafka:9092 \
  --describe \
  --group payments-processor

# Output:
# GROUP               TOPIC      PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID
# payments-processor  payments   0          45000           45100           100  consumer-1
# payments-processor  payments   1          44900           45050           150  consumer-2
# payments-processor  payments   2          44800           46000          1200  consumer-3 ← spike

# Total lag = sum of all partition lags = 1450
# Partition 2 has 10x the lag of others → partition imbalance

Why per-partition lag matters: a consumer group may show low average lag while one partition is 10,000 messages behind. Average lag hides the worst case. Always look at max lag per partition.

Lag alert with Prometheus (Kafka Exporter)

# kafka-exporter exposes: kafka_consumergroup_lag{consumergroup, topic, partition}
- alert: KafkaConsumerLagHigh
  expr: |
    sum(kafka_consumergroup_lag{consumergroup="payments-processor"}) by (consumergroup, topic) > 10000
  for: 5m
  labels:
    severity: warning

- alert: KafkaConsumerLagCritical
  expr: |
    max(kafka_consumergroup_lag{consumergroup="payments-processor"}) by (partition) > 50000
  for: 2m
  labels:
    severity: critical
  annotations:
    summary: "Single partition lag > 50k — consumer likely dead or partition hot"

Root causes and fixes

Producer tuning for throughput vs durability

# High throughput (analytics, logs) — batch more, weaker guarantees
acks=1                     # leader ACK only (not all replicas)
batch.size=65536           # 64KB batch (default 16KB)
linger.ms=10               # wait 10ms to fill batch before sending
compression.type=lz4       # compress batches (lz4 best CPU/ratio tradeoff)
buffer.memory=67108864     # 64MB producer buffer
max.in.flight.requests.per.connection=5

# High durability (payments, orders) — ensure no data loss
acks=all                   # all ISR replicas must ACK
retries=2147483647         # retry forever (Java MAX_INT)
max.in.flight.requests.per.connection=1   # prevent message reordering on retry
enable.idempotence=true    # exactly-once on producer side
delivery.timeout.ms=120000 # 2 minutes total retry window

Partition strategy — choosing partition count

Partition count determines max consumer parallelism.
More partitions → more parallelism + more overhead (open file handles, leader elections).

Rule of thumb:
  target_throughput_MB/s  ÷  throughput_per_partition_MB/s = partitions needed

Single partition throughput (approximate):
  Producer:  ~50-100 MB/s (disk sequential write speed)
  Consumer:  ~50-100 MB/s (network + processing bound in practice)

Example:
  Need 500 MB/s total throughput
  Each partition handles ~50 MB/s
  → 10 partitions minimum

For consumer parallelism:
  max_consumers_in_group = partition_count
  If you have 20 consumer instances, you need ≥20 partitions
  Extra consumers beyond partition count sit idle

# Add partitions to an existing topic (can only increase, never decrease)
kafka-topics.sh \
  --bootstrap-server kafka:9092 \
  --alter \
  --topic payments \
  --partitions 20
# WARNING: adding partitions changes key→partition mapping for new messages.
# Old messages stay on old partitions. Consumers must handle reordering.
# For strict ordering by key: pre-plan partition count at topic creation.

Rebalance debugging

Consumer rebalances (triggered by member join/leave/timeout) pause ALL consumers in the group while a new partition assignment is computed.

# Check rebalance frequency
kafka-consumer-groups.sh \
  --bootstrap-server kafka:9092 \
  --describe --group payments-processor
# CONSUMER-ID changes on rebalance

# Common causes of excessive rebalancing:
# 1. max.poll.interval.ms too low — consumer takes longer to process than allowed
#    Fix: increase max.poll.interval.ms or reduce max.poll.records
# 2. session.timeout.ms too low — consumer heartbeat misses under GC pause
#    Fix: increase session.timeout.ms (but lag detection slower)
# 3. Rolling restart — each pod restart triggers two rebalances (leave + rejoin)
#    Fix: use static group membership

# Static group membership — survive restarts without rebalance
group.instance.id=payments-consumer-0   # unique, stable ID per consumer instance
session.timeout.ms=60000                # how long before a static member is considered dead
# With static membership, restarts within session.timeout.ms skip rebalance

Try It Yourself: Live Partition Assignment (Range Assignor)

The mermaid diagram earlier in this file already showed the shape of it — 6 partitions split into contiguous ranges across consumers. This is that assignment live: 6 fixed partitions (P0–P5), starting with a 2-consumer group. Add or remove a consumer by name to trigger a rebalance and watch the Range assignor — Kafka's default — recompute the whole assignment from scratch every time.