Replication

Active contributors: PoAn Yang, Ken Huang, Calvin Liu, Mickael Maison, José Armando García Sancio

Each Kafka partition is replicated across replication.factor brokers. One replica is the leader (the one clients read and write); the others are followers that fetch from the leader. The set of followers caught up to the leader is the In-Sync Replica set (ISR). Only ISR members are eligible to become leader, and a Produce with acks=all returns only after all ISR followers have appended the record.

Components

graph LR
    Client -->|Produce / Fetch| L[Leader broker]
    L -->|UnifiedLog.append| LD[(Leader log dir)]
    F1[Follower broker A] -->|FetchRequest| L
    F2[Follower broker B] -->|FetchRequest| L
    L -->|FetchResponse| F1 & F2
    F1 --> FD1[(Follower A log dir)]
    F2 --> FD2[(Follower B log dir)]
    L -->|AlterPartition - add/remove ISR| Ctrl[Controller<br/>QuorumController]
    Ctrl -->|PartitionRecord| MetaLog[(Metadata log)]
    MetaLog --> All[All brokers replay]

Source files:

Component	File
Partition (per-partition state)	core/src/main/scala/kafka/cluster/Partition.scala
ReplicaManager (the entry point)	core/src/main/scala/kafka/server/ReplicaManager.scala
AbstractFetcherThread	core/src/main/scala/kafka/server/AbstractFetcherThread.scala
ReplicaFetcherThread / Manager	core/src/main/scala/kafka/server/ReplicaFetcherThread.scala, ReplicaFetcherManager.scala
LocalLeaderEndPoint / RemoteLeaderEndPoint	core/src/main/scala/kafka/server/{LocalLeaderEndPoint,RemoteLeaderEndPoint}.scala
UnifiedLog (the log itself)	storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java
LeaderEpochFileCache	storage/.../log/epoch/LeaderEpochFileCache.java
ReplicationControlManager (controller)	metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java

Append path (acks=all)

sequenceDiagram
    participant P as Producer
    participant L as Leader (KafkaApis -> ReplicaManager)
    participant F1 as Follower A (ReplicaFetcherThread)
    participant F2 as Follower B
    P->>L: ProduceRequest(records, acks=all)
    L->>L: UnifiedLog.appendAsLeader → LEO advances
    L->>L: park request in DelayedProducePurgatory
    F1->>L: Fetch(replica, fetchOffset)
    F2->>L: Fetch(replica, fetchOffset)
    L-->>F1: records + leader HW
    L-->>F2: records + leader HW
    F1->>L: Fetch (next), reporting LEO
    F2->>L: Fetch (next), reporting LEO
    L->>L: minISR followers caught up → advance HW
    L->>L: HW past produced offset → complete delayed produce
    L-->>P: ProduceResponse(success)

Key invariants:

• The leader appends locally first.
• Followers replicate via the standard Fetch RPC; the same code path serves consumer fetches.
• The leader maintains a Replica per follower with its reported LEO.
• The high watermark (HW) is min(LEO of every ISR member); consumers only see records below the HW.
• A produce request with acks=all is parked in DelayedProducePurgatory until the HW catches up to the produced offset, then unparks and returns.

ISR maintenance

Followers must keep up. If a follower's last reported LEO is older than replica.lag.time.max.ms, the leader removes it from the ISR by sending an AlterPartitionRequest to the controller. The controller validates and writes a PartitionChangeRecord into the metadata log; brokers see it via MetadataLoader and update their local Partition state.

A removed follower can rejoin once it catches up — the leader sends another AlterPartition adding it back. ISR membership is therefore eventually consistent across the cluster, with the controller as the source of truth.

min.insync.replicas is the operator-level guarantee: writes with acks=all will fail if the ISR drops below this number, ensuring durability.

Leader election

When the leader fails (or its broker is fenced by the controller for missing heartbeats), the controller picks the first member of the ISR as the new leader (ReplicationControlManager's preferred-replica algorithm) and writes a PartitionChangeRecord with the new leader and a bumped leaderEpoch. Brokers replay the record and adjust local state.

If unclean.leader.election.enable=true, the controller may pick a non-ISR replica when the ISR is empty, accepting potential data loss. The default in 4.x is false.

ELR (Eligible Leader Replicas, KIP-966) is a newer concept: a replica that was last known to be in the ISR but has since gone offline. ELRs persist across controller failovers so that a broker that comes back online after a controlled shutdown can become leader without unclean election. ELR support is gated on its own eligible.leader.replicas.version feature level.

Truncation and leader epochs

Followers must truncate to a known consistent offset when joining a new leader. Each replica keeps a LeaderEpochFileCache ((epoch, startOffset) pairs); when a follower becomes a follower of a new leader, it sends OffsetForLeaderEpochRequest to learn the leader's last offset for the follower's last epoch and truncates to that offset.

This protocol replaced the older "truncate to high watermark" rule, which had subtle data-loss bugs around in-flight produce traffic.

Fetcher threads

Each broker runs:

• One ReplicaFetcherThread per leader broker it follows. Threads are pooled by ReplicaFetcherManager.
• ReplicaAlterLogDirsThread for moves between log directories (JBOD reassignment).

The fetcher loop is a do { fetch; append; advance HW; } while (running) driven by AbstractFetcherThread. It's also the place where followers learn the truncation offset on epoch transitions and where they trigger OffsetMovedToTieredStorage handling.

Reassignment

Operators move partitions between brokers via AlterPartitionReassignments (issued by kafka-reassign-partitions.sh):

1. The reassignment specifies a new replica set (addingReplicas, removingReplicas).
2. The controller writes the change.
3. Brokers in addingReplicas start as followers, replicate, eventually join the ISR.
4. Once the ISR contains all the new members, the controller drops removingReplicas from the assignment.
5. Old replicas delete their local logs.

Throttling (leader.replication.throttled.rate / follower.replication.throttled.rate) caps replication bandwidth during reassignment so that ongoing client traffic isn't starved.

Fetch sessions

For high-fan-out clusters, a follower fetching thousands of partitions per leader could spend many bytes per FetchRequest re-listing partitions. KIP-227 introduced fetch sessions: the first Fetch registers a session and lists all partitions; subsequent fetches reference the session ID and only mention partitions whose fetchOffset changed. FetchSessionHandler.java (clients/) is the client side; the broker side lives in core/.../server/FetchManager.

Configuration knobs

• replication.factor — default replicas per partition.
• min.insync.replicas — write durability.
• replica.lag.time.max.ms — how long before a slow follower drops from ISR.
• replica.fetch.max.bytes, replica.fetch.min.bytes, replica.fetch.wait.max.ms — fetcher tuning.
• unclean.leader.election.enable — durability vs availability tradeoff.
• num.replica.fetchers — fetcher threads per broker.
• leader.imbalance.check.interval.seconds, leader.imbalance.per.broker.percentage — preferred-leader auto-rebalance.

Entry points for modification

• Change replication semantics: Partition.scala + ReplicaManager.scala. Anything that affects HW or ISR must round-trip through AlterPartition and the controller.
• Change fetcher behavior: AbstractFetcherThread.scala, ReplicaFetcherThread.scala. Be mindful of the truncation protocol.
• Change leader election: ReplicationControlManager.java on the controller. KIP required.

Related pages

• Modules: core (broker).
• Modules: storage.
• Modules: metadata.
• Features: KRaft mode.