Apache ZooKeeper

Overview

Apache ZooKeeper is a centralized service for maintaining configuration information, naming, providing distributed synchronization, and group services. Originally developed at Yahoo! Research as the open-source counterpart to Google’s Chubby lock service, ZooKeeper has become the coordination backbone for the entire Hadoop ecosystem and many other distributed systems.

ZooKeeper solves one of the hardest problems in distributed computing: how do independent processes agree on shared state? Rather than every distributed system implementing its own consensus protocol, ZooKeeper provides coordination as a service — a small, fast, reliable system that other systems can build upon.

Key capabilities include:

• Configuration management: Centralized, versioned configuration that processes watch for changes
• Leader election: Building blocks for electing a leader among distributed processes
• Distributed locks: Mutual exclusion across machines via sequential ephemeral nodes
• Service discovery: Dynamic registration and lookup of available services
• Group membership: Tracking which processes are alive in a cluster
• Barriers and queues: Synchronization primitives for distributed workflows

Systems that depend on ZooKeeper include Apache Kafka (broker coordination), Apache HBase (region server coordination), Apache Solr (SolrCloud), Apache Hadoop (YARN resource management), and LinkedIn’s Helix.

Architecture & Core Components

Data Model: The ZNode Tree

ZooKeeper’s data model is a hierarchical namespace, similar to a filesystem:

graph TD
    Root["/"] --> Config["/config"]
    Root --> Services["/services"]
    Root --> Election["/election"]

    Config --> DB["/config/database<br/>data: connection_string"]
    Config --> Cache["/config/cache<br/>data: redis://..."]

    Services --> Web1["/services/web-1<br/>ephemeral"]
    Services --> Web2["/services/web-2<br/>ephemeral"]
    Services --> Web3["/services/web-3<br/>ephemeral"]

    Election --> Leader["/election/leader-0000000001<br/>ephemeral sequential"]
    Election --> Candidate["/election/leader-0000000002<br/>ephemeral sequential"]

ZNode types:

Each ZNode can store up to 1 MB of data (intentionally small — ZooKeeper is for coordination metadata, not bulk storage).

Server Architecture

graph TD
    subgraph "ZooKeeper Ensemble (3 or 5 servers)"
        L[Leader] -->|"Proposals (ZAB)"| F1[Follower 1]
        L -->|"Proposals (ZAB)"| F2[Follower 2]
        F1 -->|"ACKs"| L
        F2 -->|"ACKs"| L
    end

    C1[Client 1] --> F1
    C2[Client 2] --> L
    C3[Client 3] --> F2

    subgraph "Write Path"
        W["Write → Leader → Propose to all → Majority ACK → Commit"]
    end

    subgraph "Read Path"
        R["Read → Any server (local read, eventually consistent)"]
    end

• Leader: Handles all write requests. Proposes writes to followers via ZAB protocol.
• Followers: Participate in write quorum. Serve read requests directly (may be slightly stale).
• Observers: Non-voting members that receive committed updates. Scale read capacity without affecting write quorum.

ZAB Protocol (ZooKeeper Atomic Broadcast)

ZAB is ZooKeeper’s consensus protocol, similar to Raft but developed independently:

1. Leader election: When the leader fails, followers elect a new leader with the most up-to-date transaction log
2. Discovery: New leader synchronizes its state with followers
3. Synchronization: Followers catch up to the leader’s latest state
4. Broadcast: Leader proposes writes, waits for majority ACK, then commits

ZAB guarantees:

• Reliable delivery: If a message is delivered to one server, it’s eventually delivered to all
• Total order: All servers see the same sequence of state changes
• Causal order: If message A causally precedes B, A is delivered before B

Key Features

Watches

Clients can set watches on ZNodes to receive notifications when data changes:

Client: getData("/config/database", watch=true)
→ Returns current data + registers watch

[Another client updates /config/database]

Client receives WatchEvent: NodeDataChanged on /config/database
→ Client re-reads and re-sets watch

Watches are one-time triggers — after firing, the client must re-register. This design avoids the thundering herd problem that persistent subscriptions would cause.

Recipes (Coordination Patterns)

ZooKeeper provides primitives, not high-level abstractions. Common patterns built on these primitives:

Leader Election

1. Each candidate creates an ephemeral sequential node:
   /election/leader-0000000001
   /election/leader-0000000002

2. Candidate with the lowest sequence number is the leader

3. Non-leaders watch the node with the next-lower sequence number
   (not the leader — this prevents the "herd effect")

4. If that node disappears (session timeout), check if you're now lowest → become leader

Distributed Lock

1. Create ephemeral sequential node: /locks/resource-0000000005

2. Get all children of /locks, sort by sequence number

3. If you have the lowest number → you hold the lock

