Free lesson · System Design Advanced

ZooKeeper Purpose and Design - Coordination as a Service and the ZAB Protocol

Hard10 min readsystem-designadvanced

Here's a thought experiment. You have five servers running your application. One of them needs to be the leader - the one that handles writes, coordinates the others, makes the authoritative decisions. How do you decide which one? You could hard-code it: "server 1 is always the leader." But what happens when server 1 crashes? Now you need the other four to agree on a new leader. And they need to agree fast - before clients notice. And they need to agree even if the network between some of them is temporarily broken. And they need to make sure that exactly one leader is chosen, not two (that's called split-brain, and it corrupts data).

This problem - getting distributed nodes to agree on shared state - is called coordination. And it turns out to be brutally hard to get right. The history of distributed systems is littered with projects that tried to implement their own coordination logic and introduced subtle bugs that manifested as data corruption, split-brain, or deadlocks under pressure.

ZooKeeper's insight was: stop reimplementing coordination in every project. Build coordination as a reliable, centralized service that other systems can depend on. ZooKeeper doesn't store your data. It doesn't run your application logic. It does exactly one thing: it maintains a small amount of coordination state (who is the leader, what's the current config, who holds the lock) with strong consistency guarantees, and it notifies interested parties when that state changes.

Why Coordination Needs Its Own Service

graph TD
    subgraph "Without ZooKeeper: Every System Reinvents Coordination"
        KAFKA_OWN["Kafka<br/>──────────────────<br/>Implements its own leader<br/>election for partitions.<br/>Implements its own failure<br/>detection for brokers.<br/>Implements its own config<br/>distribution for topics.<br/><br/>Each of these is a consensus<br/>problem. Getting any of them<br/>wrong causes data loss or<br/>split-brain."]
        
        HBASE_OWN["HBase<br/>──────────────────<br/>Implements its own region<br/>assignment protocol.<br/>Implements its own master<br/>election.<br/>Implements its own<br/>failure detection.<br/><br/>Different bugs than Kafka's<br/>implementation. Same class<br/>of problems."]
        
        HDFS_OWN["HDFS<br/>──────────────────<br/>Implements its own<br/>NameNode failover.<br/>Implements its own<br/>block location tracking.<br/>Implements its own<br/>lease management.<br/><br/>Yet another implementation<br/>of the same fundamental<br/>coordination primitives."]
    end
    
    subgraph "With ZooKeeper: Coordination Extracted to a Service"
        ZK["ZooKeeper Ensemble<br/>──────────────────<br/>One service that provides:<br/>• Leader election<br/>• Configuration management<br/>• Distributed locking<br/>• Group membership<br/>• Failure detection (via sessions)<br/><br/>Implemented once. Tested<br/>extensively. Battle-hardened<br/>across thousands of deployments."]
        
        KAFKA_ZK["Kafka<br/>Uses ZooKeeper for<br/>leader election,<br/>broker registry,<br/>topic config."]
        
        HBASE_ZK["HBase<br/>Uses ZooKeeper for<br/>region assignment,<br/>master election,<br/>server tracking."]
        
        HDFS_ZK["HDFS<br/>Uses ZooKeeper for<br/>NameNode failover,<br/>lease management."]
        
        ZK --> KAFKA_ZK
        ZK --> HBASE_ZK
        ZK --> HDFS_ZK
    end

This diagram contrasts two approaches. Without ZooKeeper, every distributed system - Kafka, HBase, HDFS - implements its own leader election, failure detection, and configuration distribution, each a consensus problem prone to different bugs but the same class of problems. With ZooKeeper, coordination is extracted to a single service that provides leader election, configuration management, distributed locking, group membership, and failure detection via sessions. This is implemented once, tested extensively, and battle-hardened across thousands of deployments. Kafka, HBase, and HDFS all delegate their coordination needs to ZooKeeper instead of reimplementing them.

Important: ZooKeeper is not a general-purpose database. It stores small amounts of metadata - kilobytes to low megabytes of coordination state. The data model is a tree of nodes (znodes), each holding up to 1 MB of data. If you're storing application data in ZooKeeper (user records, product catalogs, session data), you're using it wrong. ZooKeeper is optimized for reads (reads can be served by any node) and tolerates slower writes (writes must go through the leader and be replicated to a quorum). The typical ZooKeeper workload is 90%+ reads, with writes happening only when coordination state changes - a new leader is elected, a configuration value is updated, a lock is acquired or released.

The ZooKeeper Data Model: A Tree of Znodes

ZooKeeper's data model is a hierarchical namespace - like a file system, but simpler. Every node in the tree is called a znode. Each znode can hold data (up to 1 MB) and can have children. There are no directories versus files distinction - every znode can simultaneously hold data and have children.

