Leader Election

We've explored the theoretical foundations of leader election—Bully, Ring, and Lease-based algorithms—but theory only takes us so far. Production systems face constraints that textbooks rarely address: legacy compatibility, operational complexity, failure correlation, and the cold reality that any system you deploy will eventually fail in ways you didn't anticipate.

This page bridges theory and practice by examining how real-world systems implement leader election. From PostgreSQL's streaming replication to Kubernetes' controller architecture, from Kafka's partition leadership to etcd's Raft consensus, we'll see how production systems adapt, extend, and sometimes deviate from theoretical ideals.

More importantly, we'll extract practical lessons about choosing and operating leader election mechanisms—lessons that only emerge from running these systems at scale.

Databases were among the first systems to implement distributed leader election because the primary-replica pattern fundamentally requires a single write coordinator. Different database systems take remarkably different approaches, reflecting their design priorities and historical evolution.

PostgreSQL with Patroni:

PostgreSQL itself doesn't include leader election—it provides streaming replication between a primary and replicas. The Patroni project adds cluster management with leader election using external coordination services.

• Uses etcd, Consul, or Zookeeper as the coordination backend
• Implements lease-based leadership with configurable TTL
• Leader holds a key in the coordination service; replicas monitor this key
• On leader failure, replicas race to acquire the key (lease acquisition)
• Winner is promoted to primary using PostgreSQL's pg_ctl promote
• Includes fencing via pg_rewind to handle split-brain scenarios during promotion

Key design decisions:

• External coordination avoids embedding consensus in PostgreSQL
• Patroni can pause failover during maintenance windows
• DCS (Distributed Configuration Store) unavailability prevents failover (safety over availability)

MySQL with Group Replication:

MySQL Group Replication (GR) takes a different approach—it embeds Paxos-based consensus directly into the database:

• Nodes form a group and replicate transactions through group consensus
• Primary mode: one primary (elected by consensus) handles writes
• Multi-primary mode: any node can accept writes (conflict detection)
• Built-in failure detection and automatic primary re-election
• No external coordination service required

Key design decisions:

• Integrated solution simplifies deployment
• Paxos overhead impacts write latency
• Partition handling: minority partitions become read-only

CockroachDB and TiDB:

NewSQL databases like CockroachDB and TiDB use Raft consensus for both data replication and leader election:

• Each data range (shard) has its own Raft group
• The Raft leader for each range handles writes to that range
• No single cluster-wide leader; leadership is distributed by range
• Highly available: any range leader failure affects only that range

Coordination services like Zookeeper, etcd, and Consul provide leader election as a primitive that other systems can use. But these services themselves need internal leader election—creating an interesting recursive problem.

Apache Zookeeper:

Zookeeper uses the ZAB (Zookeeper Atomic Broadcast) protocol for consensus and leader election:

• Cluster elects a leader through a voting process
• Leader processes all write requests and replicates to followers
• Leader election triggered on startup or leader failure
• Uses epoch numbers (fencing tokens) to prevent split-brain
• Leader holds a lease that must be renewed via heartbeats to followers

ZAB election process:

1. Nodes start in LOOKING state
2. Each node proposes itself (or its preferred leader)
3. Votes exchanged until majority agrees
4. Winner becomes leader, others become followers
5. Leader synchronizes state with followers before serving requests

Typical failover time: 2-10 seconds (configurable via tickTime and syncLimit)

etcd:

etcd implements Raft consensus, which includes leader election as a core component:

• Raft leader handles all writes and replicates to followers
• Election triggered by heartbeat timeout (150-300ms default)
• Randomized election timeouts prevent election storms
• Term numbers (fencing tokens) ensure stale leaders are rejected
• Leader sends heartbeats; followers reset election timer on receipt

Raft election process:

1. Follower's election timer expires (no heartbeat from leader)
2. Follower becomes candidate, increments term, votes for itself
3. Candidate requests votes from all nodes
4. If candidate receives majority, becomes leader
5. New leader begins sending heartbeats

Typical failover time: 1-3 seconds

HashiCorp Consul:

Consul also uses Raft consensus for its server nodes:

• Similar to etcd; leader handles writes for the consensus log
• Additionally supports distributed locks for application-level leader election
• Lock sessions with TTL provide lease-based locks
• Applications can elect leaders using Consul's lock API

Container orchestration platforms like Kubernetes manage thousands of workloads across potentially thousands of nodes. Leader election is essential for ensuring exactly one controller processes each type of resource.

Kubernetes Control Plane Components:

Kubernetes runs multiple replicas of control plane components for high availability. Each component type uses leader election to ensure only one is active:

• kube-scheduler: Only one scheduler should assign pods to nodes
• kube-controller-manager: Only one controller manager should process controller logic
• cloud-controller-manager: Only one should interact with cloud provider APIs

Kubernetes leader election implementation:

Kubernetes uses a lease-based approach with the Lease resource (or previously ConfigMap/Endpoints):

1. A Lease resource is created for each component type
2. The current leader's identity is stored in the Lease
3. The leader renews the Lease periodically (default: 15s lease, 10s renew, 2s retry)
4. If renewal fails, the leader stops processing
5. Other replicas attempt to acquire the Lease (compare-and-swap)
6. First to successfully update the Lease becomes the new leader

apiVersion: coordination.k8s.io/v1
kind: Lease
metadata:
  name: kube-scheduler
  namespace: kube-system
spec:
  holderIdentity: scheduler-replica-1
  leaseDurationSeconds: 15
  renewTime: "2024-01-15T10:00:00Z"
  acquireTime: "2024-01-15T09:00:00Z"

Application-level leader election in Kubernetes:

Applications running in Kubernetes can also use leader election. The Kubernetes client libraries provide leader election primitives:

