ZooKeeper

Apache ZooKeeper is a replicated coordination service that came out of Yahoo in 2008 and was described by Hunt, Konar, Junqueira, and Reed at USENIX ATC 2010. It keeps a small hierarchical tree of data nodes, called znodes, fully in memory on every server of an ensemble, and exposes a filesystem-flavored API of about a dozen calls such as create, delete, exists, getData, setData, and getChildren. Applications do not store their data in it. They store the metadata that keeps a fleet honest, which server is the leader, which workers are alive, what the current configuration says, and who holds a lock.

The paper calls the design a coordination kernel. The server ships no locks and no elections. It ships wait-free primitives with two ordering guarantees, linearizable writes and FIFO order for each client's own operations, and everything else is a client-side recipe built on top. That is the opposite of Google's Chubby, which bakes locking into the service, and the wait-free choice means a slow or stuck client can never block a fast one inside ZooKeeper itself.

The ideas outlived the tool's dominance. Kubernetes standardized on etcd, Kafka replaced its ZooKeeper dependency with the Raft-based KRaft controller, and ClickHouse shipped its own Keeper. ZooKeeper still sits under HBase, SolrCloud, Pulsar's BookKeeper, and Hadoop HA, and its recipes, ephemeral nodes plus sequential nodes plus watches, are the vocabulary every later system borrowed.

What it is and the mental model

Think of ZooKeeper as a synchronized metadata filesystem replicated across an ensemble of three, five, or seven servers. Every server holds the entire data tree in RAM, backed by a write-ahead transaction log and periodic snapshots on disk, so the dataset must stay small. Each znode stores at most one megabyte by default and in practice holds a few bytes to a few kilobytes, an address, a config value, a leader's name.

The service makes five documented guarantees. Sequential consistency, updates from one client apply in the order sent. Atomicity, an update either happens fully or not at all. Single system image, a client never sees an older view after reconnecting to a different server, enforced by comparing the client's last-seen transaction id against the new server's. Durability, an acknowledged update survives server crashes because it was fsynced to a quorum of transaction logs first. Timeliness, a client's view lags the truth by at most a bounded interval.

The sharp edge in that list is what is missing. Reads are served from whatever replica the client happens to be connected to, without consulting the leader, so a read can return stale data even though every write is linearizable. A client that needs read-your-latest issues sync before the read, which flushes the pending write pipeline from the leader to that replica. This asymmetry, linearizable writes with fast possibly-stale local reads, is the single design decision that gives ZooKeeper its read throughput.

How to use it

Clients connect with a list of ensemble addresses and hold one TCP session to a single server, failing over to another server on disconnect. The session, not the connection, is the unit of liveness. The client library heartbeats after the session has been idle for one third of the timeout and abandons the server for a new one after two thirds without a response, so a session survives individual server failures.

The dominant usage pattern is cache locally, watch for changes, never poll. Each application server reads the znodes it cares about once, sets watches, and answers all its own lookups from a local copy. ZooKeeper only sees traffic when something actually changes. A service registry makes this concrete, servers register ephemeral znodes under a well-known path, peers watch the children list, and routing tables are rebuilt from the notification rather than looked up per message.

In Python the kazoo library wraps the raw API and also implements the official recipes, so registration, membership watching, dynamic config, locks, and election are each a few lines.

from kazoo.client import KazooClient

zk = KazooClient(hosts="zk1:2181,zk2:2181,zk3:2181", timeout=10.0)
zk.start()   # opens a session; the client heartbeats it for you

# Service registration: the ephemeral flag ties the znode's life
# to this session. If this process dies, the ensemble deletes it.
zk.ensure_path("/app/workers")
zk.create("/app/workers/w-", b"10.0.0.7:8080",
          ephemeral=True, sequence=True)

