Leader Election

The Coordination Problem

In distributed systems, we often need a single coordinator to:

• Make decisions: Which node handles a request?
• Manage resources: Coordinate access to shared resources
• Maintain consistency: Ensure all nodes agree on state
• Handle failures: Detect and recover from node failures

The Challenge: How do multiple nodes agree on who the leader is? How do we handle leader failures? How do we prevent split-brain (multiple leaders)?

What is Leader Election?

Leader election is the process of selecting a single node as the coordinator (leader) in a distributed system. The leader handles coordination tasks, while other nodes (followers) follow the leader’s decisions.

Key Requirements

1. Safety: Only one leader at a time (no split-brain)
2. Liveness: Eventually a leader exists (even after failures)
3. Fault Tolerance: Handles node failures gracefully
4. Performance: Fast election, minimal overhead
5. Uniqueness: No conflicting leaders

Simple Analogy

Think of leader election like electing a class president:

• Multiple candidates (nodes) can run
• Election process determines winner (leader)
• If president is absent (leader fails), new election held
• Only one president at a time (safety)
• Eventually someone is elected (liveness)

Bully Algorithm

The Bully algorithm elects a leader based on node ID. The node with the highest ID wins.

How It Works:

1. When a node detects leader failure, it initiates election
2. Node sends election message to all nodes with higher IDs
3. If no response from higher nodes, node becomes leader
4. If response received, wait for leader announcement
5. Leader announces itself to all nodes

Advantages:

• Simple to understand and implement
• Fast election (O(n) messages)
• Deterministic (highest ID always wins)

Disadvantages:

• Can have multiple elections if multiple nodes detect failure simultaneously
• Requires all nodes to know all other nodes
• Not fault-tolerant if highest ID node is unstable

Raft Consensus Algorithm

Raft is a consensus algorithm that provides leader election and log replication. It’s more complex than Bully but provides stronger guarantees.

Raft States

Nodes in Raft can be in one of three states:

• Leader: Handles all client requests, replicates log to followers
• Follower: Receives log entries from leader, votes in elections
• Candidate: Campaigning to become leader

How Raft Election Works:

1. Follower doesn’t receive heartbeat from leader (timeout)
2. Follower becomes candidate and increments term
3. Candidate requests votes from all nodes
4. If candidate receives majority votes, becomes leader
5. Leader sends heartbeats to prevent new elections

Key Features:

• Majority voting: Prevents split-brain
• Terms: Each election has a term number (monotonically increasing)
• Log replication: Leader replicates log entries to followers
• Safety: Only one leader per term

Leader Election Implementation

💡 Tip: Click dropdown to switch between languages

Python

Python

"leader_election.py

import threading
import time
import random
from enum import Enum
from typing import List, Optional, Dict
from dataclasses import dataclass

class NodeState(Enum):
    FOLLOWER = "follower"
    CANDIDATE = "candidate"
    LEADER = "leader"

@dataclass
class VoteRequest:
    candidate_id: int
    term: int

@dataclass
class VoteResponse:
    voter_id: int
    term: int
    granted: bool

