📨 Partitions & Offsets

Partition Anatomy

A partition is a distributed, replicated, append-only log. Understanding the anatomy is essential for reasoning about ordering guarantees, replication, and data loss scenarios.

Replication internals: AR, ISR, LEO, HW

Every partition has a leader and zero or more followers. Followers consume from the leader to stay in sync. The key metadata structures:

How replication works (the fetch cycle)

Producer writes to leader (broker-0):
  Record { offset=5, value="payment-123" }

Leader appends to local log, LEO: 5 → 6
Leader waits for follower acks (depends on acks config):
  acks=1    → immediate success (leader only)
  acks=all  → wait for all ISR to confirm write

Follower (broker-1) fetch loop (runs every replica.fetch.wait.max.ms=500ms):
  → Leader: give me records from offset=5
  ← Leader: [offset=5, offset=6, ...]
  Follower appends to local log, updates LEO
  Follower sends ack to leader with its new LEO

Leader on receiving ISR ack:
  if (follower.LEO >= current_HW) {
    advance HW to min(all_ISR_LEO)
  }

End result: HW=5, data at offset 0-4 is committed (won't be lost if leader fails).
            Data at offset 5-7 is uncommitted (leader has it, followers are catching up).

If leader crashes now, broker-1 is the new leader (it was in ISR and has HW=5).
Offsets 5-7 are lost (producer will retry based on its retry config).

Data loss scenario: unclean leader election

By default, if all replicas for a partition are out of ISR, Kafka will wait unclean.leader.election.enable=false (default) and the partition goes offline. If you enable unclean.leader.election.enable=true, a replica that is out of sync can become leader — meaning offset 5-7 (uncommitted on the old leader) could be lost or could overlap with data the new leader has. This is why acks=all matters for data durability: it ensures data is committed to ISR before the producer considers the write successful.

Partition Assignment

When a topic is created, Kafka distributes its partitions across the broker fleet. The assignment algorithm considers rack awareness (for fault tolerance), broker capacity, and (optionally) custom placement constraints.

Rack awareness: not just for AWS

# Tell Kafka about your topology so it doesn't put all replicas in one rack
# broker.rack property in server.properties on each broker

# Example: 6 brokers across 3 racks (2 brokers per rack)
broker.rack=us-east-1a   # broker-0, broker-1
broker.rack=us-east-1b    # broker-2, broker-3
broker.rack=us-east-1c    # broker-4, broker-5

# Replication factor 3 with rack awareness:
# Kafka ensures replicas are spread across at least 2 different racks.
# If rack us-east-1a loses power, you still have 2/3 replicas alive.

# Verify rack placement:
$ kafka-topics.sh --describe --topic payments --bootstrap-server broker:9092
Topic: payments  PartitionCount: 12   ReplicationFactor: 3
Topic: payments  Partition: 0    Leader: 2       Replicas: 2,4,0  Isr: 2,4,0
Topic: payments  Partition: 1    Leader: 4       Replicas: 4,0,2  Isr: 4,0,2
...
# Replicas: [2,4,0] — three different brokers, likely three different racks.
# If broker 2 dies, leader moves to broker 4 (both still on different racks).

Sticky partitioner: the default since 2.4

Before Kafka 2.4, when a producer sent messages with a new key (or no key), it used round-robin across all partitions. This caused high churn in the partition mapping and excessive records in-flight. The sticky partitioner batches records with the same partition until the batch is full or a timeout expires, then moves to a new partition. This reduces latency by ~30-40% in testing.

ProducerRecord("order-1", ...)  → partition 0
ProducerRecord("order-2", ...)  → partition 1
ProducerRecord("order-3", ...)  → partition 2
ProducerRecord("order-4", ...)  → partition 0
# Every send goes to a different partition.
# Every batch has only one record → high overhead.

ProducerRecord("order-1", ...)  → partition 0 (batch fills with 16 records)
ProducerRecord("order-2", ...)  → partition 0
...
ProducerRecord("order-16", ...)  → partition 0
ProducerRecord("order-17", ...)  → partition 1 (batch fills)
# 16 sends in one batch → one Produce request for 16 records.
# Then move to next partition. Reduces RPC overhead dramatically.

Custom partitioner example

import org.apache.kafka.clients.producer.*;

public class RegionPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                        Object value, byte[] valueBytes,
                        Cluster cluster) {
        // Route by region suffix on the key
        String keyStr = key.toString();
        String region = keyStr.substring(keyStr.lastIndexOf('-') + 1);

        // Count partitions per prefix
        List partitions = cluster.partitionsForTopic(topic);
        int regionPartition = switch (region) {
            case "us-east" -> 0;
            case "us-west" -> partitions.size() / 3;
            case "eu"      -> (partitions.size() * 2) / 3;
            default        -> key.hashCode() % partitions.size();
        };
        return Math.abs(regionPartition) % partitions.size();
    }

    @Override
    public void close() {}
}