# Membership: kazoo re-arms ZooKeeper's one-shot watch after every
# event, so this callback runs on each join, leave, or crash.
@zk.ChildrenWatch("/app/workers")
def rebuild(children):
    ring.rebuild(children)        # keep a local cache, never poll

# Dynamic config with optimistic concurrency on the znode version.
data, stat = zk.get("/app/config/batch_size")
zk.set("/app/config/batch_size", b"500", version=stat.version)

# Recipes from the ZooKeeper docs ship inside the client library.
with zk.Lock("/app/locks/reindex", identifier="w-7"):
    run_reindex()                 # one holder across the cluster

zk.Election("/app/election", "w-7").run(lead_the_shard)

Znodes, sessions, and watches

Znodes come in two base types with one modifier. Persistent znodes live until explicitly deleted and hold configuration and directory structure. Ephemeral znodes are bound to the creating session and are deleted by the ensemble the moment that session ends, whether by clean close or by timeout, which is what turns liveness into data. Ephemeral znodes cannot have children. The sequential flag appends a monotonically increasing ten-digit counter to the name, and the paper guarantees the counter never goes backward under a given parent. Later releases added TTL znodes and container znodes that are garbage collected when empty, both arriving in the 3.5 line.

Every znode carries a stat structure with version counters and the transaction ids that created and last modified it, and setData and delete accept an expected version, giving compare-and-swap without server-side locks. Sessions carry a negotiated timeout, typically ten to thirty seconds, and expiry is decided by the ensemble, not the client, so a partitioned client's ephemerals vanish even though the client still believes it is alive.

Watches are one-time triggers. A read call with the watch flag registers interest on the server, and when the znode changes the server sends exactly one event and drops the watch. The event says that something changed, not what, so the client re-reads and re-arms in the same call, and ZooKeeper guarantees the client sees the event before it sees any data written after the change. Two changes in quick succession can therefore collapse into one notification, which is fine for the cache-refresh pattern and dangerous if you try to treat watches as a message queue. Since version 3.6 there are also persistent and recursive watch modes that survive triggering, closing part of the gap with etcd.

/app
  /config
    batch_size          persistent            "500"   version=7
  /workers
    w-0000000012        ephemeral+sequential  "10.0.0.7:8080"
    w-0000000013        ephemeral+sequential  "10.0.0.9:8080"
  /election
    n_0000000041        ephemeral+sequential  "w-12"
  /locks
    /reindex
      lock-0000000007   ephemeral+sequential  owner metadata

ZAB, quorums, and the write path

Every state change is stamped with a zxid, a sixty-four bit number whose high thirty-two bits hold an epoch that increments on each leadership change and whose low thirty-two bits count proposals within that epoch. ZAB, the ZooKeeper Atomic Broadcast protocol, runs in two phases. In leader activation, servers elect the peer with the most recent transaction history, the new leader syncs each follower by sending missing proposals or a full snapshot, discards any uncommitted proposals from dead epochs, and becomes active once a quorum acknowledges its NEW_LEADER proposal. In active messaging, the leader turns every write into an idempotent state-delta transaction, streams a PROPOSAL to all followers over per-follower TCP connections that preserve FIFO order, each follower fsyncs the proposal to its transaction log and replies ACK, and once a majority has acknowledged, the leader sends COMMIT and answers the client.

The correctness argument is the quorum overlap. A transaction only commits after a majority persisted it, and any two majorities of the same ensemble share at least one server, so whichever candidate holds the highest zxid during the next election provably holds every committed transaction. An ensemble of five survives two dead servers, an ensemble of three survives one, and if no partition holds a majority the service refuses writes rather than split-brain. Recovery is cheap because transactions are idempotent, a server loads its latest fuzzy snapshot, taken without locking the tree, and simply replays the log over it, applying some changes twice with no harm.