graph TD
    subgraph "ZooKeeper Znode Tree (Example: Kafka Metadata)"
        ROOT["/"]
        
        ROOT --> BROKERS["/brokers<br/>──────────────────<br/>Kafka broker registry"]
        ROOT --> CONTROLLER["/controller<br/>──────────────────<br/>Data: {brokerId: 1}<br/>The current Kafka controller.<br/>Ephemeral znode - disappears<br/>if broker 1's session expires."]
        ROOT --> CONFIG["/config<br/>──────────────────<br/>Configuration overrides"]
        
        BROKERS --> IDS["/brokers/ids<br/>──────────────────<br/>Active broker list"]
        BROKERS --> TOPICS["/brokers/topics<br/>──────────────────<br/>Topic-to-partition mapping"]
        
        IDS --> B1["/brokers/ids/1<br/>──────────────────<br/>Data: {host: 10.0.1.1,<br/>port: 9092, rack: us-east-1a}<br/><br/>EPHEMERAL: exists only while<br/>broker 1's ZK session is alive.<br/>If broker 1 crashes, this znode<br/>is automatically deleted —<br/>triggering watches on /brokers/ids."]
        
        IDS --> B2["/brokers/ids/2<br/>──────────────────<br/>Data: {host: 10.0.1.2,<br/>port: 9092, rack: us-east-1b}<br/><br/>EPHEMERAL"]
        
        IDS --> B3["/brokers/ids/3<br/>──────────────────<br/>Data: {host: 10.0.1.3,<br/>port: 9092, rack: us-east-1c}<br/><br/>EPHEMERAL"]
    end

This diagram shows ZooKeeper's znode tree using Kafka's metadata as an example. The root node has children including /brokers for the broker registry, /controller holding the current Kafka controller as an ephemeral znode, and /config for configuration overrides. Under /brokers/ids, each active broker registers an ephemeral znode containing its host, port, and rack information. Ephemeral znodes exist only while the broker's ZooKeeper session is alive - if a broker crashes, its znode is automatically deleted, triggering watches on the parent path that notify other components of the failure.

Four Types of Znodes

Tip: Ephemeral znodes are ZooKeeper's killer feature. They tie the lifecycle of a piece of data to the liveness of a process. When a server registers itself with an ephemeral znode, other servers don't need to poll "is server X alive?" - they set a watch on the znode, and ZooKeeper notifies them the instant the znode disappears (meaning the server died or lost connectivity). This converts failure detection from a polling problem (expensive, slow, imprecise) to an event-driven notification (cheap, fast, precise).

The ZAB Protocol: How ZooKeeper Stays Consistent

ZooKeeper uses the ZooKeeper Atomic Broadcast (ZAB) protocol to replicate state across its ensemble. ZAB is a consensus protocol specifically designed for ZooKeeper - it's similar to Paxos and Raft in guarantees but optimized for ZooKeeper's specific workload pattern (primary-backup replication with ordered broadcasts).

Interview: "How does ZooKeeper maintain consistency across replicas?" is asked in distributed systems interviews. The strong answer: "ZooKeeper uses the ZAB (ZooKeeper Atomic Broadcast) protocol, which is a primary-backup consensus protocol. All writes go through a single leader, which assigns each write a monotonically increasing transaction ID (zxid). The leader broadcasts each write proposal to all followers. Once a majority (quorum) of servers - including the leader - have written the proposal to their transaction logs and acknowledged, the leader sends a COMMIT message. This guarantees that committed writes survive any minority of server failures. Reads can be served by any server, which means reads may return slightly stale data (the follower might not have received the latest commit yet). For linearizable reads, clients use the sync() operation, which forces the follower to catch up with the leader before responding. The key guarantee: all clients see writes in the same order - ZAB provides total order broadcast."

Sessions and Heartbeats: How ZooKeeper Detects Failures

The mechanism that makes ephemeral znodes work reliably is ZooKeeper's session model:

graph TD
    subgraph "ZooKeeper Session Lifecycle"
        CONNECT["Client connects to ensemble<br/>──────────────────<br/>Client establishes a TCP connection<br/>to any ZooKeeper server. ZooKeeper<br/>assigns a session ID and negotiates<br/>a session timeout (typically 5-30<br/>seconds, configurable by client)."]
        
        CONNECT --> HEARTBEAT["Session maintained by heartbeats<br/>──────────────────<br/>Client sends periodic heartbeat<br/>pings to the server. As long as<br/>heartbeats arrive within the<br/>session timeout, the session<br/>stays alive and all ephemeral<br/>znodes owned by this session<br/>continue to exist.<br/><br/>If the client loses connection<br/>to its server, it transparently<br/>reconnects to another server<br/>in the ensemble - the session<br/>transfers with the same session ID."]
        
        HEARTBEAT --> EXPIRE["Session expiry<br/>──────────────────<br/>If no heartbeat arrives within<br/>the session timeout, the<br/>ensemble declares the session<br/>expired. ALL ephemeral znodes<br/>owned by this session are deleted<br/>atomically. Watches on those<br/>znodes fire, notifying other<br/>clients of the failure.<br/><br/>This is how ZooKeeper detects:<br/>• Process crashes<br/>• Network partitions<br/>• Long GC pauses<br/>• Any state where the client<br/>  can't communicate"]
    end