// Go client-go example
lock := &resourcelock.LeaseLock{
    LeaseMeta: metav1.ObjectMeta{
        Name:      "my-app-lock",
        Namespace: "default",
    },
    Client: client.CoordinationV1(),
    LockConfig: resourcelock.ResourceLockConfig{
        Identity: os.Hostname(),
    },
}

leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
    Lock:          lock,
    LeaseDuration: 15 * time.Second,
    RenewDeadline: 10 * time.Second,
    RetryPeriod:   2 * time.Second,
    Callbacks: leaderelection.LeaderCallbacks{
        OnStartedLeading: func(ctx context.Context) {
            // Start doing leader work
        },
        OnStoppedLeading: func() {
            // Clean up, stop processing
        },
    },
})

Streaming platforms like Apache Kafka handle massive data volumes and require leader election for both partition leadership and cluster coordination. Their approach to leader election demonstrates sophisticated production requirements.

Apache Kafka:

Kafka has two levels of leadership:

1. Partition Leadership:

• Each topic partition has one leader and multiple followers
• The leader handles all reads and writes for that partition
• Followers replicate from the leader
• If leader fails, a follower in the ISR (In-Sync Replicas) is promoted

2. Controller Leadership:

• One broker serves as the cluster controller
• Controller manages partition assignments and leader elections
• Controller election itself uses Zookeeper (or KRaft)

Traditional Kafka (with Zookeeper):

• Brokers race to create ephemeral znodes in Zookeeper
• First to create /controller znode becomes controller
• Zookeeper sessions provide lease behavior
• Controller failure triggers new controller election

KRaft mode (Kafka Raft Metadata mode):

• Eliminates Zookeeper dependency
• Dedicated controller quorum uses Raft consensus
• Controller quorum maintains cluster metadata
• Partition leaders elected by the controller quorum

Partition leader election in Kafka:

1. Controller detects broker failure (via heartbeat)
2. Controller identifies affected partitions
3. For each partition, controller selects new leader from ISR
4. Controller updates metadata and notifies all brokers
5. Clients refresh metadata and connect to new leaders

Apache Pulsar:

Pulsar uses a different architecture with separate serving and storage layers:

• Brokers are stateless and don't own data
• BookKeeper stores data durably
• Topic 'ownership' (similar to leadership) is managed by Zookeeper
• Multiple brokers can own different topics
• Broker failure: topics are reassigned to other brokers quickly (no data movement needed)

Key insight: By separating compute (brokers) from storage (BookKeeper), Pulsar makes 'leadership' failover nearly instantaneous—new broker just needs to start serving requests from the existing storage.

Operating leader election systems at scale reveals challenges that aren't obvious from theory. These operational lessons come from running production systems through failures, outages, and edge cases.

Maintenance windows and planned failovers:

Production systems need maintenance. Graceful leadership handoff is essential:

1. Drain before failover: Allow current leader to complete in-flight operations
2. Preferred leader configuration: Some systems support specifying preferred leaders for planned promotions
3. Pause failover during maintenance: Systems like Patroni allow pausing automatic failover during manual operations
4. Rolling updates: Update followers first, then trigger controlled failover, then update old leader

Node placement for availability:

Distribute leader election participants across failure domains:

• Minimum 3 nodes for quorum (tolerates 1 failure)
• Place nodes in different availability zones
• For cross-region: consider latency impact on consensus
• Odd numbers preferred (3, 5, 7) to avoid tie scenarios

Effective monitoring is essential for operating leader election systems. You need to know when elections happen, how long they take, and when the system is at risk of unnecessary failover.

Example monitoring setup for Patroni (PostgreSQL):

# Prometheus alerting rules for Patroni
groups:
- name: patroni-leader-election
  rules:
  - alert: PostgresLeaderUnknown
    expr: patroni_leader == 0
    for: 30s
    labels:
      severity: critical
    annotations:
      summary: "No PostgreSQL leader detected"
  
  - alert: TooManyLeaderElections
    expr: increase(patroni_failovers_total[1h]) > 3
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "Excessive PostgreSQL failovers"
  
  - alert: LeaderLeaseExpiringSoon
    expr: patroni_leader_ttl_seconds < 5
    for: 10s
    labels:
      severity: warning
    annotations:
      summary: "PostgreSQL leader lease almost expired"

When designing a new system that requires leader election, how do you choose an approach? Here's a practical decision framework based on system characteristics and requirements.

Decision tree:

Q1: Do you already have a coordination service (etcd, Zookeeper, Consul)?

• YES → Use its leader election primitives (leases/locks)
• NO → Continue to Q2

Q2: How critical is split-brain prevention?

• CRITICAL (data corruption possible) → Use coordination service or embedded Raft
• TOLERABLE (duplicates acceptable) → Simpler approaches may work

Q3: What's your availability requirement?

• Five 9s (< 5 min downtime/year) → Embedded consensus (faster failover, no external dependency)
• Three-four 9s → Coordination service is fine

Q4: What's your team's expertise?

• Distributed systems experts → Can implement/operate any approach
• General developers → Use battle-tested libraries (Kubernetes client-go, etc.)

Q5: What's your scaling requirement?

• Few leaders (< 100) → Any approach works
• Many leaders (1000+) → Watch coordination service load; consider sharding

We've explored how production systems implement leader election, learning lessons that only emerge from operating these systems at scale. Let's consolidate the key takeaways:

Module complete:

You've now journeyed through the complete landscape of leader election—from the fundamental need for coordination, through classic algorithms (Bully, Ring), to modern lease-based approaches, and finally to production implementations. This knowledge equips you to design, implement, and operate distributed systems that require single-leader coordination, and to evaluate the trade-offs inherent in different approaches.