Reads never touch this pipeline. Each replica answers from local memory and tags the response with its last-seen zxid, which is how the paper got its throughput numbers on 2010-era dual-core Xeon machines, roughly eighty-seven thousand reads per second on three servers rising to four hundred sixty thousand on thirteen, while write throughput fell from about twenty-one thousand to eight thousand per second as the quorum grew. Observers, replicas that receive committed transactions but do not vote, extend read capacity without paying that write penalty. The paper also reports leader elections completing in under two hundred milliseconds in their failure injections, so a follower crash barely dents throughput and a leader crash costs well under a second of write availability.

A quorum write in a three-server ensemble. The leader orders the change with a zxid, followers fsync then ACK, and the commit both answers the writer and fires the one-shot watch that Service A registered on its local follower. Reads never cross the quorum path.

Recipes: leader election and locks

The official leader election recipe uses sequential ephemeral znodes as tickets. Every candidate creates one under an election path, lists the children, and the lowest sequence number leads. The naive follow-up, everyone watches the leader's znode, has a failure mode the docs name the herd effect. When the leader dies, all remaining candidates get woken at once, and all of them stampede the ensemble with getChildren calls at the same instant, a burst that grows linearly with contenders and lands exactly when the system is already stressed by a failure. The fix is a chain, each candidate watches only the znode with the next-lower sequence number, so a death wakes exactly one process. If the woken process's znode is now the lowest it leads, otherwise it re-watches the new next-lower znode, which handles the middle-of-the-chain crash case cleanly.

The lock recipe is the same chain with one extra rule, list the children without a watch, then set a single exists watch on the immediate predecessor. Fairness falls out for free since grants proceed in sequence order, and a crashed holder releases automatically because its ephemeral znode disappears with its session. Read-write locks are a small variant, a read request only waits on earlier write znodes while a write request waits on all earlier znodes. The same primitives compose into barriers, double barriers, and priority queues, all documented as recipes rather than server features.

One caveat is standard distributed-systems knowledge rather than a ZooKeeper bug. The lock is advisory, and there is a window where a paused holder, say one stuck in a long garbage collection, keeps acting after its session expired and the lock passed on. Protecting the resource requires a fencing token that the storage layer checks, and the sequence number or the zxid of the lock znode serves exactly that purpose. ZooKeeper locks also are not built for high-frequency acquisition, every acquire and release is a quorum write, so hundreds of lock operations per second belong in Redis or a database while ZooKeeper handles long-lived, correctness-first locks.

acquire(lockpath):
  1. me = create(lockpath + "/lock-", EPHEMERAL | SEQUENTIAL)
  2. kids = get_children(lockpath)        # no watch on this call
  3. if me is the lowest sequence number: return HELD
  4. prev = the child immediately below me
  5. if exists(prev, watch=True) is false: goto 2
  6. block until the delete event for prev, then goto 2

release():
  delete(me)     # or crash: session expiry deletes it for you

Trade-offs and when NOT to use it

Write capacity shrinks as the ensemble grows, because every write must reach a larger quorum, the paper measured writes falling from about twenty-one thousand per second on three servers to eight thousand on thirteen while reads scaled the other way. The whole tree must fit in the JVM heap of every server, znodes are capped at one megabyte each by default, and swapping is fatal to latency because the totally ordered pipeline stalls behind any single slow disk touch. The docs are blunt that the transaction log wants its own dedicated device.

Semantics carry their own traps. Follower reads can be stale unless you pay for sync. Classic watches are one-shot and coalescing, so a fast sequence of changes delivers fewer events than changes, and KIP-500 calls out that the practical number of outstanding watches is limited for performance reasons. Hot znodes watched by thousands of clients recreate the herd effect at the notification layer. Operationally you are running a second distributed system with its own configs, security model, JVM tuning, and disk layout, which is precisely the cost Kafka decided to stop paying.

