Zab1.0

This documents the Zab, an atomic broadcast protocol used by ZooKeeper to propagate state changes.

Zab at a high level is a leader based protocol similar to Paxos. Some of the aspects that distinguish Zab from Paxos is that Zab recovers histories rather than single instances of protocols; Zab has prefix ordering properties to allow primary/backup implementations to have multiple outstanding operations outstanding at a time; and Zab does special processing at leadership change to guarantee unique sequence numbers for values proposed by a leader.

Apart from the state transfer, all communication between follower and leader is done using a single data structure:

class QuorumPacket {
    int type; // Request, Ack, Commit, Ping, etc
    long zxid;
    buffer data;
    vector<org.apache.zookeeper.data.Id> authinfo; // only used for requests
}

The following structure is used the SERVERINFO packet:

class Info {
    long sid; // server id
    int version; // protocol version
    int currentEpoch;
    int acceptedEpoch;
}

There are a couple of different types of packets:

Zab servers have the following state:

Implementation assumptions

Implementations must conform to the following:

1. Use a best-effort leader election algorithm that will elect a leader with the latest history from a quorum of servers.
2. An observer or follower will only connect to a single leader at a time.
3. Servers must process packets in the order that they are received. Since TCP maintains ordering when sending packets, this means that packets will be processed in the order defined by the sender.

Leader election

All servers start off looking for a leader. Once the instance of leader election at a given server indicates a leader has emerged it will move to phase 1. If the leader election instance indicates that the server is the leader, it moves to phase 1 as a leader, otherwise it moves to phase 1 as a follower.

Phase 1: Establish an epoch

In this phase an elected leader makes sure that previous leaders cannot commit new proposals and decides on an initial history. (Note a leader is also considered a follower of itself.)

Any failures or timeouts will cause the server to go back to leader election.

1. l The leader starts accepting connections from followers.
2. f Followers connect the the leader and send SERVERINFO(lastZxid, info).
3. l The leader sends SERVERINFO(lastZxid, info) to followers that connect.
4. l Once the leader has quorum, it stops accepting connections, and sends NEWEPOCH(e) to all followers, where e is greater than all f.acceptedEpoch in the quorum.
5. f Followers send ACK(e<<32) if e > f.acceptedEpoch. (If e < f.acceptedEpoch, the follower will abandon the leader. If e == f.acceptedEpoch, the follower will not send ACK(e<<32), but will still maintain the connection to the leader.)
6. l The leader waits for all followers to ack NEWEPOCH(e)
7. l If the following conditions are not met for all connected followers, the leader disconnects followers and goes back to leader election:
   • f.currentEpoch <= l.currentEpoch
   • if f.currentEpoch == l.currentEpoch, then f.lastZxid <= l.lastZxid

Note, if a follower is connecting, but the leader is already established (in phase 3) the follower follows the phases, but the leader ignores any ACKs.

There are two noticeable differences between this description of phase 1 and the one found in http://research.yahoo.com/files/YL-2010-007.pdf:

1. the leader does not sync with the most up-to-date follower
2. the followers do not send their histories in the ACK of the NEWEPOCH message.

The leader election protocols that we use try to guarantee that the elected leader has the most up-to-date history from a quorum of processes. It also tries to guarantee that there is only one leader elected. It is possible for these guarantees not to hold due to race conditions, stale information, or perhaps even implementation error, so we make sure that the guarantees are satisfied in phase 1. In the tech report we do not assume that the elected leader has the most up-to-date history, so we must find the most up-to-date follower and sync with it. However since the elected leader should be the most up-to-date process, using the leader election protocols that we use, we just check the condition in step 7. If the leader meets the conditions of step 7 we are the most up-to-date, so we have implicitly synced with the most up-to-date process in the quorum. If it doesn't, we go back to leader election.

In this description and in our implementation we use the notion of connections which map to TCP connections. A follower will only be connected to a single leader at a time. Also, a follower only changes its history under the direction of a leader. The phase 1 described in the protocol, sends its history in the ACK to NEWEPOCH, but we do not send the history in step 5. Since we just need to verify that the leader has a super set of history of the follower, we only need to know the lastZxid in the history. The lastZxid is sent in step 2 in the protocol described here rather than step 5. However if the follower executes step 5 and sends the ACK to the leader, it means that for the whole time period between step 2 and 5 it has been connected to the leader. Since the follow can only be connected to one leader at a time, and since it has been connect to the same leader from step 2 to 5, and since that leader has not asked the follower to change its history, then the lastZxid sent in step 2 is still the same lastZxid in step 5, so there is no reason to resend it.

Another difference from the protocol described in the tech report is the extra information sent in SERVERINFO and the SERVERINFO sent by the leader in step 3. The extra information is for implementation purposes: making sure that the protocol version is compatible between client and server, and information for debugging purposes.

Phase 2: Sync with followers

1. l The leader does the following with each follower connected to it:
   1. adds the follower to the list of connections to send new proposals, so while the server is performing the next steps, it is queuing up any new proposals sent to the follower.
   2. does one of the following:
      • SNAP if the follower is so far behind that it is better to do a state transfer than send missing transactions.
      • TRUNC(zxid) if the follower has transactions that the leader has chosen to skip. The leader sets zxid to the last zxid in its history for the epoch of the follower.
      • DIFF if the leader is sending transactions that the follower is missing. The leader sends missing messages to the follower.
   3. sends a NEWLEADER(e);
   4. The leader releases any queued messages to the follower.
2. f The follower syncs with the leader, but doesn't modify its state until it receives the NEWLEADER(e) packet. Once it does it applies any changes and sends ACK(e << 32).
3. l Once the leader has received an acknowledgements from a quorum of followers, it takes leadership of epoch e and queues UPTODATE to all followers.
4. f When a follower receives the UPDATE message, it starts accepting client connections and serving new state.
5. l Leader starts accepting connections from followers again.

Phase 3: Broadcast

The leader and followers can have multiple proposals in process and at various stages in the pipeline. The leader will remain active as long as there is a quorum of followers acknowledging its proposals or pings within a timeout interval. The follower will continue to support a leader as long as it receives proposals or pings within a timeout interval. Any failures or timeouts will result in the server going back to leader election.

1. l the leader queues a packet PROPOSE(zxid, data) to all connected followers.
2. f a follower will log and sync proposals to disk and send ACK(zxid).
3. l when the leader receives acknowledgements from a quorum of followers it queues COMMIT(zxid) to all followers.