class RaftNode:
    """Raft node with leader election"""

    def __init__(self, node_id: int, all_nodes: List[int]):
        self.node_id = node_id
        self.all_nodes = all_nodes
        self.state = NodeState.FOLLOWER
        self.current_term = 0
        self.voted_for: Optional[int] = None
        self.leader_id: Optional[int] = None
        self.election_timeout = random.uniform(1.5, 3.0)  # Random timeout
        self.last_heartbeat = time.time()
        self.votes_received = 0
        self.lock = threading.Lock()

        # Start election timeout thread
        threading.Thread(target=self._election_timeout_loop, daemon=True).start()

    def _election_timeout_loop(self):
        """Check for election timeout"""
        while True:
            time.sleep(0.1)
            with self.lock:
                if self.state == NodeState.FOLLOWER:
                    if time.time() - self.last_heartbeat > self.election_timeout:
                        # Start election
                        self._start_election()

    def _start_election(self):
        """Start leader election"""
        print(f"Node {self.node_id}: Starting election for term {self.current_term + 1}")

        self.state = NodeState.CANDIDATE
        self.current_term += 1
        self.voted_for = self.node_id
        self.votes_received = 1  # Vote for self

        # Request votes from all other nodes
        for node_id in self.all_nodes:
            if node_id != self.node_id:
                # In real implementation, send RPC to other nodes
                # For demo, simulate vote response
                self._request_vote(node_id)

    def _request_vote(self, node_id: int):
        """Request vote from another node (simplified)"""
        # In real implementation, this would be an RPC call
        # For demo, simulate: higher ID nodes grant vote
        if node_id > self.node_id:
            # Simulate vote granted
            self._receive_vote(node_id, True)

    def _receive_vote(self, voter_id: int, granted: bool):
        """Receive vote response"""
        with self.lock:
            if granted:
                self.votes_received += 1
                majority = len(self.all_nodes) // 2 + 1

                if self.votes_received >= majority:
                    # Become leader
                    self._become_leader()

    def _become_leader(self):
        """Become leader"""
        print(f"Node {self.node_id}: Becoming leader for term {self.current_term}")
        self.state = NodeState.LEADER
        self.leader_id = self.node_id

        # Start sending heartbeats
        threading.Thread(target=self._send_heartbeats, daemon=True).start()

    def _send_heartbeats(self):
        """Send heartbeats to followers"""
        while self.state == NodeState.LEADER:
            time.sleep(0.5)  # Heartbeat interval
            # In real implementation, send heartbeat RPCs to all followers
            print(f"Node {self.node_id}: Sending heartbeats (term {self.current_term})")

    def receive_heartbeat(self, leader_id: int, term: int):
        """Receive heartbeat from leader"""
        with self.lock:
            if term >= self.current_term:
                self.current_term = term
                self.leader_id = leader_id
                self.state = NodeState.FOLLOWER
                self.last_heartbeat = time.time()
                self.voted_for = None

    def request_vote(self, candidate_id: int, term: int) -> bool:
        """Handle vote request from candidate"""
        with self.lock:
            if term > self.current_term:
                self.current_term = term
                self.voted_for = candidate_id
                self.state = NodeState.FOLLOWER
                return True
            elif term == self.current_term and self.voted_for is None:
                self.voted_for = candidate_id
                return True
            return False

# Usage (simplified demo)
nodes = [1, 2, 3]
raft_nodes = [RaftNode(node_id, nodes) for node_id in nodes]

# Simulate: Node 1 becomes leader
time.sleep(2)
print(f"Node 1 state: {raft_nodes[0].state}")

Java

"RaftNode.java

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

enum NodeState {
    FOLLOWER, CANDIDATE, LEADER
}

public class RaftNode {
    private final int nodeId;
    private final List<Integer> allNodes;
    private volatile NodeState state = NodeState.FOLLOWER;
    private final AtomicInteger currentTerm = new AtomicInteger(0);
    private volatile Integer votedFor = null;
    private volatile Integer leaderId = null;
    private volatile long lastHeartbeat = System.currentTimeMillis();
    private final AtomicInteger votesReceived = new AtomicInteger(0);
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    private final Random random = new Random();

    public RaftNode(int nodeId, List<Integer> allNodes) {
        this.nodeId = nodeId;
        this.allNodes = allNodes;

        // Start election timeout
        scheduler.scheduleAtFixedRate(this::checkElectionTimeout,
            1, 1, TimeUnit.SECONDS);
    }

    private void checkElectionTimeout() {
        if (state == NodeState.FOLLOWER) {
            long timeout = 1500 + random.nextInt(1500); // 1.5-3 seconds
            if (System.currentTimeMillis() - lastHeartbeat > timeout) {
                startElection();
            }
        }
    }