// Usage:
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, RegionPartitioner.class);

Partition reassignment

# Move partitions to balance the cluster
$ kafka-reassign-partitions.sh --generate-then-execute

# Generate a reassignment plan (moves partitions from heavy brokers to light ones)
$ kafka-reassign-partitions.sh --bootstrap-server broker:9092 \
  --generate --topics-to-move-json-file topics.json \
  --broker-list "0,1,2,3,4,5" \
  --output-path reassignment.json

# Review before executing
$ cat reassignment.json
{"partitions":[{"topic":"payments","partition":0,
  "replicas":[2,4,0],"log_dirs":["any","any","any"]},...]}

# Execute the reassignment (runs in background, throttled by default)
$ kafka-reassign-partitions.sh --bootstrap-server broker:9092 \
  --execute --reassignment-json-file reassignment.json

# Monitor progress
$ kafka-reassign-partitions.sh --bootstrap-server broker:9092 \
  --verify

# Throttle to avoid saturating network (add to reassignment.json):
{"partitions":[...],"properties":{"throttle": "100MB"}}  # max 100 MB/s

Preferred replica election

Normally, the first replica in the replicas list is the "preferred replica" — the one that should be leader when the cluster is balanced. Over time, broker restarts and failures cause leadership to drift to other brokers. The preferred replica election tries to restore the ideal assignment.

# Auto-enable preferred replica election (runs every 10 minutes by default)
# In server.properties:
auto.leader.imbalance.check.interval.seconds=300
auto.leader.imbalance.per.broker.percentage=10
# If a broker has >10% of leadership vs its share, trigger election.

# Manual preferred replica election:
$ kafka-leader-election.sh --bootstrap-server broker:9092 \
  --path-to-json-file election.json \
  --all

# election.json:
{"partitions":[{"topic":"payments","partition":0}]}

Partition Count: How Many Do You Need?

Partition count is the most consequential topic configuration. It sets the ceiling on parallelism — both for producers and consumers — and it's effectively permanent for key-ordered topics. Getting it wrong in either direction costs you.

The formula

partitions_needed = max(
    target_producer_throughput_MBs / single_producer_throughput_MBs,
    target_consumer_throughput_MBs / single_consumer_throughput_MBs
)

# Example:
# Target producer throughput: 400 MB/s
# Single producer (one partition, one thread): ~50 MB/s
# → producer side needs: ceil(400 / 50) = 8 partitions

# Target consumer throughput: 800 MB/s
# Single consumer thread: ~50 MB/s
# → consumer side needs: ceil(800 / 50) = 16 partitions

# Use the higher: 16 partitions

# Add headroom for broker failures:
partitions_needed = base_partitions * (1 / (1 - max_failover_ratio))
# If you want to survive losing 1 broker with no lag spike:
# 16 * (1 / (1 - 0.1)) ≈ 18 partitions (10% headroom)

Tradeoffs: More partitions

Real-world partition counts by use case

Segment Files: The On-Disk Structure

Each partition is a directory. Inside it, messages are stored in segment files — contiguous chunks of the log. Kafka never writes to one giant file; it rolls to a new segment based on size or time. Understanding segments is essential for sizing brokers and tuning retention.

# Inspect segment files directly
$ ls -la /data/kafka/my-topic-0/
-rw-r--r-- 1 kafka kafka  1.0G  May 19 10:00  00000000000000000000.log
-rw-r--r-- 1 kafka kafka   10M  May 19 10:00  00000000000000000000.index
-rw-r--r-- 1 kafka kafka    8M  May 19 10:00  00000000000000000000.timeindex
-rw-r--r-- 1 kafka kafka  1.0G  May 18 22:00  00000000001000000000.log
-rw-r--r-- 1 kafka kafka   10M  May 18 22:00  00000000001000000000.index
-rw-r--r-- 1 kafka kafka    8M  May 18 22:00  00000000001000000000.timeindex

# Segment naming: base offset of the first message in the segment
# 00000000001000000000.log starts at offset 1 billion

# Segment config (topic-level):
# segment.bytes = 1GB (default: 1GB, max reasonable: 2GB)
# segment.ms = 7 days (default: 168h = 7 days)
# index.interval.bytes = 4096 (default: 4KB — one index entry every 4KB of data)

# When does a segment roll?
# 1. size-based:  .log file reaches segment.bytes (1GB)
# 2. time-based:  segment.ms elapses (7 days)
# 3. both conditions checked on each append (so size hits first = rolls immediately)

