Leader & Follower #

In large-scale distributed systems, hardware failure isn’t a question of “if” but “when”. A physical broker server can suffer power outages, hard drive failures, or network isolation at any time without warning. To ensure our business data flows continue without interruption, Apache Kafka designs a robust partition replication architecture based on the Leader & Follower model. Every topic partition in a Kafka cluster runs redundantly across several brokers. One replica is designated the Leader, while the other replicas act as Followers. Understanding how these two roles interact, how the Log End Offset (LEO) and High Watermark (HW) markers limit consumer data visibility, and how the consensus protocol handles automatic power transitions during failover is the key to designing data processing systems with absolute reliability guarantees (zero data loss).

Replica Roles: Leader vs Follower #

Replication in Kafka is configured at the topic level through the replication.factor parameter. If we set the replication factor to 3, every partition of that topic has 3 physical copies spread across 3 different brokers. Of those three copies, the system divides roles as follows:

1. Leader Replica #

For each partition, one broker is designated as the Leader. This broker acts as the single gateway for all data interaction activities.

  • Receiving Writes: All PRODUCE requests from our producer applications must be addressed and written directly to the Leader.
  • Serving Reads: By default, all FETCH requests from consumers are also served by the Leader.

2. Follower Replica #

Other brokers storing copies of the same partition act as Followers. Their role is very different from the secondary database concept in traditional relational databases:

  • Passive Copying: Followers don’t serve producer write requests. Their sole job is to act like internal consumers. They run a background thread named ReplicaFetcherThread that constantly sends binary Fetch requests to the Leader broker to pull the latest messages and copy them to their own local logs in order.
  • Warm Standby: Followers stand fully ready as backups. If the Leader broker crashes, one of the healthiest, most in-sync Followers is immediately promoted by the cluster to become the new Leader.

Modern Feature: Follower Fetching (Read from Closest Replica) #

Starting from Kafka 2.4, Kafka introduced a network optimization feature called Follower Fetching. In multi-zone (Multi-AZ) clusters, if our consumers are in a different Availability Zone (AZ) than the Leader broker, transferring data across zone boundaries triggers very expensive cross-zone data transfer costs from cloud providers.

By enabling the replica.selector.class configuration on the broker side and setting the consumer client configuration to the same rack/zone location, consumers are allowed to read data directly from the nearest Follower in the same AZ. This dramatically cuts network latency and saves our operational infrastructure costs.

Replication Indicators: Log End Offset (LEO) vs High Watermark (HW) #

To coordinate data synchronization between brokers without slow transaction locking, Kafka uses two crucial numeric pointers:

1. Log End Offset (LEO) #

Log End Offset (LEO) is the coordinate of the next offset that will be written to a replica’s physical log. LEO indicates the total log length on that replica at a given time.

  • Every time a producer successfully writes a new message to the Leader, the Leader’s LEO value increases.
  • Every time a Follower successfully copies a new message from the Leader to its local disk, that Follower’s LEO value also increases.
  • The Leader’s LEO value is always the highest or equal to the most in-sync Follower’s LEO.

2. High Watermark (HW) #

High Watermark (HW) is the highest offset at which all replicas in the In-Sync Replicas (ISR) group have successfully copied that message to their respective logs.

  • The HW value is mathematically calculated by the Leader as the smallest LEO value among all active ISR replicas.
  • Consumer Visibility Limit: This is the most critical rule in Kafka: Consumers are only allowed to read messages up to the High Watermark. Messages between HW and the Leader’s LEO (messages already written to the Leader but not yet fully copied by all ISR followers) are considered uncommitted and hidden from consumers.

Why are messages above HW hidden from consumers?

Let’s take a bad scenario: A producer sends a message with offset 5 to the Leader (Broker 1). Broker 1’s LEO becomes 6. However, before Broker 2 (Follower) manages to copy that message, Broker 1 suddenly dies. If consumers were allowed to read offset 5 as soon as it’s written on the Leader, consumers would process that data.

Then, when Broker 2 is promoted to the new Leader, offset 5 doesn’t exist in its log. When the producer sends a new message, it’s written at offset 5 with different content. Consumers would now see a fatal data inconsistency (dirty read). By limiting reads only up to HW, Kafka guarantees that messages already read by consumers will never be lost or changed even if a broker crashes.

