Partitioned commit log

Deep dive

How it works

Parts:

• Partitioner (producer side): hash(key) mod N; keyless records round robin or sticky.
• Partition: one ordered log per replica; offset = position (files: segmented log).
• Leader and followers: producers write to the leader; followers fetch like consumers.
• In-sync replica set (ISR): replicas the leader sees as caught up; only they can become leader.
• High watermark (HW): smallest log end in the ISR; consumers read only below it.
• Leader epoch: counter raised per election, stamped on records; decides truncation.
• Producer id + sequence: per partition, so retries are not appended twice.
• Consumer group: one reader per partition, position in consumer offsets.

1. The producer picks the partition, adds the record to its batch (batch.size 16 KiB, linger.ms 5 in Kafka 4.1), compresses whole batches, sends one request per broker.
2. The leader appends, assigns consecutive offsets, stamps the epoch. For an idempotent producer a duplicate (producer id, sequence) is dropped and a gap raises OutOfOrderSequenceException (KIP-98).
3. Followers fetch from the leader. One silent or behind for replica.lag.time.max.ms (30 s) leaves the ISR. HW = minimum log end over the ISR.
4. acks=0 returns at once; acks=1 after the leader's append; acks=all after every ISR member has it, failing with NotEnoughReplicas if the ISR is below min.insync.replicas.
5. Consumers fetch up to the HW, never past it.
6. Leader failure: the controller picks a new leader from the ISR and raises the epoch. Each follower asks it for the end offset of its own last epoch and truncates to that (KIP-101); the old leader does the same on return. Lost records were above the HW: no reader saw them and no acks=all producer was acked.
7. Transactions (KIP-98): commit or abort markers are written into every partition touched; read_committed consumers skip uncommitted and aborted records.

key ─hash mod 3─► P0  leader A: [0 1 2 3 4 5 6]  ◄─fetch── B: [0..5]  C: [0..4]
                    P1  leader B: [...]    ISR={A,B,C}  HW = 5: consumers read offsets 0..4
 A crashes ─► B leads, epoch+1, C cut to B's epoch end; offsets 5,6 on A were uncommitted

Code

In-process model, compiled and tested: committed records survive failover, unreplicated acks=1 records are lost, NotEnoughReplicas fires.

#[derive(Clone, Copy, PartialEq, Debug)]
pub struct Rec { pub epoch: u32, pub key: u64, pub val: u64 }
#[derive(Clone, Copy, PartialEq)]
pub enum Acks { None, Leader, All }
#[derive(Debug, PartialEq)]
pub enum ProduceError { NotEnoughReplicas }
/// One partition: a leader log and two follower logs; offset = index.
pub struct Partition {
    pub logs: [Vec<Rec>; 3],
    pub leader: usize,
    pub isr: Vec<usize>, // caught-up replicas, leader included
    pub epoch: u32,
}
impl Partition {
    pub fn new() -> Self {
        Partition { logs: Default::default(), leader: 0, isr: vec![0, 1, 2], epoch: 0 }
    }
    /// High watermark: first offset not on every ISR member.
    pub fn hw(&self) -> usize {
        self.isr.iter().map(|&r| self.logs[r].len()).min().unwrap()
    }
    pub fn produce(&mut self, key: u64, val: u64, acks: Acks, min_isr: usize) -> Result<usize, ProduceError> {
        if acks == Acks::All && self.isr.len() < min_isr {
            return Err(ProduceError::NotEnoughReplicas);
        }
        let off = self.logs[self.leader].len();
        self.logs[self.leader].push(Rec { epoch: self.epoch, key, val });
        if acks == Acks::All {
            self.replicate(usize::MAX);
        }
        Ok(off)
    }
    /// One follower-fetch round: each ISR follower copies up to `max` records.
    pub fn replicate(&mut self, max: usize) {
        let l = self.leader;
        for f in self.isr.clone().into_iter().filter(|&f| f != l) {
            let from = self.logs[f].len();
            let to = self.logs[l].len().min(from.saturating_add(max));
            let chunk = self.logs[l][from..to].to_vec();
            self.logs[f].extend(chunk);
        }
    }
    /// Consumers see only records below the HW.
    pub fn fetch(&self, from: usize) -> &[Rec] {
        let hw = self.hw();
        &self.logs[self.leader][from.min(hw)..hw]
    }
    /// Leader dies: first remaining ISR member leads, new epoch.
    pub fn crash_leader(&mut self) {
        let dead = self.leader;
        self.isr.retain(|&r| r != dead);
        self.leader = *self.isr.first().expect("ISR empty");
        self.epoch += 1;
        for f in self.isr.clone() {
            self.truncate_to_leader(f);
        }
    }
    /// Epoch truncation: cut to where the leader's log leaves the follower's
    /// last epoch (or to the leader's end), not to a high watermark.
    fn truncate_to_leader(&mut self, f: usize) {
        let Some(last) = self.logs[f].last().map(|r| r.epoch) else { return };
        let ll = &self.logs[self.leader];
        let end = ll.iter().position(|r| r.epoch > last).unwrap_or(ll.len());
        self.logs[f].truncate(end);
    }
    /// A crashed replica returns as follower.
    pub fn recover(&mut self, r: usize) {
        self.truncate_to_leader(r);
        self.isr.push(r);
        self.replicate(usize::MAX);
    }
}
pub struct Topic { pub parts: Vec<Partition> }
fn fnv1a(key: u64) -> u64 { // stand-in for the client's murmur2
    key.to_le_bytes().iter().fold(0xcbf29ce484222325, |h, &b| (h ^ b as u64).wrapping_mul(0x100000001b3))
}
impl Topic {
    pub fn new(n: usize) -> Self { Topic { parts: (0..n).map(|_| Partition::new()).collect() } }
    /// Same key, same partition: per-key order = that partition's order.
    pub fn route(&self, key: u64) -> usize { (fnv1a(key) % self.parts.len() as u64) as usize }
}
#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn failover() {
        let mut p = Partition::new();
        for v in 0..5 { p.produce(1, v, Acks::All, 2).unwrap(); }    // committed
        for v in 5..8 { p.produce(1, v, Acks::Leader, 2).unwrap(); } // leader only
        assert_eq!(p.fetch(0).len(), 5);                             // HW = 5
        p.replicate(1);                                              // followers reach 6
        p.crash_leader();
        p.produce(1, 100, Acks::All, 2).unwrap();
        p.recover(0);                                                // old leader cut from 8 to 6
        let vals: Vec<u64> = p.fetch(0).iter().map(|r| r.val).collect();
        assert_eq!(vals, [0, 1, 2, 3, 4, 5, 100]);                   // 6 and 7 lost
        p.crash_leader();
        p.crash_leader();                                            // ISR = {2}
        assert_eq!(p.produce(1, 1, Acks::All, 2), Err(ProduceError::NotEnoughReplicas));
    }
}