# The .log file format (on disk):
# [offset 4B][crc 4B][magic 1B][attributes 1B][key length 4B][key N bytes]
#         [value length 4B][value N bytes][timestamp 8B][headers...]
# Records are variable-length; no padding between them.

# The .index file: sparse, offset-position mapping
# Every index.interval.bytes (4KB by default) of log data → one index entry:
#   relative_offset (4 bytes) + physical_position (4 bytes)
# "relative offset" saves space: 4 bytes instead of 8 for the absolute offset.
#   relative = absolute - segment_base_offset
#
# .timeindex: timestamp → relative_offset mapping (for timestamp searches)

# Lookup flow when reading offset X from segment with base_offset B:
#   1. binary search .index for largest relative_offset <= X - B
#   2. get physical_position from .index entry
#   3. scan .log from physical_position to find offset X
# Sparse index means we may scan a few records after the index position.

Retention and compaction

# Topic retention config
# Keep data for 7 days OR 50GB per partition, whichever comes first
retention.ms = 604800000        # 7 days in ms
retention.bytes = 53687091200   # 50GB

# After retention expires, segment files are deleted
# Check what's using disk:
$ kafka-log-dirs.sh --describe --bootstrap-server broker:9092 \
  --topic-list my-topic 2>&1 | grep -E "(partition|size|offset)"

# Log compaction: key-value topics only
# Keep the latest value for each key (used for changelog topics, KSQL materialized views)
cleanup.policy = compact   # vs. delete (default)
# Compaction runs continuously in the background.
# A message with null value = tombstone, removes the key after cleanup.