Do not use it as a general key-value store or message queue, for datasets beyond low gigabytes, for lock rates in the hundreds per second, or for cross-datacenter writes where quorum round trips dominate latency. If you are on Kubernetes you already operate etcd and rarely need a second coordinator. If you are deep in a cloud, managed options such as Parameter Store or App Configuration cover dynamic config without any ensemble to babysit. Reach for ZooKeeper, or a managed copy of it, when you are building infrastructure itself, a distributed queue, a task scheduler, a storage system, and need battle-tested membership, election, and metadata with strict ordering.

ZooKeeper next to etcd and Raft

ZAB and Raft solve the same problem with the same shape, a single elected leader, a replicated ordered log, majority quorums for both commit and election, and the rule that the next leader must hold every committed entry. ZAB predates Raft's 2014 publication and differs mainly in structure. It separates discovery, synchronization, and broadcast into explicit phases and marks leadership with a thirty-two bit epoch inside the zxid, while Raft folds election into the log with terms and randomized timeouts and keeps one integrated rule set. ZAB replicates idempotent state deltas computed by the leader, Raft replicates the commands themselves. In guarantees delivered to applications the two are interchangeable, which is why the migration conversation is about operations rather than correctness.

etcd is the closest living comparison. It replaces the tree with a flat MVCC keyspace where every revision is retained and queryable, so its watches replay history from any revision and, per the etcd docs, never silently drop events, where a classic ZooKeeper watch is a one-shot edge trigger. Sessions become leases, liveness objects decoupled from any single connection, a cleaner version of the ephemeral idea. The API is gRPC plus HTTP and JSON against ZooKeeper's custom Jute protocol, and etcd shipped dynamic membership reconfiguration years before ZooKeeper added it in 3.5. ZooKeeper still holds its own on hierarchical namespaces, mature recipes, and read fan-out through observers, and Consul competes as well while the etcd docs note its key-value store scales worse into millions of keys.

The clearest signal of the industry's direction is that new infrastructure tends to embed consensus rather than delegate it. Kafka built KRaft, ClickHouse replaced its ZooKeeper dependency with the Raft-based ClickHouse Keeper speaking the same client protocol, and CockroachDB, TiKV, and etcd itself all embed Raft directly. External coordination survives where many heterogeneous services must share one source of truth.

In production, and why Kafka left

The original paper documents production use inside Yahoo. The crawler's Fetching Service used ZooKeeper for master failover and fetcher health, and a single ZooKeeper server in that deployment handled bursts up to about two thousand operations per second in one-second samples, with read-to-write ratios between ten to one and one hundred to one during busy periods. Yahoo Message Broker kept per-topic primary and backup server assignments, group membership, and centralized shutdown switches in the tree. Outside Yahoo the same patterns run HBase region assignment and master election, SolrCloud cluster state, Hadoop NameNode HA failover, and BookKeeper ledger metadata under Pulsar, and ClickHouse required ZooKeeper for replication and distributed DDL until Keeper replaced it.

Kafka is the instructive exit. Its controller broker kept cluster metadata in ZooKeeper, and KIP-500 lists what broke at scale. Controller failover required reading all metadata back from ZooKeeper, a load linear in the number of partitions, and a controlled broker shutdown meant per-partition ZooKeeper writes that Confluent describes as taking seconds or more. The controller's in-memory state and the state in ZooKeeper could diverge, sometimes fixable only by restarting the controller, the controller could lag many seconds behind ISR changes made by partition leaders because watch counts are bounded, and every operator ran and secured two distributed systems with different tooling.

KRaft's answer, production-ready in Kafka 3.3 and mandatory since Kafka 4.0 dropped ZooKeeper entirely, treats metadata as an ordered event log replicated by a Raft quorum of controllers. Standby controllers already hold the latest state, so failover stops being proportional to partition count and becomes near-instantaneous, brokers fetch metadata deltas the same way consumers fetch records, and Confluent's stated ceiling moves to millions of partitions. The lesson generalizes cleanly, ZooKeeper's watch-and-notify model pushes edges while replicated logs ship the full ordered history, and systems whose metadata itself looks like a log eventually want the log.