    private void startElection() {
        System.out.println("Node " + nodeId + ": Starting election for term " +
            (currentTerm.get() + 1));

        state = NodeState.CANDIDATE;
        currentTerm.incrementAndGet();
        votedFor = nodeId;
        votesReceived.set(1); // Vote for self

        // Request votes from all other nodes
        for (int nodeId : allNodes) {
            if (nodeId != this.nodeId) {
                requestVote(nodeId);
            }
        }
    }

    private void requestVote(int targetNodeId) {
        // In real implementation, send RPC to other nodes
        // For demo, simulate vote response
        if (targetNodeId > nodeId) {
            receiveVote(targetNodeId, true);
        }
    }

    private void receiveVote(int voterId, boolean granted) {
        if (granted) {
            int votes = votesReceived.incrementAndGet();
            int majority = allNodes.size() / 2 + 1;

            if (votes >= majority && state == NodeState.CANDIDATE) {
                becomeLeader();
            }
        }
    }

    private void becomeLeader() {
        System.out.println("Node " + nodeId + ": Becoming leader for term " +
            currentTerm.get());
        state = NodeState.LEADER;
        leaderId = nodeId;

        // Start sending heartbeats
        scheduler.scheduleAtFixedRate(this::sendHeartbeats,
            0, 500, TimeUnit.MILLISECONDS);
    }

    private void sendHeartbeats() {
        if (state == NodeState.LEADER) {
            System.out.println("Node " + nodeId + ": Sending heartbeats (term " +
                currentTerm.get() + ")");
            // In real implementation, send heartbeat RPCs to all followers
        }
    }

    public void receiveHeartbeat(int leaderId, int term) {
        synchronized (this) {
            if (term >= currentTerm.get()) {
                currentTerm.set(term);
                this.leaderId = leaderId;
                state = NodeState.FOLLOWER;
                lastHeartbeat = System.currentTimeMillis();
                votedFor = null;
            }
        }
    }

    public boolean requestVote(int candidateId, int term) {
        synchronized (this) {
            if (term > currentTerm.get()) {
                currentTerm.set(term);
                votedFor = candidateId;
                state = NodeState.FOLLOWER;
                return true;
            } else if (term == currentTerm.get() && votedFor == null) {
                votedFor = candidateId;
                return true;
            }
            return false;
        }
    }
}

// Usage
List<Integer> nodes = Arrays.asList(1, 2, 3);
RaftNode node1 = new RaftNode(1, nodes);
RaftNode node2 = new RaftNode(2, nodes);
RaftNode node3 = new RaftNode(3, nodes);

TypeScript

"leader-election.ts

enum NodeState {
    Follower = "follower",
    Candidate = "candidate",
    Leader = "leader"
}

class RaftNode {
    // Raft node with leader election
    private nodeId: number;
    private allNodes: number[];
    private state: NodeState = NodeState.Follower;
    private currentTerm: number = 0;
    private votedFor: number | null = null;
    private leaderId: number | null = null;
    private lastHeartbeat: number = Date.now();
    private votesReceived: number = 0;
    private electionTimeout: number;
    private heartbeatInterval: NodeJS.Timeout | null = null;
    private electionTimeoutInterval: NodeJS.Timeout | null = null;

    constructor(nodeId: number, allNodes: number[]) {
        this.nodeId = nodeId;
        this.allNodes = allNodes;
        this.electionTimeout = 1500 + Math.random() * 1500; // 1.5-3 seconds

        // Start election timeout check
        this.startElectionTimeoutCheck();
    }

    private startElectionTimeoutCheck(): void {
        // Check for election timeout
        this.electionTimeoutInterval = setInterval(() => {
            if (this.state === NodeState.Follower) {
                if (Date.now() - this.lastHeartbeat > this.electionTimeout) {
                    this.startElection();
                }
            }
        }, 100);
    }

    private startElection(): void {
        // Start leader election
        console.log(`Node ${this.nodeId}: Starting election for term ${this.currentTerm + 1}`);

        this.state = NodeState.Candidate;
        this.currentTerm++;
        this.votedFor = this.nodeId;
        this.votesReceived = 1; // Vote for self

        // Request votes from all other nodes
        for (const nodeId of this.allNodes) {
            if (nodeId !== this.nodeId) {
                this.requestVote(nodeId);
            }
        }
    }