Not modelled: network, timeouts, controller, producer ids, transactions.

Complexity and constants

• Produce O(1) amortized per record; routing is one hash over the key; fetch is a sequential read.
• Parallelism = N: at most N active consumers per group; throughput grows with N until a broker saturates.
• Per partition: a log per replica, a follower fetch stream, election work, a batch.size buffer per producer (16 KiB x N).
• Durability comes from replication, not fsync (flush.ms defaults to unlimited).

Efficiency

• Cost: one acks=all produce = client-to-leader round trip + follower-fetch round trip + page-cache append, no fsync by default. Bound: network latency, not CPU. Sim, 1,000,000 records: 1,000,000 follower-fetch rounds at batch 1, 1,000 at batch 1,000.
• Levers:
  • batching and linger — rounds fall as 1/batch (sim count) — always; costs up to linger.ms latency.
  • compression per batch — better ratio with bigger batches, default none (Kafka producer docs) — repetitive payloads; costs CPU.
  • sticky partitioning, keyless records — fills one batch before switching (Kafka producer docs) — when cross-key order is irrelevant.
  • acks=1 — skips the follower round trip but loses unreplicated records on failover (sim) — only where loss is acceptable.
  • zero-copy fetch — sendfile cuts four copies to one (Kafka design docs) — consumers served from page cache.
• Naive → standard → state of the art:
  • one partition, synchronous send per record · one round trip per record
  • N keyed partitions, linger batching, idempotent acks=all, factor 3, min.insync.replicas=2 · throughput grows with N until a broker limit
  • Raft per partition (Redpanda), adaptive partitioner (Kafka partitioner.adaptive.partitioning.enable) · same shape, fewer stalls

Implementation notes

• N = peak throughput / measured per-partition rate, or the wanted consumer parallelism if larger; leave headroom, since growing N moves keys.
• Use one hash in all producers: Pulsar's default Java string hash differs from Murmur3, so mixed clients can split a key.
• Set min.insync.replicas=2 with replication factor 3; at the default 1, acks=all succeeds with a one-replica ISR.
• Keep in-flight requests at 5 or fewer with idempotence on, so retries cannot reorder.

Pitfalls

• Hot key: its rate caps one partition. Salt the key (key#0..k), giving up order across the salt.
• Repartitioning: after N changes hash mod N moves keys, old records stay (Event Hubs warns of this too); migrate to a new topic instead.
• Thousands of small partitions: file handles, election time and metadata grow.
• No order across partitions: sort by timestamp, accept skew (see watermarks).

Variants and alternatives

In production

Kafka and Redpanda carry event pipelines and change-data-capture feeds (topic per table, primary key as record key). Pulsar, AWS Kinesis shards and Azure Event Hubs partitions serve telemetry, log and metric streams keyed by device or host.

Implementations

Sources

• Apache Kafka 4.1 documentation: design · producer configs · topic configs
• KIP-101: Alter Replication Protocol to use Leader Epoch rather than High Watermark for Truncation. KIP
• KIP-98: Exactly Once Delivery and Transactional Messaging. KIP
• Redpanda documentation, architecture. docs
• Apache Pulsar documentation, messaging concepts. docs
• AWS, Kinesis Data Streams. concepts · limits
• Microsoft, Azure Event Hubs features. docs
• Jay Kreps, Neha Narkhede, Jun Rao, Kafka: a Distributed Messaging System for Log Processing, NetDB 2011.