Leader election in ZooKeeper

Olga Semme
NOTE

The concepts and the main components of ZooKeeper are described in the ZooKeeper architecture article.

ZooKeeper is a centralized service used to store information about services configuration.

The atomic messaging system at the heart of ZooKeeper is based on the principles of the consensus algorithm.

Consensus algorithm — a set of principles and rules, due to which all nodes participating in the cluster automatically come to a consensus on the current state of the network.

ZooKeeper implements one of the consensus algorithm protocols — ZAB (ZooKeeper Atomic BroadCast).

Consensus algorithm protocol

The ZAB protocol ensures that ZooKeeper replication occurs in order, and is also responsible for selecting lead nodes and recovering any failed nodes.

The main provisions of ZAB are described below.

Message exchange

In the messaging part, ZAB uses the following concepts:

  • Packet is a sequence of bytes sent over the FIFO channel.

  • Message is a sequence of bytes that will be atomically broadcast to all ZooKeeper servers. The message is placed in the offer and agreed upon before it is delivered.

  • Proposal is a unit of agreement. Proposal are agreed by exchanging packets when there is a quorum of ZooKeeper servers.

The proposal approval messaging model used by ZooKeeper is shown below.

Messaging model
Messaging model
Messaging model
Messaging model

To communicate with replicas, proposal approval uses a model in which each new record requires at least three transactions: a proposal (propose), an acknowledgment (ack), and a commit.

Message delivery principles

The ZAB protocol involves the creation of FIFO channels (in each session, the server executes client requests one by one in the order in which they are received) of the point-to-point type between servers. The TCP protocol is used for communication, providing the following properties:

  • Reliable delivery — if message m is delivered by one server, it will eventually be delivered by all servers.

  • Common order — if message a is delivered before message b by one server, then a will be delivered before b by all servers. If messages a and b are delivered, then either a will be delivered before b or b will be delivered before a.

  • Causal order — if message b is sent after message a was delivered by sender b, message a must be ordered before b. If the sender sends c after sending b, c must be ordered after b.

  • Ordered delivery — data is delivered in the same order as it was sent, and message m is only delivered after all messages sent before m have been delivered. The implication of this is that if the m message is lost, all messages after m will be lost.

  • No message after close — once a FIFO channel is closed, no messages will be received from it.

The use of timeouts ensures that consensus is reached in the presence of failures.

Controller epoch

Every cluster has one leader node and the remaining nodes are followers.

The ZooKeeper transaction ID — zxid — is used to guarantee the overall order of the offers.

Zxid consists of two parts: the epoch number and the transaction counter.

Zxid is a 64-bit number — the upper 32 bits for the epoch number and the lower 32 bits for the transaction counter.

A new epoch number represents a new leader. As a result of each transaction in the epoch of each leader, a unique zxid is assigned.

Below is the general sequence for a leader election.

Leader election in ZooKeeper
Leader election in ZooKeeper
Leader election in ZooKeeper
Leader election in ZooKeeper

  1. A proposal (propose) of a new epoch arrives from the server, and each proposal is assigned its own zxid.

  2. Every node acknowledges (ack) the offer only if it does not know any other leader with a higher epoch number in zxid, or if the epoch is the same, with a higher transaction counter. Before a leader can be chosen, it must collect votes from a quorum of nodes. Quorums to determine the leader must be (n/2+1), where n is the number of servers that make up the ZooKeeper service.

  3. After the quorum is confirmed, the new leader commits the creation of a new epoch and sets the next zxid to use.

  4. The follower will record the formation of a new epoch.

Each phase of the controller epoch is described below.

Controller epoch
Controller epoch
Controller epoch
Controller epoch

Each epoch consists of three phases, and each node can be in one of these three phases at any given time:

  • Leader election — leader election phase based on current quorum configurations. There can be at most one leader at any given time. The end of the phase comes after the confirmation of the new epoch by all followers.

  • Synchronization — synchronization phase, during which the new leader synchronizes the available replicas with the previous epoch and with all the followers as a leader. The end of the phase occurs after the quorum of followers has recognized that they are in sync with the leader.

  • Broadcast — translation phase, the normal mode of operation in which the leader continues to offer new client requests. The end of the phase occurs after the leader failure.

Found a mistake? Seleсt text and press Ctrl+Enter to report it