Papers
ZooKeeper: Wait-free Coordination for Internet-scale Systems
Patrick Hunt, Mahadev Konar, Flavio P. Junqueira, Benjamin Reed · USENIX Annual Technical Conference (ATC) 2010, 2010
The problem: every distributed application needs coordination, and each one reinvented it
Large-scale distributed applications need coordination: shared configuration, leader election, knowing which processes are alive (membership), locks and barriers. Before ZooKeeper, the common approach was a specific service for each need, or a lock service (the reference was Google's Chubby), in which the server implements the ready-made primitives. The authors, from Yahoo!, chose another path: instead of building operations like lock or election into the server, offer a simple, high-performance "kernel" with a small API, so that clients themselves build the primitives they need. The service combines ideas from group messaging, shared registers and distributed locks in a replicated, centralized service. The central promise, which gives the paper its name, is "wait-free" access: no client request is blocked waiting for another client (for example, waiting for a lock to be released inside the server). The authors consider this property important for performance and fault tolerance, but not sufficient for coordination: that is why it comes with well-chosen ordering guarantees.
The data model: znodes, ephemeral nodes, sequential flag, and watches
ZooKeeper exposes a hierarchical namespace similar to a file system. Each node, called a znode, holds a small amount of coordination data (metadata, not application data) and can also have children. There are two kinds of znode. Regular ones are created and deleted explicitly by clients. Ephemeral ones are removed automatically when the session of the client that created them ends, which makes them ideal for detecting failures and representing process presence. In addition, when creating either kind the client can turn on the sequential flag: the node name gets an increasing counter appended, which provides ordering and uniqueness among several clients creating nodes under the same parent. Updates can be conditioned on the znode's version. To avoid polling, the client can register a watch: a one-time notification, triggered when the znode changes; once triggered, the client must register it again if it wants to keep following it. The client maintains a session, whose state is distributed, and can therefore reconnect to another server after a failure. The API is asynchronous, allowing a client to have several operations pending at the same time.
Guarantees and trade-off: linearizable writes, per-client FIFO order, and local reads
The core of the paper's trade-off lies in two guarantees: requests from the same client are executed in FIFO order, and requests that change state are linearizable. The authors argue that this pair is enough to build the coordination primitives applications need and allows an efficient implementation. Because writes go through a leader-based atomic broadcast protocol, all servers apply them in the same total order. Reads, on the other hand, are served by the local replica the client is connected to, without going through the agreement protocol; so they are not ordered with respect to other clients' writes and may return stale data. For those who need an up-to-date value, there is the sync operation. Watches close the gap in another way: a client that registered a watch never sees the new value before receiving the change notification. The cost is clear: reads are fast and scale with the number of replicas, while writes are slower because they require agreement among servers and durable persistence. The service remains available as long as a majority (quorum) of the servers is working.
Primitives built on top of the kernel: configuration, membership, locks, and election
The strength of the design shows in the recipes that clients build with znodes and watches. To manage configuration, the value is stored in a znode, read with a watch, and read again when the notification arrives. For group membership, each process creates an ephemeral child under the group's znode: listing the children shows who is alive, and a watch on the parent signals joins and leaves, including when a process dies, because its session expires and the node disappears. For locks, each client creates an ephemeral, sequential znode under the lock znode. Whoever has the lowest number holds the lock; the others watch only the node immediately before their own. This way, when the lock is released, only one client wakes up, avoiding the "herd effect", in which many clients compete at the same time and overload the server. The Apache ZooKeeper recipes documentation describes leader election with the same pattern: the lowest number wins and each candidate watches its predecessor. Barriers can also be built with znodes and watches. None of this is a server operation: ZooKeeper only provides the basic building blocks.
Implementation, performance, and legacy
Each server keeps a copy of the database (the tree of znodes) in memory. Before applying an update, the server logs it and forces the write to disk. For recovery, 'fuzzy' snapshots are taken: the server walks the tree without locking the state, so the snapshot may not correspond to any single instant; this remains correct because state changes are idempotent and are reapplied in order. According to the paper, ZooKeeper reaches tens to hundreds of thousands of transactions per second on workloads with read-to-write ratios from 2:1 to 100:1, which allows heavy use and not just coarse-grained locks. The project's current documentation states that it performs best on read-dominated workloads, around 10:1, which is typical of a coordination service. The design became the Apache ZooKeeper project, used by systems such as Hadoop, HBase, Flink, Pulsar, Solr, and Spark (and, historically, by Kafka before KRaft) for leader election, failure detection, service discovery, and configuration management.
See the ideas at work
Practice the designs that use this paper in the System Design school.