4. Otherwise, watch the node immediately before yours

5. When that node is deleted → re-check if you're now lowest

6. Release: delete your ephemeral node (or let session expire)

This creates a fair, FIFO lock without thundering herd.

Service Discovery

1. Service starts → creates ephemeral node:
   /services/payment-service/instance-001
   data: {"host": "10.0.1.5", "port": 8080}

2. Clients list children of /services/payment-service
   → get all live instances

3. Service crashes → ephemeral node auto-deleted
   → watching clients get notified

4. Clients re-list children → updated set of instances

Use Cases

When to Use ZooKeeper

• Leader election for distributed services (which instance handles writes?)
• Configuration management with change notifications (push-based config updates)
• Service discovery with automatic deregistration on failure
• Distributed locking for coordinating access to shared resources
• Cluster membership (who’s alive?) for systems like Kafka, HBase, Solr
• Work distribution (assigning partitions/shards to workers)

When NOT to Use ZooKeeper

• Bulk data storage: ZNode limit is 1MB, and ZooKeeper keeps all data in memory
• Message queuing: Use Kafka, RabbitMQ, or SQS instead
• Service mesh / load balancing: Use Envoy, Istio, or Consul
• Key-value store at scale: Use Redis, DynamoDB, or etcd
• New projects that don’t need Hadoop ecosystem integration: Consider etcd (simpler, Raft-based, gRPC API, Kubernetes-native) or Consul (built-in service mesh, health checks, DNS)

Comparison with Alternatives

When to Choose Which

• ZooKeeper: You’re in the Hadoop/Kafka ecosystem, or you need the hierarchical namespace and sequential nodes
• etcd: You’re on Kubernetes, or you want a simpler, modern alternative with persistent watches
• Consul: You need built-in service discovery with health checks, or you want service mesh capabilities

Performance Characteristics

Typical Numbers

• Read throughput: 10K-100K reads/sec (reads served locally from follower memory)
• Write throughput: 1K-10K writes/sec (limited by ZAB consensus — all writes go through leader)
• Read latency: <1ms (in-memory data store)
• Write latency: 2-10ms (depends on cluster size, network latency, disk fsync)
• Session timeout: Typically 30-60 seconds (trade-off: faster failure detection vs false positives)
• Cluster size: 3, 5, or 7 nodes (always odd for majority quorum)

Write vs Read Scaling

• Reads scale horizontally: Add observers for more read capacity
• Writes do NOT scale: All writes go through the single leader. Adding more followers/observers does not increase write throughput (it may decrease it due to more ACKs needed)

Production Operational Concerns

Monitoring Key Metrics

• Outstanding requests: Queued requests on each server (should be near 0)
• Latency (avg/p99): Read and write latency per server
• Alive connections: Number of client sessions
• Watch count: Total registered watches (high count = potential thundering herd)
• ZNode count: Total nodes (memory proportional)
• Data size: Total data in memory (should fit comfortably in RAM)
• Leader election time: How long elections take after leader failure

Common Failure Modes

1. Leader election storm: Frequent leader elections due to GC pauses or network instability → tune JVM GC, increase tick time
2. Session timeouts: Clients disconnecting and reconnecting frequently → increase session timeout, investigate network issues
3. Out of memory: Too many ZNodes or watches → monitor data size, clean up stale nodes
4. Split brain: Network partition separates the ensemble → ensure odd number of nodes, configure proper quorum
5. Transaction log growth: Unbounded log growth → configure autopurge (snapRetainCount + purgeInterval)

JVM Tuning

ZooKeeper is Java-based, so GC tuning is critical:

• Use G1GC or ZGC to minimize pause times
• Allocate sufficient heap (4-8 GB typical)
• Monitor GC pause times — pauses > tick interval can trigger leader elections
• Use dedicated SSDs for transaction logs (separate from snapshots)

Interview Tips

• “How does Kafka use ZooKeeper?” — Broker registration, topic/partition metadata, controller election, consumer group coordination (note: KRaft mode in newer Kafka removes ZooKeeper dependency)
• “How would you implement distributed locking?” — Ephemeral sequential nodes, watch the predecessor, FIFO ordering avoids thundering herd
• “ZooKeeper vs etcd — when would you choose each?” — ZooKeeper for Hadoop ecosystem and hierarchical data; etcd for Kubernetes and simpler operations (Go vs Java, gRPC vs custom protocol)
• “What consistency guarantees does ZooKeeper provide?” — Sequential consistency (writes totally ordered), client-session ordering, atomicity of writes. Reads may be stale (served from followers). Use sync() before read for linearizable reads.
• “How does ZooKeeper handle the leader failing?” — Remaining followers elect new leader via ZAB. Quorum required (majority of configured members). Typically completes in seconds.