Kafka's replication: durable until it isn't

Kafka is a messaging system built around an immutable, linearizable, sharded log. Its throughput and storage scale linearly with nodes, and each node can push enough volume to saturate disk or network. Consumers coordinate reads via Zookeeper, which provides efficient at-least-once delivery and replayability.

The upcoming 0.8 release introduces replication to improve durability and availability by duplicating each shard's data across nodes. In the context of CAP, Kafka claims serializability and availability while sacrificing partition tolerance—a tradeoff based on the assumption that brokers live in a datacenter where partitions are rare. That assumption, however, raises a question: what happens during a network partition, and can a single failure cascade into committed-data loss?

How replication works

Kafka's replication design elects one leader per shard via Zookeeper. The leader tracks a set of in-sync replicas (ISR): nodes that are current with the leader's log and actively acknowledging writes. Every write goes through the leader and is replicated to each node in the ISR. Only after all ISR nodes acknowledge does the leader commit and respond to the client.

When a node fails, the leader detects timed-out writes, removes the node from the ISR, and continues acknowledging writes from the remaining set. That allows Kafka to tolerate some node failures. The documentation goes further, claiming that with f nodes, Kafka can tolerate f-1 failures—a stronger guarantee than most CP systems, which typically require a majority (n/2-1 failures).

Kafka achieves that claim by allowing the ISR to shrink to a single node: the leader alone. In that state, the leader acknowledges writes persisted only locally. If the leader then loses its Zookeeper connection, the system cannot safely continue—but it proceeds anyway, promoting whatever node remains (which could be arbitrarily behind) as the new leader.

When the original leader returns, its log conflicts with the new leader's: the two agree up to some point, then diverge. Kafka preserves linear ordering by dropping the old leader's writes—destroying acknowledged, committed data.

A specific failure mode

Losing committed data requires two conditions:

  1. The ISR must shrink so that any node other than the current leader is removed from it.
  2. All nodes in the ISR must lose their Zookeeper connection.

Both can follow from a single complex failure. For example, a lossy NIC might drop Kafka packets while Zookeeper connections survive briefly, isolating the leader from followers until Zookeeper later times out. Or a network partition separates the leader, followed by a crash or administrator restart. The old leader's writes become causally disconnected from the new leader's log, making loss possible.

Testing the claim

Kafka includes parameters controlling write consistency. The default mode does not replicate writes before acknowledging, which maximizes throughput at the cost of safety. The test runs in synchronous mode, so every acknowledged write should be durable on all ISR nodes.

The experiment loads a series of integers into the cluster, then isolates the leader with iptables. Latency spikes show timeouts while the missing nodes fail to respond. After a few seconds, the ISR shrinks and writes succeed again. While the leader acknowledges writes locally, everything appears healthy—but those writes exist only on one node.

Partitioning the leader entirely causes Zookeeper to promote a new leader and data loss occurs, again preceded by a latency spike. At the end of the run, Kafka acknowledges 98-100% of writes; roughly half of them—those made during the partition—are gone.

Design alternatives

Kafka's replication sacrificed partition safety while claiming high availability. That still leaves a tradeoff between required node count per write and fault-induced data loss. Quorum theory offers a proven optimum: for a network where nodes fail independently with probability p < 1/2, the majority coterie maximizes availability. Worst-case are the extremes—single-node quorums (one failure causes data loss) and all-node quorums (one failure blocks writes). Since Kafka keeps only 1-10 replicas per shard, majority quorums are provably optimal for availability.

Two changes would address the failure mode:

  1. Bounding the ISR to at least N/2 nodes, which reduces the chance that a single additional failure loses committed writes.
  2. Blocking and alarming instead of progressing when the ISR becomes empty. Administrators should have a chance to recover an old leader or preserve conflicting writes rather than letting the system silently discard them.

This is candidate design for pre-release software, not a finished product. A "stronger safety" mode that keeps the ISR bounded and halts when it empties was under discussion—if it reaches the release and strong safety matters, verify it is enabled.