    private requestVote(targetNodeId: number): void {
        // In real implementation, send RPC to other nodes
        // For demo, simulate vote response
        if (targetNodeId > this.nodeId) {
            this.receiveVote(targetNodeId, true);
        }
    }

    private receiveVote(voterId: number, granted: boolean): void {
        if (granted) {
            this.votesReceived++;
            const majority = Math.floor(this.allNodes.length / 2) + 1;

            if (this.votesReceived >= majority && this.state === NodeState.Candidate) {
                this.becomeLeader();
            }
        }
    }

    private becomeLeader(): void {
        // Become leader
        console.log(`Node ${this.nodeId}: Becoming leader for term ${this.currentTerm}`);
        this.state = NodeState.Leader;
        this.leaderId = this.nodeId;

        // Start sending heartbeats
        this.heartbeatInterval = setInterval(() => {
            if (this.state === NodeState.Leader) {
                console.log(`Node ${this.nodeId}: Sending heartbeats (term ${this.currentTerm})`);
                // In real implementation, send heartbeat RPCs to all followers
            }
        }, 500);
    }

    receiveHeartbeat(leaderId: number, term: number): void {
        // Receive heartbeat from leader
        if (term >= this.currentTerm) {
            this.currentTerm = term;
            this.leaderId = leaderId;
            this.state = NodeState.Follower;
            this.lastHeartbeat = Date.now();
            this.votedFor = null;
        }
    }

    requestVoteFrom(candidateId: number, term: number): boolean {
        // Handle vote request from candidate
        if (term > this.currentTerm) {
            this.currentTerm = term;
            this.votedFor = candidateId;
            this.state = NodeState.Follower;
            return true;
        } else if (term === this.currentTerm && this.votedFor === null) {
            this.votedFor = candidateId;
            return true;
        }
        return false;
    }
}

// Usage
const nodes = [1, 2, 3];
const node1 = new RaftNode(1, nodes);
const node2 = new RaftNode(2, nodes);
const node3 = new RaftNode(3, nodes);

C++

"leader_election.cpp

#include <iostream>
#include <vector>
#include <thread>
#include <chrono>
#include <atomic>
#include <mutex>
#include <random>

enum class NodeState {
    Follower,
    Candidate,
    Leader
};

class RaftNode {
private:
    int nodeId;
    std::vector<int> allNodes;
    std::atomic<NodeState> state{NodeState::Follower};
    std::atomic<int> currentTerm{0};
    int votedFor = -1;
    int leaderId = -1;
    std::atomic<long> lastHeartbeat{0};
    std::atomic<int> votesReceived{0};
    std::mutex mtx;
    std::random_device rd;
    std::mt19937 gen{rd()};

    void checkElectionTimeout() {
        while (true) {
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            if (state == NodeState::Follower) {
                long timeout = 1500 + (gen() % 1500); // 1.5-3 seconds
                if (std::chrono::duration_cast<std::chrono::milliseconds>(
                    std::chrono::steady_clock::now().time_since_epoch()).count() -
                    lastHeartbeat > timeout) {
                    startElection();
                }
            }
        }
    }

    void startElection() {
        std::lock_guard<std::mutex> lock(mtx);
        std::cout << "Node " << nodeId << ": Starting election for term "
                  << (currentTerm + 1) << std::endl;

        state = NodeState::Candidate;
        currentTerm++;
        votedFor = nodeId;
        votesReceived = 1; // Vote for self

        // Request votes from all other nodes
        for (int nodeId : allNodes) {
            if (nodeId != this->nodeId) {
                requestVote(nodeId);
            }
        }
    }

    void requestVote(int targetNodeId) {
        // In real implementation, send RPC to other nodes
        // For demo, simulate vote response
        if (targetNodeId > nodeId) {
            receiveVote(targetNodeId, true);
        }
    }

