ZooKeeper Example

The following code excerpt shows how to use ZooKeeper to implement a "barrier." A barrier separates a process into two logical halves. Multiple machines running in coordination with one another will all perform the first half of the process. No machine can begin the second half of the process until everyone has completed the first half. The barrier sits between these processes. As nodes reach the barrier, they all wait until everyone has reached the barrier. Then all nodes are released to begin the second half. A distributed barrier implementation written for ZooKeeper follows:

Watcher watcher = new Watcher() {
public void process(WatchEvent event) {}
};

ZooKeeper zk = new ZooKeeper(hosts, 3000, watcher);

Object notifyObject = new Object();
String root;
int size;

Barrier(ZooKeeper zk, String name, int size) throws KeeperException, InterruptedException {
this.zk = zk;
this.root = name;
this.size = size;
// Make sure the barrier node exists
try {
zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE, 0);
} catch (NodeExistsException e) {}
}


b.enter()
/** work with everyone **/
b.leave()



/**
* Join barrier
* @return
* @throws KeeperException
* @throws InterruptedException */
boolean enter() throws KeeperException, InterruptedException {
zk.create(root + "/" + name, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateFlags.EPHEMERAL);
while (true) {
synchronized (notifyObject) {
ArrayList list = zk.getChildren(root, new Watcher() {
public void process(Event e) { notifyObject.notifyAll(); }
});

if (list.size() < size) {
notifyObject.wait();
} else {
return true;
}
}
}
}

/**
* Wait until all reach barrier
* @return
* @throws KeeperException
* @throws InterruptedException */
boolean leave() throws KeeperException, InterruptedException {
zk.delete(root + "/" + name, 0);
while (true) {
synchronized (notifyObject) {
ArrayList list = zk.getChildren(root, new Watcher() {
public void process(Event e) { notifyObject.notifyAll(); }
});

if (list.size() > 0) {
notifyObject.wait();
} else {
return true;
}
}
}
}

Distributed Consensus

A reasonable question is how the ZooKeeper service can function across multiple nodes and remain synchronized. If distributed synchronization is why your services must use ZooKeeper, how does ZooKeeper itself bootstrap this capability?

ZooKeeper implements a distributed consensus protocol. ZooKeeper internally uses a leader election protocol such as Paxos to determine which node in the ZooKeeper service is the master. While clients connect to any node in the ZooKeeper service, these additional nodes will forward agreed-upon facts back to clients. Updates to the shared state require the intervention of the master. All updates to the shared state are ordered with timestamps. These timestamped updates are then disseminated to the nodes in the ZooKeeper service. When a majority of nodes acknowledge an update, it is said to be held by a quorum of the nodes. Any fact that a quorum has agreed upon may be returned to clients. Conversely, any updates that have not reached a quorum will not be returned to the clients. The timestamps are used to order the updates to elements of the data store. If multiple updates are made to the state of an individual node, the newest update is used.

The use of a quorum ensures that the service always returns consistent answers. Because a vote is effectively held before returning a response, any nodes which hold stale data will be outvoted by the nodes with more current information. This also makes ZooKeeper resilient to failure. Up to 49% of the ZooKeeper service nodes can shut down or become desynchronized before ZooKeeper loses its ability to authoritatively answer responses. So if 11 nodes run ZooKeeper, up to 5 of these may disconnect without incident. After more than half the nodes fail, ZooKeeper will refuse service until the machines are restored.

If the node of the ZooKeeper cluster which was elected leader fails, then a new leader election will be held and the cluster will continue to function.

The reason for electing a leader in such a system is to ensure that timestamps assigned to updates are only issued by a single authority. ZooKeeper is designed to reduce or eliminate possible race conditions in distributed applications.

One consequence of ZooKeeper's design is that it is intended to serve many more read requests than writes. A ZooKeeper cluster can handle hundreds or thousands of clients, issuing many tens of thousands of requests per second--if the majority of these requests (90--99%) are reads. Only a small fraction should be updates.