Follow-up questions

• Why does the lock recipe watch the next-lowest znode instead of the lock holder? To avoid the herd effect. If all waiters watch the holder's znode, its deletion wakes every waiter at once and they stampede the ensemble with simultaneous getChildren calls right after a failure. Chaining watches to the immediate predecessor means each znode has exactly one watcher, so a release or crash wakes exactly one client, and grants stay fair in sequence order.
• How does ZooKeeper guarantee a new leader has every committed write? A write commits only after a majority of servers fsynced its proposal, and any two majorities of the same ensemble overlap in at least one server. Election picks the candidate with the highest zxid, which by the overlap argument must have seen every committed transaction, and the new leader discards uncommitted proposals from old epochs before serving, since without a quorum they could never have committed.
• Are ZooKeeper reads linearizable? No. Writes are linearizable because they all pass through the leader and a quorum, but reads are answered from the local replica's memory and can be stale if that follower lags. A client needing freshness calls sync first, which orders a marker through the leader's pipeline so the follower catches up before answering. That split is deliberate and is what lets read throughput scale with server count.
• What breaks if you treat a ZooKeeper lock as absolutely exclusive? A holder paused by a long GC or a network partition can keep acting after its session expired and the ensemble granted the lock to someone else. The lock is advisory, so the protected resource needs a fencing token it can check, and the monotonically increasing sequence number or zxid of the lock znode works, the resource rejects any request carrying a token older than the newest it has seen.
• Compare ZAB and Raft in a minute. Both are leader-based majority-quorum log replication with the invariant that the next leader holds all committed entries. ZAB splits discovery, sync, and broadcast into phases, stamps leadership with an epoch in the high bits of the zxid, and replicates idempotent state deltas computed by the leader. Raft integrates election into the log using terms and randomized timeouts and replicates commands directly. Application-visible guarantees are equivalent, the differences are protocol structure and operational packaging.
• Why did Kafka replace ZooKeeper with KRaft? Controller failover had to reload all metadata from ZooKeeper, linear in partition count, controlled shutdowns issued per-partition writes taking seconds or more, controller memory and ZooKeeper state could diverge, watch limits meant the controller learned about ISR changes many seconds late, and operators ran two distributed systems. KRaft stores metadata as a Raft-replicated event log inside Kafka, standbys already have current state so failover is near-instantaneous, and the design targets millions of partitions. Kafka 4.0 removed ZooKeeper entirely.

References

1. Hunt, Konar, Junqueira, Reed. ZooKeeper: Wait-free Coordination for Internet-scale Systems (USENIX ATC 2010), The original paper. Source for the coordination-kernel design, linearizable writes plus FIFO client order, the throughput tables, sub-two-hundred-millisecond elections, and the Yahoo Fetching Service and Message Broker workloads.
2. Apache ZooKeeper Overview (zookeeperOver), Official docs for znodes, ephemerals, one-shot watch semantics, the five consistency guarantees, session heartbeats, and ensemble roles.
3. Apache ZooKeeper Internals (ZAB), zxid layout with epoch and counter, leader activation and quorum sync, two-phase broadcast over FIFO TCP, and the quorum-overlap correctness argument.
4. Apache ZooKeeper Recipes, Canonical election, lock, shared lock, barrier, and queue recipes, including the herd effect and the watch-the-predecessor fix.
5. KIP-500: Replace ZooKeeper with a Self-Managed Metadata Quorum, Kafka's stated reasons for leaving, metadata divergence, watch limits, O(partitions) failover, and the Raft metadata log design. Confluent's KRaft explainer and 'Log of All Logs' post add the shutdown and scalability measurements.
6. etcd docs: Why etcd (comparison chart), etcd's side-by-side with ZooKeeper and Consul, MVCC model, non-dropping watches, leases versus sessions, gRPC versus Jute, and membership reconfiguration history.