What is a consensus in a distributed system?

In the context of distributed systems design, a consensus is often loosely used to mean some form of agreement. Consensus involves multiple servers agreeing on values. Once they reach a decision on a value, that decision is final. Typical consensus algorithms make progress when any majority of their servers is available; for example, a cluster of 5 servers can continue to operate even if 2 servers fail. If more servers fail, they stop making progress (but will never return an incorrect result).

i.e 2f+1 nodes to survive f failed nodes

There are a few properties we expect from a solution to consensus:
Agreement: Every correct process must agree on the same value.
Validity: If all processes propose the same value v, then all correct processes decide v
Termination: Every correct process decides some value. If the protocol never terminates, then the processes are vacuously agreeing on the same thing, which is not deciding.

To summarize, fundamentally, the goal of consensus is not that of the negotiation of an optimal value of some kind, but just the collective agreement on some value that was previously proposed by one of the participating servers in that round of the consensus algorithm. With the help of consensus, the distributed system is made to act as though it were a single entity.

An example scenario:

For the purpose of simplicity, let’s assume a distributed storage system with 2f+1 nodes participating to form a cluster and these participants act at their own speed, may fail at any time and rejoin after recovering from the failure. And these nodes are connected via a network which transmits messages asynchronously at an arbitrary speed. In short, everything can fail at any time; after failure, participants can recover and rejoin the system. Yes, we are looking at a fault tolerant storage system. As these nodes can fail at various stages, it’s important to have more than one copy of our data. For now, let’s assume all the data is replicated across all the cluster nodes (but in reality it may affect overall performance)
And we have a client which is not part of the cluster, requesting for some operation from our distributed storage, like a write or read to a data file. Read operation can be served by any node in our cluster, but write has to be handled by one node at a time to avoid any kind of conflicts. Also, read of the same data which is being written can be served only after the write has been committed. And no other client can request write to the same data file when a write is in progress. In order to achieve this data integrity in our distributed storage, a member of the cluster should be elected as a leader to accept writes or redirect the read requests to other cluster members. Wait, what if the leader which receives the write requests loses its leadership? So the node which is handling the write should lock the data file until the write is committed. But how can we achieve this? How will the cluster know it’s membership?

What is Apache Zookeeper?

ZooKeeper is a distributed, open-source coordination service for distributed applications. By providing a robust implementation of a few basic operations, ZooKeeper simplifies the implementation of many advanced patterns in distributed systems.
The service itself is simple, replicated, distributed, fast, highly reliable and exposes a file-system-like API so that clients can manipulate simple data files called znodes. These znodes can have some special properties like being ephemeral or sequential, but nonetheless, they are small data files.
Ephemeral: Clients create such znodes, and they either delete them explicitly or let the system remove them automatically when the session that creates them terminates.
Sequential: These Znodes when created, gets a unique number (sequence) suffixed to its name.
Unlike standard file systems, each znode in a ZooKeeper namespace can have data associated with it as well as children.
Fig 1 shows the high-level components of the ZooKeeper service. With the exception of the request processor, each of the servers that make up the ZooKeeper service replicates its own copy of each of the components.

All requests that update ZooKeeper state are forwarded to the leader. The leader executes the request and broadcasts the change to the ZooKeeper state using a Zookeeper Atomic Broadcast (ZAB) protocol.
Many prominent Apache projects like Hadoop, HBase, Kafka are using Zookeeper to solve their distributed complexities.

How can we use it?

Coming back to our distributed storage cluster, we can use Zookeeper to elect a leader using sequential znodes and create a write lock using Ephemeral znodes.

Leader Election
A simple way of doing leader election with ZooKeeper is to use the SEQUENCE|EPHEMERAL flags when creating znodes that represent “proposals” of clients. The idea is to have a znode, say “/election”, such that each znode creates a child znode “/election/n_” with both flags SEQUENCE|EPHEMERAL. With the sequence flag, ZooKeeper automatically appends a sequence number that is greater than anyone previously appended to a child of “/election”. The process that created the znode with the smallest appended sequence number is the leader.
That’s not all, though. It is important to watch for failures of the leader so that a new client emerges as the new leader in the case the current leader fails. A trivial solution is to have all application processes watching upon the current smallest znode, and checking if they are the new leader when the smallest znode goes away (note that the smallest znode will go away if the leader fails because the node is ephemeral). But this causes a herd effect: upon of failure of the current leader, all other processes receive a notification, and execute getChildren on “/election” to obtain the current list of children of “/election”. If the number of clients is large, it causes a spike in the number of operations that ZooKeeper servers have to process. To avoid the herd effect, it is sufficient to watch for the next znode down on the sequence of znodes. If a client receives a notification that the znode it is watching is gone, then it becomes the new leader in the case that there is no smaller znode. Note that this avoids the herd effect by not having all clients watching the same znode. Here’s the pseudo code:
Let ELECTION be a path of choice of the application. To volunteer to be a leader:
Create znode z with path “ELECTION/n_” with both SEQUENCE and EPHEMERAL flags;
Let C be the children of “ELECTION”, and i be the sequence number of z;
Watch for changes on “ELECTION/n_j”, where j is the largest sequence number such that j < i and n_j is a znode in C;
Upon receiving a notification of znode deletions:
Let C be the new set of children of ELECTION;
If z is the smallest node in C, then execute leader procedure;
Otherwise, watch for changes on “ELECTION/n_j”, where j is the largest sequence number such that j < i and n_j is a znode in C;
Locks
Fully distributed locks that are globally synchronous, meaning at any snapshot in time no two clients think they hold the same lock. These can be implemented using ZooKeeper. Clients wishing to obtain a lock do the following:
Call create() with a pathname of “locknode/lock-” and the sequence and ephemeral flags set.
Call getChildren() on the lock node without setting the watch flag (this is important to avoid the herd effect).
If the pathname created in step 1 has the lowest sequence number suffix, the client has the lock and the client exits the protocol.
The client calls exists() with the watch flag set on the path in the lock directory with the next lowest sequence number.
if exists() returns false, go to step 2. Otherwise, wait for a notification for the pathname from the previous step before going to step 2.
The unlock protocol is very simple: clients wishing to release a lock simply delete the node they created in step 1.

The example discussed here is a simple distributed system for the purpose of this presentation only but nevertheless, Apache Zookeeper can be used to solve many distributed system complexities.

Rootconf Hyderabad edition