# View segment files marked for deletion after retention:
$ ls /data/kafka/my-topic-0/*.deleted  # old segments awaiting cleanup

Offsets: How Consumers Track Position

An offset is a monotonically increasing integer that marks a consumer's position in a partition. The consumer group stores its committed offset in the __consumer_offsets topic — a special compacted internal topic. Getting offset commit semantics right is the difference between "no data loss" and "we processed that twice."

Offset commit strategies

// Strategy 1: Auto-commit (enable.auto.commit = true, default)
// Commits the last offset every auto.commit.interval.ms (5 seconds)
// ❌ At-least-once risk: if you crash after processing but before commit,
//   the re-process is from the last committed offset (you skip nothing).
// ❌ You can't control exactly when offsets are committed.
// ✅ Simplicity. Good for non-critical monitoring/event pipelines.

auto.commit.interval.ms = 5000

// Strategy 2: Manual sync commit (enable.auto.commit = false)
// You call consumer.commitSync() or consumer.commitAsync() explicitly.
// ✅ You commit after successfully processing, guaranteeing at-least-once.
// ❌ Must handle commit failures and retry.

while (true) {
    ConsumerRecords records = consumer.poll(Duration.ofMillis(200));
    for (ConsumerRecord record : records) {
        process(record);  // do work
    }
    try {
        consumer.commitSync();  // commit after processing batch
    } catch (CommitFailedException e) {
        log.error("offset commit failed", e);
        // handle: maybe partitions rebalanced and we lost ownership
    }
}

// Strategy 3: Async commit with callback
consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        log.error("commit failed for {}", offsets, exception);
        // retry later, but don't block current processing
    }
});

// Strategy 4: Manual offset management (store externally)
// Useful for exactly-once semantics in transactional systems.
Map offsets = new HashMap<>();
for (ConsumerRecord record : records) {
    offsets.put(
        new TopicPartition(record.topic(), record.partition()),
        new OffsetAndMetadata(record.offset() + 1, record.metadata())
    );
    process(record);
}
consumer.commitSync(offsets);  // commit the offset AFTER processing the record

__consumer_offsets: the internal topic

# __consumer_offsets topic anatomy
# 50 partitions, replication factor 1 by default (⚠ change for production!)
# Key = [consumer_group_id, topic, partition] → Value = committed_offset

# View consumer group offsets:
$ kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --group my-app --describe

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG     OWNER
my-app          payments        0          4821            5103             282     consumer-1
my-app          payments        1          4390            4390               0     consumer-1
my-app          payments        2          5011            5011               0     consumer-1

# Lag = how far behind the slowest consumer is
# Lag > 0 means consumers are behind producers — monitor this.

# Reset offset (careful!):
$ kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --group my-app --topic payments:0 \
  --reset-offset --to-earliest --execute

# Offset reset strategies:
# --to-earliest, --to-latest, --shift-by N, --to-offset N, --by-duration N

# What __consumer_offsets stores (simplified):
# Key:   group-id="my-app", topic="payments", partition=0
# Value: offset=4821, leader_epoch=14, metadata="", timestamp=now
# This gets compacted: only latest value per key survives.

Rebalancing: when partitions change hands

When a consumer joins, leaves, or dies, the group coordinator triggers a rebalance: partitions are reassigned among active consumers. During rebalancing, all consumers stop processing — and any uncommitted offsets at the time of the last commit will cause reprocessing.

# Rebalance triggers:
# - Consumer joins/leaves group (including crash detection via session.timeout)
# - Topic partition count changes
# - Consumer group leader reassigns partitions (static membership helps)

# Join process:
# 1. Any consumer sends JoinGroup request to coordinator
# 2. Coordinator picks one as leader (first to join)
# 3. Leader receives full member list, runs partition assignment algorithm
# 4. Leader sends SyncGroup with assignment to coordinator
# 5. Coordinator forwards assignment to all members
# 6. Members start consuming

# Common issue: long rebalance due to slow consumer
# session.timeout.ms = 10s (default) — if consumer doesn't poll in 10s, kicked out
# max.poll.interval.ms = 5min — gap between polls before rebalance triggered
# max.poll.records = 500 — fewer records = faster poll = less likely to timeout

# Static membership (reduces rebalances):
# Set group.instance.id to a stable consumer id. On restart, consumer
# gets same partitions without full rebalance.
partition.assignment.strategy = cooperative.sticky  # Kafka 2.4+

Exactly-once: is it possible?

There are three semantics: at-least-once (default, may duplicate), at-most-once (may lose), exactly-once (neither duplicates nor loses). Kafka supports exactly-once within the Kafka ecosystem via transactions, but end-to-end exactly-once requires producer idempotency + consumer transactions + an idempotent sink.

// Kafka exactly-once: transactions API
// Requires: transactional.id (producer) + enable.idempotence = true

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(record1);            // writes to output topic
    producer.sendOffsetsToTransaction(
        offsets,  // map of TopicPartition → offset
        new Metadata(groupInstanceId)    // consumer's group metadata
    );
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

// This guarantees: if producer crashes mid-transaction, all offsets
// and records roll back. No duplicate output, no missing output.
//
// The trick: consumer stores its offset in the same transaction as the
// output record. Either both commit or both abort. Consumer reads its
// own committed offset = exactly-once semantics.
//
// End-to-end exactly-once from source DB to sink DB:
// 1. Source: Debezium CDC → Kafka (exactly-once via transaction ID)
// 2. Sink: Kafka Connect with exactly-once (idempotent insert by primary key)
// If the sink is NOT idempotent, you still need transactional outbox pattern.

The Controller: One Broker Owns All Partition Leadership

The controller is a single broker that manages cluster-wide partition state: leader elections, ISR changes, broker failure detection, and partition reassignment. The controller was originally ZooKeeper-dependent; in KRaft mode (Kafka 3.3+), ZooKeeper is eliminated. But the logic remains the same.

Controller election (ZooKeeper mode)

# ZooKeeper stores /controller epoch
# First broker to create /controller (ephemeral znode) becomes controller
# If it dies, the znode disappears → next broker to create it becomes controller

# Watch /controller to know when leadership changes:
$ zookeeper-shell broker:2181
> get /controller
{"version":1,"brokerid":0,"timestamp":"1620912345000","kafka_version":"2.8.0"}

# When broker-0 dies:
# 1. ZooKeeper detects session expiry, deletes /controller
# 2. broker-1 and broker-2 both try to create /controller
# 3. broker-1 wins (first write wins in ZooKeeper)
# 4. broker-1 becomes controller
# 5. broker-1 sends LeaderAndIsr request to all brokers
# 6. broker-1 becomes leader for all partitions where it was in ISR

# Controller responsibilities:
# - Track live brokers, partition count, replication factor per topic
# - Detect broker failures (via ZooKeeper session)
# - When broker dies: reassign partition leadership to ISR
# - When broker rejoins: trigger preferred replica election
# - Handle topic creation/deletion metadata updates
# - Manage isr.expiry.time (default: 10s) — when a replica is considered dead

# KRaft mode (Kafka 3.3+, ZooKeeper-less):
# Controller is a Kafka process itself, part of a Raft quorum.
# Controller quorum: 3 brokers form a Raft group.
# No ZooKeeper dependency. Controller failover within ~200ms.

Partition leader election on broker failure

# Scenario: broker-0 (leader for partition 0) dies
# Controller (broker-1) detects via ZooKeeper session timeout

# Controller runs leader election:
# 1. Get ISR for partition 0: [broker-1, broker-2] (both were followers)
# 2. Pick new leader: first broker in ISR list (broker-1)
# 3. Controller sends LeaderAndIsr to broker-1 and broker-2:
#    - broker-1: "you are now leader of partition 0"
#    - broker-2: "your new leader is broker-1, start fetching from offset N"
# 4. Update partition state in ZooKeeper
# 5. Send UpdateMetadata to all brokers (so producers know new leader)

# Total time: ~5-10 ms for the election itself
# Time to detect broker dead: session.timeout.ms (default 10s on broker side,
# but controller uses replica.lag.time.max.ms = 10s to exclude from ISR)

# After broker-0 comes back:
# - It rejoins as a follower for its partitions
# - Preferred replica election restores original leadership over time
# - auto.leader.rebalance.enable=true (default) triggers it

Offline partitions: the worst case

# If all replicas for a partition are out of ISR:
# - unclean.leader.election.enable=false (default) → partition goes offline
# - unclean.leader.election.enable=true → replica with highest LEO becomes leader

# Identify offline partitions:
$ kafka-topics.sh --bootstrap-server broker:9092 \
  --list --under-min-isr
# Or check via JMX:
# kafka.server:type=ReplicaManager,name=UnderMinIsrPartitionCount

# Fix offline partitions:
# 1. Bring the dead broker back online
# 2. If broker is unrecoverable: reassign replicas
# 3. Increase replication.factor for the topic (future-proof)
# 4. Set unclean.leader.election.enable=true only as a last resort

# Prevention:
replication.factor = 3          # at least 3
min.insync.replicas = 2         # need 2 for producer to succeed
unclean.leader.election.enable = false  # never elect out-of-sync leader

Partitioned Producer Patterns

The producer's partitioning strategy determines which partition each record lands in. This has downstream consequences for ordering, parallelism, and consumer group behavior. Getting it right at design time avoids painful schema migrations later.

Key-based partitioning for ordering

// Default partitioner: murmur2 hash of key modulo partition count
// Records with the same key ALWAYS go to the same partition.
// This guarantees ordering within that key's stream.

ProducerRecord record = new ProducerRecord<>(
    "payments",           // topic
    "user-1234",          // key — all of user-1234's events go to one partition
    "{\"amount\":99.99}"  // value
);

// murmur2 hash: fast, well-distributed, non-cryptographic
// hash(key) % partition_count = partition
// ⚠ Partition count change breaks this mapping → key redistribution
// Never change partition count on a key-ordered topic.

props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
                  StringSerializer.class.getName());
props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
                  StringSerializer.class.getName());
props.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG, 
                  DefaultPartitioner.class.getName());  // sticky by default

// What if key is null?
// DefaultPartitioner: sticky partitioner picks a random partition and fills
// it with all null-key records until batch is full.
// This is efficient for throughput but breaks ordering across null-key records.

// For null keys with round-robin: use UniformStickyPartitioner explicitly,
// or use a round-robin partitioner (older behavior).

Custom partitioning for multi-tenant systems

// Multi-tenant: route by tenant_id, then by key within tenant
public class TenantAwarePartitioner implements Partitioner {

    private int partitionsPerTenant;

    @Override
    public void configure(Map configs) {
        // e.g., 10 partitions per tenant, 100 tenants = 1000 total partitions
        this.partitionsPerTenant = 
            Integer.parseInt(configs.get("partitions.per.tenant").toString());
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                        Object value, byte[] valueBytes, Cluster cluster) {
        String keyStr = key.toString();
        String[] parts = keyStr.split("/", 2);  // "tenant-7/user-1234"
        String tenantId = parts[0];
        String entityKey = parts.length > 1 ? parts[1] : keyStr;

        int tenantHash = Math.abs(tenantId.hashCode());
        int basePartition = (tenantHash % 10) * partitionsPerTenant;

        // Within tenant, route by entity key for ordering
        int entityHash = Math.abs(entityKey.hashCode());
        return (basePartition + (entityHash % partitionsPerTenant));
    }

    @Override
    public void close() {}
}

// Usage:
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, 
          TenantAwarePartitioner.class.getName());
props.put("partitions.per.tenant", "10");

Idempotent producer: avoiding duplicates at scale

// Idempotent producer: Kafka deduplicates in-flight messages automatically
// Each producer has a PID (producer ID), and each message has a sequence number.
// If the same sequence number arrives twice, the second is silently dropped.

props.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
// Idempotency is free (no performance cost at scale), enable it in production.
// Requires: acks=all (can't be idempotent with acks=1)

props.setProperty(ProducerConfig.ACKS_CONFIG, "all");
props.setProperty(ProducerConfig.RETRIES_CONFIG, "Integer.MAX_VALUE");
props.setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5");
// With idempotence, in-flight reqs > 1 can cause duplicates on retries
// (broker can receive seq 1, then seq 2, then retry seq 1 → duplicates).
// Kafka 2.5+ handles this with idempotent sequence tracking, so 5 is fine.

producer = new KafkaProducer<>(props);
// Every retry of the same message gets the same sequence number.
// Kafka tracks last successful sequence per (PID, partition) and dedupes.
// Exactly-once semantics within Kafka is now free with idempotence + transactions.

Ordered Delivery: When Partitioning Breaks Things

Kafka guarantees ordering within a partition, not across partitions. If you need total order across all records in a topic, you must have exactly one partition. If you partition by a key, you get ordering per key, not global ordering.

The single-partition trap

# Problem: global ordering with one partition
# Throughput ceiling: single consumer can do ~50 MB/s
# With gzip compression, maybe 20-30 MB/s sustained
# For a high-volume order system, this becomes a bottleneck.

# Solution: partition by order_id, not by user_id
# Key insight: "all events for order-1234 must go to the same partition"
# If user-42 has 3 orders, their events go to 3 different partitions —
# but within each order's stream, they're correctly ordered.

# Wrong partitioning for ordering:
# producer.send("orders", user_id, order_event)
# user-42's order-a and order-b go to different partitions → interleaved
# order-a's "create" happens after order-b's "update" on a different partition

# Correct partitioning:
# producer.send("orders", order_id, order_event)
# All events for order-a → same partition (in order: create, update, fulfill, complete)
# All events for order-b → same partition (in order: create, update, cancel)
# Within order-a's stream: events are correctly ordered
# Order-b's stream is completely independent — no interleaving

# When you NEED a single partition:
# - Financial transactions where global sequence matters
# - Audit logs that must be strictly ordered across all events
# - Systems with external ordering constraints (legacy DB sequences)
# Cost: limited throughput, no horizontal scaling of consumers

Caution: partition count change breaks key ordering

# Scenario: You partition by user_id with 12 partitions.
# hash(user_id) % 12 = partition
# Now you need more throughput → increase to 24 partitions.
# hash(user_id) % 24 = different partitions for many keys.
# All those users' events are now split across two partitions.
# Consumer ordering: user-42's create on partition 3, update on partition 15.
# Application sees update before create → business logic bug.

# Solutions:
# 1. Never change partition count on key-ordered topics (plan ahead)
# 2. Use a composite key: version the partitioning scheme in the key itself
#    e.g., key = v1:user-42 (when you change partitioning, use v2:...)
#    Or: include a partition scheme version byte in the key
# 3. Migrate data to new topic with new partition count (expensive)
# 4. Use a partitioner that is stable across partition count changes
#    (impossible with hash modulo — this is fundamental)

# Best practice: add partition count to your key schema
# Key: "tenant-42|entity-99|v2" (version the partitioner logic in the key)
# When you change partitioner logic, increment version in new records.
# Consumers read both v1 and v2, routing is backward compatible.

Out-of-order consumption: detection and mitigation

# Detecting out-of-order offsets in consumer:
# When consuming, detect gaps in the offset sequence:

Map endOffsets = consumer.endOffsets(
    consumer.assignment().stream()
        .map(p -> new TopicPartition(p.topic(), p.partition()))
        .collect(Collectors.toList())
);

for (ConsumerRecord record : records) {
    long expectedOffset = lastOffsetSeen.get(record.partition()) + 1;
    if (record.offset() != expectedOffset) {
        log.warn("Gap detected in partition {}: expected {}, got {}",
                 record.partition(), expectedOffset, record.offset());
        // Gap means: either messages were deleted (retention expired)
        // or there's a rebalance timing issue
    }
    lastOffsetSeen.put(record.partition(), record.offset());
}

# Mitigating out-of-order delivery:
# 1. Use consumer.seek() to reset to last committed offset on rebalance
# 2. Process records in offset order explicitly
# 3. For time-series data: filter out late-arriving records by timestamp
# 4. Stateful consumers: track sequence numbers per key, discard stale ones

Split-Brain and Broker Failure Scenarios

Kafka's leader-based replication prevents split-brain by design: only the leader accepts writes, and only in-sync replicas can become leader. But network partitions can still create scenarios where some brokers think others are dead while they're actually alive. Understanding these failure modes helps you tune timeouts and replication settings correctly.

Network partition scenario: step by step

# Scenario: network partition splits the cluster in two
# DC-A (brokers 0, 1) can't talk to DC-B (brokers 2, 3)

# What Kafka does (with acks=all, minISR=2, unclean.leader.election=false):

# Partition 0: replicas=[0,2,3], ISR=[0,2]
# - DC-A: broker-0 (leader) alive, broker-2 (ISR) alive
# - DC-B: broker-3 alive but can't reach broker-0
# - Result: broker-0 continues as leader, writes succeed (acks from 0,2)
# - broker-3 falls out of ISR after replica.lag.time.max.ms=10s

# Partition 1: replicas=[1,2,3], ISR=[2,3]
# - DC-A: broker-1 (leader) can't reach ISR brokers 2,3
# - DC-B: brokers 2,3 see broker-1 as dead (session.timeout)
# - broker-1's ISR shrinks to [1] (only itself)
# - If minISR=2: no writes succeed (ISR too small)
# - Controller (broker-0) triggers leader election for partition 1
# - broker-2 or broker-3 becomes new leader
# - broker-1 in DC-A goes offline (unclean.leader.election=false)

# Key insight: no split-brain because only the controller (broker-0)
# can authorize leader elections. DC-A can't elect a new leader for partition 1
# without contacting the controller. DC-B's controller (broker-0) is alive and
# in DC-A, so it coordinates the election. Kafka's split-brain prevention
# relies on ZooKeeper's sequential consistency — one broker writes the /controller
# znode, and all partition reassignments go through it.

Split-brain risk: when it can still happen

# Misconfiguration that causes split-brain or data loss:

# 1. Unclean leader election enabled (DANGEROUS)
unclean.leader.election.enable = true
# Risk: out-of-sync replica becomes leader → some messages lost, some duplicated
# Never enable this for topics where durability matters (payments, orders)

# 2. minISR=1 (dangerous)
min.insync.replicas = 1
# Risk: only the leader must acknowledge. If leader crashes after write but before
# replication, data is lost. acks=1 is equivalent to this.
# For durability: use minISR=2 with acks=all.

# 3. session.timeout too high (slow failure detection)
# Broker failure detection by controller uses:
# - replica.lag.time.max.ms (excluded from ISR after 10s default)
# - Broker's own session.timeout (controlled via broker config)
# If session.timeout=30s, controller takes 30s to notice broker death.

# 4. No rack awareness (replicas in same rack)
# If all replicas are in the same rack and the rack loses power:
# RF=3 with all replicas in same rack = all replicas die simultaneously.
# Partition goes offline even with RF=3.

# 5. Overloaded controller broker
# Controller is single-threaded for partition state changes.
# If controller broker is overloaded, leader elections back up.
# High partition count + heavy cluster = consider dedicated controller broker.

Monitoring: The Metrics That Matter

Partition health lives in a handful of JMX metrics. The key ones to watch are under-replicated partitions (URP), offline partitions, and consumer lag. Alert on these before they become user-visible problems.

Critical metrics

# Under-replicated partitions (URP)
# Alert when > 0 for sustained > 5 minutes
kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions
# A partition is under-replicated when: ISR < replication.factor
# Causes: follower is lagging, follower is dead, network issue
# Impact: data durability risk — if the remaining ISR broker dies, data is lost

# Offline partitions (worst case)
kafka.server:type=ReplicaManager,name=OfflinePartitionsCount
# Alert: > 0 for any period
# Causes: all replicas out of ISR, unclean.leader.election=false
# Impact: topic is completely unavailable for that partition
# Fix: restore dead brokers, reassign replicas, or enable unclean election

# Consumer lag (end-to-end latency)
kafka.consumer:type=consumer-fetch-manager-metrics,client-id=...,topic=...,partition=...
  attribute=records-lag-max
kafka.consumer:type=consumer-coordinator-metrics,client-id=...
  attribute=committed-offset-lag

# Alert threshold: lag > (max.poll.records * 5) or lag growing > 10% over 10 min
# Lag spike: consumers can't keep up with producers
# Causes: slow consumer (GC pause, blocked on downstream), too few partitions,
#         or consumer group has fewer instances than expected

# Producer metrics
kafka.producer:type=producer-metrics,client-id=...
  record-error-rate          # messages failed to send
  record-send-rate           # throughput
  request-latency-avg        # p99 round-trip time to broker

# Broker metrics
kafka.server:type=ReplicaManager,name=IsrExpandsRate
kafka.server:type=ReplicaManager,name=IsrShrinksRate
  # ISR shrinking = followers falling behind. Normal occasionally; bad if frequent.
  # Frequent ISR shrinks = broker overloaded, network issue, or disk saturation.

# Controller metrics
kafka.controller:type=KafkaController,name=ActiveControllerCount
  # Should be 1 on exactly one broker, 0 on all others. Alert on > 1.
kafka.controller:type=KafkaController,name=PartitionsUnderIsrCount

jconsole / JMX inspection

# Enable JMX on broker (in startup script or server.properties):
export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote
  -Dcom.sun.management.jmxremote.port=9999
  -Dcom.sun.management.jmxremote.authenticate=false
  -Dcom.sun.management.jmxremote.ssl=false"

# Connect jconsole to broker:9999
# Navigate to kafka.server → ReplicaManager
# Key metrics:
#   UnderReplicatedPartitions      — should be 0
#   OfflinePartitionsCount        — should be 0
#   IsrShrinksPerSec              — should be near 0
#   IsrExpandsPerSec              — should be near 0

# Prometheus JMX Exporter (recommended for production):
# Add to kafka config:
KAFKA_OPTS="-javaagent:jmx_prometheus_java_agent.jar=7071:/path/to/config.yaml"

# prometheus config for kafka:
# - job_name: kafka-broker
#   static_config:
#     targets: ['broker-0:7071', 'broker-1:7071', 'broker-2:7071']
#   relabel_configs:
#     - source_labels: [__address__]
#       target_label: instance
#       regex: 'broker-(\d+)'

Kafka Lag Monitor: the consumer story

# Quick lag check for all consumer groups:
$ kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --describe --group my-consumer-group

GROUP              TOPIC       PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  
my-consumer-group  payments    0          4821            5103            282  
my-consumer-group  payments    1          4390            4390              0  
my-consumer-group  payments    2          5011            5011              0  

# LAG = LOG-END-OFFSET - CURRENT-OFFSET
# LAG > 0 = consumer is behind
# LAG growing = consumer can't keep up (check consumer throughput, GC logs)
# LAG = 0 = consumer caught up, reading latest messages

# Alert rules for consumer lag:
# - Lag > 10,000 for > 5 minutes → warn
# - Lag > 100,000 for > 10 minutes → page
# - Lag suddenly drops to 0 → consumer may have skipped messages (at-most-once)

# BURROW: Kafka's consumer lag monitoring library (LinkedIn)
# Provides: per-topic lag trends, cluster-wide lag dashboard, lag velocity
# Alternative: Kafka lag exporter for Prometheus + Grafana dashboard
# Grafana dashboard: "Kafka Consumer Lag" (search Grafana.com for community dashboards)

FAQ

How many partitions should a topic have?

There's no universal answer, but here's the rule of thumb: partitions = max(throughput_producer, throughput_consumer / consumer_throughput). A single consumer can do ~50 MB/s. If you need 500 MB/s producer throughput and your consumers can each do 50 MB/s, you need at least 10 partitions. More partitions also mean more parallelism for consumers, but also more file handles, longer leader election on broker failure, and a larger __consumer_offsets topic. Start conservative (6-12 for high-volume topics), monitor lag, adjust upward. Going from 10 to 100 partitions is a metadata operation; going from 100 to 10 is a data migration.

Does partition count affect replication factor?

Not directly — replication factor is a separate topic config. But they multiply: a topic with 100 partitions and rf=3 has 300 broker relationships to maintain. During a broker failure, the controller must reassign all under-replicated partitions. More partitions = more reassignment work during rolling restarts. The tradeoff: higher partition counts let you distribute load more evenly, but during a broker outage, the reassignment blast radius grows. With rf=3, losing one broker affects ~1/3 of your partitions on average.

Can I change partition count after creating a topic?

Yes, you can increase partitions with `kafka-topics.sh --alter --topic foo --partitions 20`. Kafka only supports increasing, never decreasing. Adding partitions breaks key ordering: messages with the same key may now go to different partitions because the key-hash space is divided across more partitions. If you need total order per key, partitioning strategy is locked once data exists. Plan partition counts ahead for key-ordered topics.

Why does Kafka use leader-based replication instead of multi-master?

Simplicity and consistency. Multi-master (like MySQL Group Replication or Cassandra) requires conflict resolution when the same key is written to two masters simultaneously — which is hard, slow, and subtly wrong in ways that only appear in production. Kafka's leader model is simple: all reads and writes go to the leader replica; followers just copy. If a follower falls behind, it's excluded from the in-sync replica (ISR) set and can't become leader. This gives you exactly-once semantics (with producers using idempotent config) and avoids split-brain scenarios at the cost of the leader being a single point for a given partition.

What's the difference between LEO and HW (high watermark)?

LEO (Log End Offset) is the offset of the last record written — it advances as producers append. HW (High Watermark) is the last offset that has been replicated to all in-sync replicas. Consumers can only read up to HW. This means if the leader crashes after writing 10 records but only 5 have been replicated to followers, the followers truncate back to HW=5 on election, and consumers see a consistent view. The gap between LEO and HW is the 'end-of-log' segment that followers are still catching up on. This is the core of Kafka's durability guarantee: no data loss as long as at least one ISR survives.

Why does __consumer_offsets use a compaction topic?

Each consumer group commits its offset per topic-partition. A naive design would append a new message per commit — over time that topic would grow forever. Kafka uses a special internal topic (50 partitions, rf=1 by default) with log compaction: when a new message with the same key arrives, it supersedes older ones. The cleaner runs in the background and removes old key versions, keeping only the latest offset per consumer group + topic-partition. This is why you should never manually create or delete __consumer_offsets — it's a system topic with special handling.