Synchronization Diagram: The LEO and HW Journey #

Let’s look at a visualization of the LEO and HW position differences inside the physical logs of the Leader and Follower during asynchronous replication:

flowchart TD
    subgraph LeaderReplica["Broker 1: Leader (Partition 0)"]
        direction TB
        L0["Msg 0 (Offset 0)"]
        L1["Msg 1 (Offset 1)"]
        L2["Msg 2 (Offset 2)"]
        L3["Msg 3 (Offset 3)"]
        
        style L0 fill:#ddffdd,stroke:#88ff88
        style L1 fill:#ddffdd,stroke:#88ff88
        style L2 fill:#ddffdd,stroke:#88ff88
        style L3 fill:#ffdddd,stroke:#ff8888
    end

    subgraph FollowerReplica["Broker 2: Follower (Partition 0)"]
        direction TB
        F0["Msg 0 (Offset 0)"]
        F1["Msg 1 (Offset 1)"]
        F2["Msg 2 (Offset 2)"]
        
        style F0 fill:#ddffdd,stroke:#88ff88
        style F1 fill:#ddffdd,stroke:#88ff88
        style F2 fill:#ddffdd,stroke:#88ff88
    end

    Consumer["Consumer Client"] -->|"Read Limit (High Watermark = Offset 2)"| L2
    Producer["Producer Client"] -->|"Write New Data (LEO = Offset 4)"| L3
    FollowerReplica -->|"Fetch Data (Fetch LEO = Offset 3)"| LeaderReplica

In the diagram above, the message at Offset 3 (Msg 3) is already written on the Leader (LEO = 4), but because the Follower hasn’t finished copying it (Follower LEO = 3), the High Watermark is locked at Offset 2. Consumers can’t read Msg 3 yet until the Follower successfully copies it in the next fetch cycle.

Replication Lifecycle (Replica Fetcher Protocol) #

The LEO synchronization and HW movement process happens asynchronously through a continuous request-response loop executed by the Follower thread:

  1. Follower Sends Fetch Request: The ReplicaFetcherThread on the Follower sends a binary ReplicaFetchRequest to the Leader. In this request, the Follower includes the last offset it has in its local log (for example, FetchOffset = 3).
  2. Leader Updates Follower LEO: When the Leader receives that request, it realizes the Follower has successfully copied all messages up to offset 2 (because it’s requesting starting from offset 3). The Leader immediately updates its internal LEO record for that Follower to 3 in its metadata memory.
  3. Leader Reads Data & Sends Reply: The Leader reads new messages starting from offset 3 from its Page Cache, assembles them into a data packet, and sends it back to the Follower along with the Leader’s current High Watermark (HW) coordinate.
  4. Follower Writes to Disk & Updates HW: The Follower receives that binary data packet, writes it to its local log, raises its local LEO, and updates its local HW value according to the HW reported by the Leader.

This cycle repeats constantly within milliseconds, keeping the data difference between the Leader and Follower close to zero.

Automatic Failover Process and the Role of Leader Epoch #

When the broker acting as a partition’s Leader suddenly crashes, the cluster coordination system (KRaft Controller Quorum) takes quick action to appoint a new leader.

1. Failure Detection #

The Controller monitors heartbeats from all active brokers. If the Leader broker fails to send a heartbeat within the timeout limit, the Controller marks that broker dead and starts the partition leader election process.

2. Electing a New Leader from the ISR #

The Controller checks the In-Sync Replicas (ISR) list for that partition. Only followers in the ISR list are eligible to be promoted as the new Leader. This is because ISR followers are guaranteed to have the most complete, in-sync data up to the High Watermark.

3. The Classic Log Reorientation Problem (Log Truncation) #

In the past, when a follower was promoted as the new Leader, it used the High Watermark as the single reference point to reconcile logs with other followers. However, this often triggered silent data loss or inconsistent duplicate data due to HW position desynchronization during cascading crashes.

4. Modern Solution: Leader Epoch #

To solve this problem absolutely, Kafka introduced the Leader Epoch concept. Every time a new partition leader election happens, the Controller increments the leadership generation number (Leader Epoch) by 1. Every broker stores this leadership history list in a secret text file named leader-epoch-checkpoint in their partition directory.