    void receiveVote(int voterId, bool granted) {
        if (granted) {
            int votes = ++votesReceived;
            int majority = allNodes.size() / 2 + 1;

            if (votes >= majority && state == NodeState::Candidate) {
                becomeLeader();
            }
        }
    }

    void becomeLeader() {
        std::cout << "Node " << nodeId << ": Becoming leader for term "
                  << currentTerm << std::endl;
        state = NodeState::Leader;
        leaderId = nodeId;

        // Start sending heartbeats
        std::thread([this]() {
            while (state == NodeState::Leader) {
                std::this_thread::sleep_for(std::chrono::milliseconds(500));
                std::cout << "Node " << nodeId << ": Sending heartbeats (term "
                          << currentTerm << ")" << std::endl;
            }
        }).detach();
    }

public:
    RaftNode(int nodeId, std::vector<int> allNodes)
        : nodeId(nodeId), allNodes(allNodes) {
        lastHeartbeat = std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::steady_clock::now().time_since_epoch()).count();

        // Start election timeout thread
        std::thread(&RaftNode::checkElectionTimeout, this).detach();
    }

    void receiveHeartbeat(int leaderId, int term) {
        std::lock_guard<std::mutex> lock(mtx);
        if (term >= currentTerm) {
            currentTerm = term;
            this->leaderId = leaderId;
            state = NodeState::Follower;
            lastHeartbeat = std::chrono::duration_cast<std::chrono::milliseconds>(
                std::chrono::steady_clock::now().time_since_epoch()).count();
            votedFor = -1;
        }
    }

    bool requestVoteFrom(int candidateId, int term) {
        std::lock_guard<std::mutex> lock(mtx);
        if (term > currentTerm) {
            currentTerm = term;
            votedFor = candidateId;
            state = NodeState::Follower;
            return true;
        } else if (term == currentTerm && votedFor == -1) {
            votedFor = candidateId;
            return true;
        }
        return false;
    }
};

// Usage
int main() {
    std::vector<int> nodes = {1, 2, 3};
    RaftNode node1(1, nodes);
    RaftNode node2(2, nodes);
    RaftNode node3(3, nodes);

    std::this_thread::sleep_for(std::chrono::seconds(2));
    return 0;
}

C#

RaftNode.cs

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

public enum NodeState {
    Follower,
    Candidate,
    Leader
}

public class RaftNode {
    // Raft node with leader election
    private readonly int nodeId;
    private readonly List<int> allNodes;
    private NodeState state = NodeState.Follower;
    private int currentTerm = 0;
    private int? votedFor = null;
    private int? leaderId = null;
    private long lastHeartbeat = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
    private int votesReceived = 0;
    private readonly object lockObject = new object();
    private readonly Random random = new Random();
    private Timer electionTimeoutTimer;
    private Timer heartbeatTimer;

    public RaftNode(int nodeId, List<int> allNodes) {
        this.nodeId = nodeId;
        this.allNodes = allNodes;

        // Start election timeout check
        electionTimeoutTimer = new Timer(CheckElectionTimeout, null,
            TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(1));
    }

    private void CheckElectionTimeout(object state) {
        lock (lockObject) {
            if (this.state == NodeState.Follower) {
                long timeout = 1500 + random.Next(1500); // 1.5-3 seconds
                if (DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() - lastHeartbeat > timeout) {
                    StartElection();
                }
            }
        }
    }

    private void StartElection() {
        lock (lockObject) {
            Console.WriteLine($"Node {nodeId}: Starting election for term {currentTerm + 1}");

            this.state = NodeState.Candidate;
            currentTerm++;
            votedFor = nodeId;
            votesReceived = 1; // Vote for self

            // Request votes from all other nodes
            foreach (int targetNodeId in allNodes) {
                if (targetNodeId != nodeId) {
                    RequestVote(targetNodeId);
                }
            }
        }
    }

