This documents the Zab protocol implemented by ZooKeeper 3.3.2 for atomic broadcast when ZooKeeper is not running standalone.

To be clear, the implementation cuts a couple of corners from the full Zab1.0 protocol. This has resulted in a couple of Jira's. The purpose of this document is to clearly specify the implemented protocol, so that we can do online upgrades of the cluster to the full protocol without bringing down the system.

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. See Zab1.0 for more details.

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
    long zxid;
    buffer data; // Only significant when type is request
    vector<org.apache.zookeeper.data.Id> authinfo;
}

There are a couple of different types of packets:

type

zxid

data

notation

meaning

FOLLOWERINFO(11)

last zxid accepted

sid (server id)

FOLLOWERINFO(lastZxid)

The follower has accepted up to lastZxid.

DIFF(13)

last committed zxid

n/a

DIFF(lastCommittedZxid)

lastCommittedZxid is the last zxid committed by the leader.

TRUNC(14)

truncZxid

n/a

TRUNC(truncZxid)

Truncate the history to truncZxid

SNAP(15)

n/a

n/a

SNAP

A state transfer (aka snapshot) will be sent to the follower. this will be a fuzzy state transfer that may include zxids being sent to the follower. The state transfer will immediately follow this packet.

OBSERVERINFO

last zxid accepted

n/a

OBSERVERINFO(lastZxid)

The observer has accepted up to lastZxid.

NEWLEADER(10)

lastZxid

n/a

NEWLEADER(lastZxid)

Accept this leader as the leader of the epoch of lastZxid.

UPTODATE(12)

n/a

n/a

UPTODATE

The follower is now uptodate enough to begin serving clients.

PROPOSAL(2)

zxid of proposal

proposed message

PROPOSAL(zxid, data)

Propose a message. (Request that it be accepted into a followers history.)

ACK(3)

zxid of proposal to ack

n/a

ACK(zxid)

Everything sent to the follower by the leader up to zxid has been accepted into its history (logged to disk).

COMMIT(4)

zxid of proposal to commit

n/a

COMMIT(zxid)

Everything in the followers history up to zxid should be committed (aka delivered).

INFORM(8)

zxid of proposal

data of proposal

INFORM(zxid, data)

Deliver the data. (Only used with observers.) i

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&2: Establish an epoch and sync with followers

In this phase an elected leader establishes leadership and syncs up with followers. It starts a new epoch and syncs with a quorum of followers to make sure that it has the most up to date 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. A process whose leader election instance indicates that it is the leader will
    1. the epoch, e, of its highest lastZxid;
    2. increment e;
    3. start accepting connections from followers;
  2. Followers open a TCP connection to a leader that their instance of the leader election indicates and send FOLLOWER(lastZxid).
  3. The leader does the following with each follower that connects:
    1. sends a NEWLEADER(lastZxid);
    2. 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.
    3. 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.
    4. The leader queues an UPTODATE packet.
    5. The leader releases any queued messages to the follower.
  4. When the follower receives the NEWLEADER(lastZxid) message, it makes sure that the follower has at least the last epoch in its history and sends ACK(lastZxid)
  5. Once the leader has received an acknowledgements from a quorum of followers, it takes leadership of epoch e.

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. the leader queues a packet PROPOSE(zxid, data) to all connected followers.
  2. a follower will log and sync proposals to disk and send ACK(zxid).
  3. when the leader receives acknowledgements from a quorum of followers it queues COMMIT(zxid) to all followers.