This diagram shows the ZooKeeper session lifecycle. A client connects to any server in the ensemble and receives a session ID with a negotiated timeout, typically 5-30 seconds. The session is maintained by periodic heartbeat pings - as long as heartbeats arrive within the timeout, the session stays alive and all ephemeral znodes continue to exist. If the client loses its connection, it transparently reconnects to another server with the same session ID. If no heartbeat arrives within the session timeout, the ensemble declares the session expired, atomically deletes all ephemeral znodes owned by that session, and fires watches on those znodes to notify other clients. This is how ZooKeeper detects process crashes, network partitions, long GC pauses, and any state where the client cannot communicate.

The Production War Story

The moment ZooKeeper's design philosophy became viscerally real for me was during my time at Razorpay, building the payment gateway that processed 800 crore rupees per day. We had a distributed processing pipeline with six worker nodes that needed leader election - one node coordinated transaction batching while the others processed individual payments. Initially, we built our own leader election using Redis. The logic was simple: a worker acquires a Redis lock with an expiry TTL, and the lock-holder is the leader. If the leader crashes, the lock expires and another worker acquires it.

It worked for eight months. Then one night at 2am, the leader's process froze (garbage collection pause in Java - 18 seconds) but didn't crash. The Redis lock expired because the TTL elapsed. A second worker acquired the lock and became the new leader. The original worker's GC completed, and it resumed processing - still believing it was the leader. Two leaders. Both coordinating transaction batches. Both issuing conflicting instructions to the payment processor. We discovered the problem at 6am when merchant reconciliation showed duplicate settlements. Fourteen merchants received double payments totaling 3.2 crore rupees. The reversal process took four days.

The root cause was that Redis locks don't have session semantics. A Redis lock doesn't know whether the holder is alive - it only knows whether the TTL has elapsed. ZooKeeper's ephemeral znodes solve exactly this problem: the znode exists as long as the client session is alive, and the session has heartbeats that detect freezes (not just crashes). If the client freezes for longer than the session timeout, ZooKeeper proactively kills the session and deletes the ephemeral znode - triggering a new leader election before the old leader can resume and cause split-brain.

After that incident, we rebuilt the leader election on ZooKeeper. We never had a split-brain event again. The lesson: coordination looks simple until it encounters the edge cases that only happen under production pressure - GC pauses, network partitions, clock skew, partial failures. ZooKeeper's entire design is built around surviving those edge cases. Building your own coordination from scratch means discovering those edge cases the hard way.

Six Takeaways

1. ZooKeeper provides coordination as a centralized, reliable service. Instead of every distributed system reimplementing leader election, locking, and configuration management, ZooKeeper implements these primitives once with strong consistency guarantees.

2. The data model is a tree of znodes - like a simplified file system. Each znode holds up to 1 MB of data. ZooKeeper is for small coordination metadata, not application data.

3. Ephemeral znodes tie data lifecycle to process liveness. When a process dies, its ephemeral znodes are automatically deleted, notifying watchers instantly. This is ZooKeeper's most powerful primitive.

4. ZAB (ZooKeeper Atomic Broadcast) provides total-order broadcast with majority quorum. All writes go through a single leader, are assigned sequential transaction IDs, and are committed once a quorum acknowledges.

5. Reads can be served by any server but may be stale. For linearizable reads, use the sync() operation to force the follower to catch up with the leader first.

6. Building your own coordination with Redis locks or database flags invites split-brain bugs. ZooKeeper's session semantics (heartbeats + ephemeral znodes) detect frozen processes, not just crashed ones - closing the gap that causes dual-leader scenarios.

Exercises

Exercise: Starter: A distributed system has 5 nodes and uses ZooKeeper for leader election. (a) What type of znode does the leader create? (b) If the leader crashes, what happens to that znode? (c) How do the remaining 4 nodes learn that the leader has failed? (d) If the leader's process freezes for 30 seconds (GC pause) but the ZooKeeper session timeout is 10 seconds, what happens? Compare this to what would happen with a Redis lock with 10-second TTL.

Exercise: Intermediate: You're designing a service registry using ZooKeeper. Ten application servers register themselves at startup. Consumers need to discover available servers. (a) Design the znode tree structure. What paths do you use? What data does each znode contain? (b) Should server registration znodes be persistent or ephemeral? Why? (c) How do consumers discover new servers and detect failed servers? (d) A server experiences a network partition - it's alive but can't reach ZooKeeper. Its session expires and its ephemeral znode is deleted. Consumers stop sending traffic. But the server is still running and processing in-flight requests. How do you handle this gracefully?

Exercise: Advanced: Compare ZAB to Raft in detail. (a) Both are leader-based consensus protocols. How does leader election differ? (ZAB uses epoch-based election; Raft uses term-based election with log comparison.) (b) Both require a quorum for writes. In a 5-node ensemble, what is the quorum? How many nodes can fail while maintaining availability? (c) ZAB provides total-order broadcast. Raft provides linearizable state machine replication. Are these equivalent? What's the practical difference? (d) ZooKeeper uses ZAB. etcd uses Raft. If you were building a new coordination service today, which protocol would you choose? Consider: implementation complexity, performance characteristics, formal verification, and community support.

Back to System Design Advanced