When a Follower broker reconnects to the newly elected Leader, it no longer directly truncates its log based on local High Watermark. The Follower sends a special metadata request asking for the Leader’s last LEO from the previous term (epoch). This Leader Epoch information ensures the log truncation process happens precisely, only on truly uncommitted messages, preventing accidental loss of historical data.

Common Mistakes (Anti-patterns) in Replication #

Here are common partition replication configuration mistakes along with their fixes:

1. Ignoring Failover Callbacks in Client Application Code #

When leader failover happens, producer message delivery fails temporarily for a few milliseconds because the Leader location is moving. Developers often don’t handle this exception on the client side, letting the application crash or lose data.

Consequences: Our application discards important messages during failover, causing business transaction data leaks.

// =========================================================================
// ANTI-PATTERN: Ignoring leader failover exceptions and letting the app crash
// When the leader broker dies, sends without recovery immediately drop the client connection.
// =========================================================================
try {
    producer.send(record).get();
} catch (Exception e) {
    System.err.println("Fatal: Failed to send message: " + e.getMessage());
    // Application crashes or data is lost without retry
    System.exit(1);
}

// =========================================================================
// THE CORRECT SOLUTION: Using asynchronous callbacks and relying on the SDK's internal auto-retry
// The Kafka SDK automatically detects NotLeaderOrFollowerException and refreshes metadata
// to dynamically find the new Leader's location.
// =========================================================================
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ProducerFailoverHandler {
    private static final Logger log = LoggerFactory.getLogger(ProducerFailoverHandler.class);

    public void sendWithRetryHandling(org.apache.kafka.clients.producer.KafkaProducer<String, String> producer, 
                                      org.apache.kafka.clients.producer.ProducerRecord<String, String> record) {
        
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    if (exception instanceof org.apache.kafka.common.errors.NotLeaderOrFollowerException) {
                        // ✓ Automatically handled by the SDK via metadata refresh,
                        // but if retries are exhausted, we can manually fall back to a backup database
                        log.error("Leader failover detected for partition: " + exception.getMessage());
                        fallbackToStorage(record);
                    } else {
                        log.error("Delivery permanently failed: " + exception.getMessage());
                    }
                } else {
                    log.info("Message successfully replicated on partition: " + metadata.partition() 
                        + ", offset: " + metadata.offset());
                }
            }
        });
    }

    private void fallbackToStorage(org.apache.kafka.clients.producer.ProducerRecord<String, String> record) {
        // Temporarily save data to a local database / backup file system to resend later
        log.warn("Saving record to backup storage: " + record.key());
    }
}

2. Setting replica.lag.time.max.ms Too Low #

Setting the follower lag detection parameter replica.lag.time.max.ms to a very sensitive value (e.g., 500 milliseconds) hoping failure detection happens fast.

Consequences: In busy network conditions or during workload spikes, healthy followers often lag a few milliseconds behind the Leader. If the limit is set too sensitive, those healthy followers get kicked out of the ISR list repeatedly by the Leader, triggering a barrage of cluster metadata updates to the Controller that waste CPU and network resources. Raise the value to something reasonable (e.g., 10000 ms to 30000 ms) to absorb network fluctuations.

Summary #

  • Leader & Follower — Kafka’s partition replication design where the Leader serves client reads and writes, while Followers actively copy data asynchronously from the Leader.
  • Log End Offset (LEO) — The coordinate of the next offset to be written at the end of a replica’s physical log, indicating the current total log length.
  • High Watermark (HW) — The highest offset successfully copied by all replicas in the ISR group. Consumers can only read data up to this HW limit.
  • Follower Fetching — A network routing optimization allowing consumers to read data directly from the nearest Follower in the same Availability Zone.
  • ReplicaFetcherThread — The background thread running on Follower brokers to constantly send binary FETCH requests to the Leader broker.
  • Leader Epoch — The partition leadership generation number guiding follower brokers to perform precise log truncation during failover.
  • Failure Tolerance — Zero data loss guarantees are achieved by distributing replicas Rack-Aware, setting replication factor to at least 3, and acks=all on producers.

← Previous: Overview Next: In-Sync Replica →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact