What is the Kafka controller broker, and how does leader election from the ISR keep a partition available when its leader broker fails? | Kafka & Messaging

Definition

The controller is the one elected broker that manages broker membership and partition leadership; on a leader's failure it elects a new leader from the ISR, which already holds all committed data, so the partition stays available with no acknowledged data lost.

The controller and ISR-based leader election are how Kafka turns a broker failure into a quick, lossless handoff.

Controller – a single broker elected to manage cluster-wide metadata: live brokers, partition assignments, leadership, and ISR sets.
KRaft – the modern Raft-based controller quorum that replaces ZooKeeper for metadata management.
ISR (In-Sync Replicas) – the subset of replicas fully caught up with the leader; a record is committed only when all ISR members have it.
Leader election – on leader failure, the controller picks a new leader from the surviving ISR members.
Unclean leader election – promoting an out-of-sync replica that may be missing committed records; disabled for data safety in banking.

Imagine National Bank of India has many branches, and for each account type one branch is the designated master handling all transactions, with backup branches keeping live copies. The bank appoints one head-office coordinator (the controller) who knows which branches are up, which is master for each account, and which backups are fully up to date. If the master branch burns down, the coordinator picks one of the fully-caught-up backups as the new master and updates the directory. Because the chosen backup was fully caught up, customers lose nothing and service continues almost without interruption.

Understanding the Concepts

What is the Kafka controller broker?

Exactly one broker in the cluster acts as the controller at any time. It maintains the authoritative cluster metadata: live brokers, replica assignments, current leaders, and ISR sets. It watches for brokers joining and leaving and drives leader elections.

BASH

# See which broker is the current controller
kafka-metadata.sh --snapshot /data/kafka/metadata/__cluster_metadata-0/00000000000000000000.log \
  --controllers

# Or in legacy ZooKeeper mode:
# zookeeper-shell.sh zk-1:2181 get /controller

How is the controller elected?

In legacy mode, controller election was coordinated through ZooKeeper. In modern Kafka (KRaft), a built-in Raft-based quorum among a set of controller nodes elects an active controller and stores cluster metadata in an internal metadata log. Either way there is always exactly one active controller, and if it fails a new one is elected.

BASH

# KRaft controller configuration (server.properties)
process.roles=controller
node.id=1
controller.quorum.voters=1@controller-1:9093,2@controller-2:9093,3@controller-3:9093

# The quorum has 3 voters: tolerates 1 failure
# Active controller is elected by Raft consensus

What is the in-sync replica set (ISR)?

Each partition has a leader and followers. The ISR is the subset of replicas, always including the leader, that are fully caught up within the allowed lag window (replica.lag.time.max.ms). A record is committed only once all ISR members have it. This is why every ISR replica is guaranteed to hold every committed record.

JAVA

package com.bank.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

public class IsrDurabilityDemo {
    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.ACKS_CONFIG, "all"); // wait for all ISR

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            RecordMetadata meta = producer.send(new ProducerRecord<>(
                "settlements", "SAV2001", "{\"amount\":50000}")).get();
            System.out.println("Committed to partition " + meta.partition()
                + " offset " + meta.offset() + " -- all ISR replicas have it");
        }
    }
}

TEXT

Committed to partition 0 offset 42 -- all ISR replicas have it

How does leader election from the ISR work on failure?

When a broker that leads partitions fails, the controller detects it through lost heartbeats, removes it from ISR sets, and for each orphaned partition elects a new leader from the surviving ISR members. It propagates updated metadata so clients redirect to the new leaders.

TEXT

Failover sequence:
1. Broker-1 (leader of settlements-0) fails
2. Controller detects: heartbeat timeout exceeded
3. Controller reads ISR for settlements-0: [1, 2, 3]
4. Removes Broker-1: surviving ISR = [2, 3]
5. Elects Broker-2 as new leader
6. Propagates metadata: settlements-0 leader = Broker-2
7. Clients refresh, redirect to Broker-2
8. No committed data lost (Broker-2 had everything)

What is unclean leader election?

If a partition's entire ISR is down, the controller faces a choice: wait for an ISR member to return (correct but unavailable) or promote an out-of-sync replica (available but may lose data). For banking, you keep unclean.leader.election.enable=false.

BASH

# Disable unclean leader election (default in newer Kafka)
# server.properties
unclean.leader.election.enable=false

# Combined with:
min.insync.replicas=2
default.replication.factor=3

# Result: tolerate 1 broker failure with zero data loss
# If 2 brokers fail, partition is unavailable but no data is silently lost

How do you configure for safe failover?

The standard banking configuration at National Bank of India uses replication factor 3, acks=all, min.insync.replicas=2, and unclean leader election disabled. This tolerates one broker failure with zero data loss.

JAVA

package com.bank.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

public class SafeProducerConfig {
    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            producer.send(new ProducerRecord<>("settlements", "SAV2001",
                "{\"amount\":50000}")).get();
            System.out.println("Settlement committed with acks=all, safe failover guaranteed");
        }
    }
}

TEXT

Settlement committed with acks=all, safe failover guaranteed

How do you monitor ISR health?

Alert on ISR shrinkage because a partition sitting at ISR=1 has no safe failover target left. Monitor the ISR shrink rate and under-replicated partition count.

BASH

# Check under-replicated partitions
kafka-topics.sh --describe --under-replicated-partitions \
  --bootstrap-server kafka-1:9092

# Key JMX metrics
# kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions
# kafka.server:type=ReplicaManager,name=IsrShrinksPerSec
# kafka.server:type=ReplicaManager,name=IsrExpandsPerSec
# kafka.controller:type=ControllerStats,name=LeaderElectionRateAndTimeMs

What should a senior engineer know?

Failover speed depends on how quickly the controller detects the failure, governed by session and heartbeat timeouts. There is a tension between fast failover and not evicting briefly slow brokers. KRaft makes controller failover and metadata propagation faster than ZooKeeper. The ISR is dynamic, so the pool of eligible leaders shrinks under load, which is why you monitor ISR shrink and keep brokers stable.

BASH

# Tune detection speed vs stability
# Faster detection (risk: evict slow-but-alive brokers)
replica.lag.time.max.ms=10000

# Safer (slower detection, fewer false evictions)
replica.lag.time.max.ms=30000

# The right value depends on your GC pause profile
# If max GC pause is 5s, set lag.time > 5s with margin

Visual Overview

DIAGRAM

  Controller and ISR-Based Leader Election

  ┌─────────────────────────────────────────────────────┐
  │  Controller (Broker-3 or KRaft quorum)               │
  │  Manages: broker list, partition leaders, ISR sets   │
  └───────────────────────┬─────────────────────────────┘
                          │ monitors heartbeats
                          ▼
  ┌──────────┐  ┌──────────┐  ┌──────────┐
  │ Broker-1 │  │ Broker-2 │  │ Broker-3 │
  │ (Leader) │  │(Follower)│  │(Follower)│
  │ ISR=[1,2,3] │          │  │          │
  └──────────┘  └──────────┘  └──────────┘
       │              │              │
  settlements-0   settlements-0  settlements-0
    (leader)       (follower)    (follower)

  Broker-1 fails:
  ┌──────────┐  ┌──────────┐  ┌──────────┐
  │ Broker-1 │  │ Broker-2 │  │ Broker-3 │
  │   DEAD   │  │ NEW LEAD │  │(Follower)│
  │          │  │ ISR=[2,3]│  │          │
  └──────────┘  └──────────┘  └──────────┘
                     ▲
                     │ Controller elects from surviving ISR
                     │ No committed data lost

Key Takeaway

The controller is a single broker elected to coordinate the cluster. It holds the authoritative metadata: live brokers, partition assignments, current leaders, and ISR sets. In KRaft it is a Raft-based quorum; in legacy mode it was coordinated through ZooKeeper. Either way there is always one active controller, and it is itself made fault-tolerant through re-election.

The mechanism that keeps a partition available on leader failure is leader election from the ISR. A record is committed only when all ISR members have it, so every ISR replica holds every committed record. When a leader fails, the controller picks a new leader strictly from the surviving ISR, propagates the metadata, and clients redirect. The partition is unavailable only for the brief detection-to-election window.

On the National Bank of India settlements topic, with replication factor 3, acks=all, and min.insync.replicas=2, losing Broker-1 simply promoted Broker-2 from the ISR and settlements for Rajesh and Priya kept flowing with every committed record intact. Pankaj keeps unclean leader election disabled so the controller will never promote a stale replica, trading availability for the guarantee that no acknowledged data is lost.

Remember one sentence: the controller plus ISR election is how Kafka turns a broker failure into a quick, lossless leadership handoff rather than an outage.

---