    private void RequestVote(int targetNodeId) {
        // In real implementation, send RPC to other nodes
        // For demo, simulate vote response
        if (targetNodeId > nodeId) {
            ReceiveVote(targetNodeId, true);
        }
    }

    private void ReceiveVote(int voterId, bool granted) {
        lock (lockObject) {
            if (granted) {
                votesReceived++;
                int majority = allNodes.Count / 2 + 1;

                if (votesReceived >= majority && state == NodeState.Candidate) {
                    BecomeLeader();
                }
            }
        }
    }

    private void BecomeLeader() {
        Console.WriteLine($"Node {nodeId}: Becoming leader for term {currentTerm}");
        state = NodeState.Leader;
        leaderId = nodeId;

        // Start sending heartbeats
        heartbeatTimer = new Timer(_ => {
            if (state == NodeState.Leader) {
                Console.WriteLine($"Node {nodeId}: Sending heartbeats (term {currentTerm})");
                // In real implementation, send heartbeat RPCs to all followers
            }
        }, null, TimeSpan.Zero, TimeSpan.FromMilliseconds(500));
    }

    public void ReceiveHeartbeat(int leaderId, int term) {
        lock (lockObject) {
            if (term >= currentTerm) {
                currentTerm = term;
                this.leaderId = leaderId;
                state = NodeState.Follower;
                lastHeartbeat = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
                votedFor = null;
            }
        }
    }

    public bool RequestVoteFrom(int candidateId, int term) {
        lock (lockObject) {
            if (term > currentTerm) {
                currentTerm = term;
                votedFor = candidateId;
                state = NodeState.Follower;
                return true;
            } else if (term == currentTerm && votedFor == null) {
                votedFor = candidateId;
                return true;
            }
            return false;
        }
    }
}

// Usage
var nodes = new List<int> { 1, 2, 3 };
var node1 = new RaftNode(1, nodes);
var node2 = new RaftNode(2, nodes);
var node3 = new RaftNode(3, nodes);

ZooKeeper-Based Election

Apache ZooKeeper provides built-in support for leader election using ephemeral sequential nodes:

1. All nodes create ephemeral sequential nodes under /election
2. Node with smallest sequence number becomes leader
3. If leader fails, ephemeral node deleted, next node becomes leader
4. Nodes watch the node with sequence number one less than theirs

Advantages:

• Handled by ZooKeeper (no custom implementation needed)
• Automatic failure detection
• No split-brain (ZooKeeper provides consistency)

Challenges and Solutions

Split-Brain

Problem: Network partition causes multiple leaders.

Solution: Require majority vote (quorum). Only partition with majority can elect leader.

Election Storms

Problem: Multiple nodes start election simultaneously.

Solution: Use random election timeout. Reduces probability of simultaneous elections.

Leader Failure Detection

Problem: How to detect leader failure quickly?

Solution: Use heartbeat mechanism. If no heartbeat received within timeout, assume leader failed.

Use Cases

Distributed Databases

Leader handles writes, replicates to followers. Ensures consistency and handles failures.

Coordination Services

Leader coordinates distributed operations, manages shared state.

Replicated Systems

Leader makes decisions, followers replicate state. Provides fault tolerance.

Trade-offs

Advantages:

• Provides single coordinator
• Handles failures gracefully
• Prevents split-brain (with majority voting)
• Enables coordination

Disadvantages:

• Adds complexity
• Leader can become bottleneck
• Network partitions can cause issues
• Requires majority for election

Key Takeaways

Single Coordinator

Leader election ensures only one node acts as coordinator. Provides mutual exclusion for coordination tasks.

Majority Voting

Require majority votes to prevent split-brain. Only partition with majority can elect leader.

Fault Tolerance

Leader failures trigger new elections. System eventually elects new leader. Handles node failures gracefully.

Heartbeat Mechanism

Leader sends heartbeats to prevent elections. Followers detect leader failure via timeout. Triggers new election.

---

Next Steps

• Learn Distributed Locks - another coordination mechanism
• Master Consistency Models - leader election ensures consistency
• Understand Replication Strategies - leader-based replication