In-Sync Replicas (ISR) in Apache Kafka
In-sync replicas (ISR) are the copies of a Kafka partition that are fully caught up with the partition's leader. Kafka counts a message as safely stored only when every replica in the ISR has it, and it only picks a new leader from the ISR. The ISR therefore decides both how durable your data is and whether writes can continue when brokers fail.
It explains common surprises: producers that suddenly time out, writes that stall, and data loss despite three replicas.
1. Replicas, Leaders and Followers
A topic is split into partitions, and each partition is copied to several brokers (Kafka servers). Each copy is a replica. The number of copies is the topic's replication factor, usually 3.
- One replica is the leader. Producers write to it, and consumers normally read from it.
- The others are followers. They keep fetching new records from the leader, the same way a consumer does, and append them to their own copy of the log.
Each replica has a log end offset (LEO): the offset of the next record it will write. If a replica holds offsets 0 to 9, its LEO is 10.
2. What Makes a Replica In-Sync
The ISR always contains the leader, plus every follower that is keeping up. Kafka 4.x runs in KRaft mode, where the leader reports ISR changes to the controller (the node that manages cluster metadata) with an AlterPartition request. A follower leaves the ISR in one of two ways:
- It falls behind. If a follower has not caught up to the leader's LEO for longer than
replica.lag.time.max.ms(30 seconds by default), the leader removes it. - Its broker dies. If a broker stops sending heartbeats to the controller for
broker.session.timeout.ms(9 seconds by default), the controller fences it and removes it from the ISR of every partition at once.
A follower rejoins the ISR as soon as its LEO reaches the leader's high watermark.
3. How In-Sync Replicas Protect Your Writes
The high watermark (HW) is the smallest LEO among the replicas in the ISR. Records below it are on every in-sync replica, so they are committed. Consumers only read committed records, so what they see survives a leader failure.
Two settings decide what a producer waits for:
acks(producer):acks=all(the default since Kafka 3.0) waits until every replica in the ISR has the record.acks=1waits only for the leader.min.insync.replicas(topic or broker, default 1): the smallest ISR size the leader accepts for anacks=allwrite. If the ISR is smaller, the leader rejects the write withNOT_ENOUGH_REPLICAS.
Note that acks=all means "all in-sync replicas", not all replicas. If the ISR has shrunk to the leader alone, "all" is just one copy unless min.insync.replicas stops it.
4. Code Examples
Example 1 is a plain Java model. Examples 2 and 3 use kafka-clients 4.1.0 against a real Kafka 4.1.0 cluster (Docker image apache/kafka:4.1.0) with one controller and three brokers, b1 to b3. All run on Java 21.
4.1 Example 1: The High Watermark and a Slow Follower
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
public class HighWatermarkModel {
static final long REPLICA_LAG_TIME_MAX_MS = 30_000; // Kafka default
static final class Replica {
final int id;
long logEndOffset; // LEO: offset of the next record this replica will write
long lastCaughtUpMs; // last time it had everything the leader had
Replica(int id) { this.id = id; }
}
public static void main(String[] args) {
Replica leader = new Replica(1), fast = new Replica(2), slow = new Replica(3);
List<Replica> isr = new ArrayList<>(List.of(leader, fast, slow));
for (long now = 0; now <= 40_000; now += 5_000) {
leader.logEndOffset += 100; // producers write 100 records
fast.logEndOffset = leader.logEndOffset; // healthy follower keeps up
fast.lastCaughtUpMs = now;
if (now < 10_000) { // slow follower: fine at first...
slow.logEndOffset = leader.logEndOffset;
slow.lastCaughtUpMs = now;
} else {
slow.logEndOffset += 10; // ...then its disk slows down
}
// The leader drops followers that have not caught up for too long
long cutoff = now - REPLICA_LAG_TIME_MAX_MS;
isr.removeIf(r -> r != leader && r.lastCaughtUpMs < cutoff);
// High watermark = smallest LEO in the ISR; consumers can read only below it
long hw = isr.stream().mapToLong(r -> r.logEndOffset).min().orElseThrow();
System.out.printf("t=%2ds leaderLEO=%3d slowLEO=%3d isr=%s HW=%d%n",
now / 1000, leader.logEndOffset, slow.logEndOffset,
isr.stream().map(r -> String.valueOf(r.id)).collect(Collectors.joining(",", "[", "]")), hw);
}
}
}t= 0s leaderLEO=100 slowLEO=100 isr=[1,2,3] HW=100
t= 5s leaderLEO=200 slowLEO=200 isr=[1,2,3] HW=200
t=10s leaderLEO=300 slowLEO=210 isr=[1,2,3] HW=210
t=15s leaderLEO=400 slowLEO=220 isr=[1,2,3] HW=220
t=20s leaderLEO=500 slowLEO=230 isr=[1,2,3] HW=230
t=25s leaderLEO=600 slowLEO=240 isr=[1,2,3] HW=240
t=30s leaderLEO=700 slowLEO=250 isr=[1,2,3] HW=250
t=35s leaderLEO=800 slowLEO=260 isr=[1,2,3] HW=260
t=40s leaderLEO=900 slowLEO=270 isr=[1,2] HW=900This is the edge case most tutorials miss. A follower that is alive but slow stays in the ISR for the full 30 seconds, and the high watermark crawls with it: acks=all writes wait and consumers see almost nothing new. Once the leader drops it, the high watermark jumps to 900.
4.2 Example 2: Watching the ISR Change
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartitionInfo;
public class IsrWatch {
public static void main(String[] args) throws Exception {
Map<String, Object> config = Map.of(
AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "b1:9092,b2:9092,b3:9092");
try (Admin admin = Admin.create(config)) {
// 1 partition, 3 replicas; acks=all writes need at least 2 in-sync copies
NewTopic topic = new NewTopic("payments", 1, (short) 3)
.configs(Map.of("min.insync.replicas", "2"));
admin.createTopics(List.of(topic)).all().get();
DateTimeFormatter clock = DateTimeFormatter.ofPattern("HH:mm:ss.S");
long start = System.currentTimeMillis();
String last = "";
while (System.currentTimeMillis() - start < 45_000) {
TopicPartitionInfo p = admin.describeTopics(Set.of("payments"))
.allTopicNames().get().get("payments").partitions().get(0);
String state = "leader=" + p.leader().id()
+ " replicas=" + ids(p.replicas()) + " isr=" + ids(p.isr());
if (!state.equals(last)) { // print only when something changes
System.out.println(LocalTime.now().format(clock) + " " + state);
last = state;
}
Thread.sleep(250);
}
}
}
private static String ids(List<Node> nodes) {
return nodes.stream().map(n -> String.valueOf(n.id()))
.collect(Collectors.joining(",", "[", "]"));
}
}13:44:36.1 leader=2 replicas=[2,3,1] isr=[2,3,1]
13:44:52.8 leader=2 replicas=[2,3,1] isr=[2,1]
13:45:00.8 leader=2 replicas=[2,3,1] isr=[2,1,3]Broker 3 was killed with docker kill at 13:44:43.7 and left the ISR 9 seconds later: the controller fenced it (broker.session.timeout.ms), without waiting for the 30-second lag limit. Restarted at 13:44:58, it rejoined within 3 seconds.
4.3 Example 3: acks=all Meets min.insync.replicas
Next, brokers 1 and 3 were killed, leaving only the leader.
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartitionInfo;
import org.apache.kafka.common.serialization.StringSerializer;
public class MinIsrProducer {
private static final String SERVERS = "b1:9092,b2:9092,b3:9092";
public static void main(String[] args) throws Exception {
try (Admin admin = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, SERVERS))) {
TopicPartitionInfo p = admin.describeTopics(Set.of("payments"))
.allTopicNames().get().get("payments").partitions().get(0);
System.out.println("leader=" + p.leader().id() + " isr=" + ids(p.isr())
+ " elr=" + ids(p.elr()) + " (min.insync.replicas=2)");
}
send("all");
send("1");
}
private static void send(String acks) throws InterruptedException {
Map<String, Object> config = Map.of(
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVERS,
ProducerConfig.ACKS_CONFIG, acks,
ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 3000,
ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 6000, // give up after 6 s instead of 2 min
ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
try (KafkaProducer<String, String> producer = new KafkaProducer<>(config)) {
long start = System.currentTimeMillis();
try {
RecordMetadata meta = producer.send(
new ProducerRecord<>("payments", "pay-1001", "INR 499.00")).get();
System.out.println("acks=" + acks + " -> stored at offset " + meta.offset());
} catch (ExecutionException e) {
System.out.println("acks=" + acks + " -> failed after ~"
+ (System.currentTimeMillis() - start) / 1000 + " s: "
+ e.getCause().getClass().getSimpleName() + ": " + e.getCause().getMessage());
}
}
}
private static String ids(List<Node> nodes) {
return nodes.stream().map(n -> String.valueOf(n.id()))
.collect(Collectors.joining(",", "[", "]"));
}
}leader=2 isr=[2] elr=[3] (min.insync.replicas=2)
acks=all -> failed after ~6 s: TimeoutException: Expiring 1 record(s) for payments-0:6003 ms has passed since batch creation
acks=1 -> stored at offset 2With one replica in the ISR, the leader rejected the acks=all write; its log shows 9 NotEnoughReplicasException errors. The producer treats that error as retriable, so it kept retrying and reported only a TimeoutException. With the default delivery.timeout.ms, that takes 2 minutes. Meanwhile acks=1 was acknowledged at offset 2, yet a consumer read only offsets 0 and 1. With Eligible Leader Replicas enabled (see below), the high watermark does not advance while the ISR is below min.insync.replicas. The record has one copy and stays invisible until the ISR recovers.
The elr=[3] field shows Eligible Leader Replicas (ELR, KIP-966), which are enabled by default on new Kafka 4.1 clusters. Broker 3 left the ISR when the ISR dropped below min.insync.replicas, and from then on the high watermark stopped advancing. Broker 3 is therefore known to hold every committed record, so the controller may elect it if broker 2 also fails.
5. acks and min.insync.replicas Combinations
| Setting (replication factor 3) | Brokers that can fail and writes still succeed | Copies of each acknowledged record |
|---|---|---|
acks=1, any min.insync.replicas | 2 | At least 1 |
acks=all, min.insync.replicas=1 | 2 | At least 1 |
acks=all, min.insync.replicas=2 | 1 | At least 2 |
acks=all, min.insync.replicas=3 | 0 | 3 |
The third row is the usual production choice: it survives one broker failure without stopping writes, and every acknowledged record has at least two copies. With ELR enabled (Kafka 4.1), records written with acks=1 while the ISR is below the minimum are acknowledged but not readable until it recovers.
6. Common Mistakes and Pitfalls
- Trusting
acks=allalone. With the defaultmin.insync.replicas=1,acks=allcan be satisfied by the leader alone. If that broker's disk then fails, the record is gone. - Setting
min.insync.replicasequal to the replication factor. With 3 and 3, a single broker restart blocks allacks=allwrites. - Waiting for
NotEnoughReplicasExceptionin code. As Example 3 shows, the producer usually reports aTimeoutExceptioninstead. Alert on under-min-ISR partitions on the broker side. - Raising
replica.lag.time.max.msto stop ISR "flapping". It hides slow brokers and makes writes stall longer, as Example 1 shows. Fix the slow disk or network instead.
7. Best Practices
- Monitor
UnderReplicatedPartitionsandUnderMinIsrPartitionCount. Above zero means reduced safety or blocked writes. - Keep
unclean.leader.election.enable=false(the default) unless availability matters more than losing records. - Spread replicas across racks with
broker.rack, so one rack failure cannot empty an ISR. - Restart brokers one at a time, waiting for the ISR to refill.
8. Summary
- In-sync replicas are the leader plus the followers that are fully caught up with it.
- Followers leave the ISR after 30 seconds of lag or 9 seconds without broker heartbeats, and rejoin when they reach the high watermark.
- The high watermark is the lowest LEO in the ISR. Consumers only read below it.
acks=allwaits for the current ISR, andmin.insync.replicassets how small that ISR may get.- Producers often see a timeout, not
NotEnoughReplicasException, when the ISR is too small. - Kafka 4.1's Eligible Leader Replicas allow safe leader election after the ISR shrinks.
Kafka In-Sync Replicas (ISR): Interview Questions
Common interview questions on Kafka In-Sync Replicas (ISR), from basic to scenario-based, with short answers.
Basic questions
Q. Is the leader always part of the ISR?
Answer: Yes. The ISR is defined relative to the leader, so the leader is always a member. When a new leader is elected, it is chosen from the ISR, and the followers then measure their progress against the new leader.
Q. Does a consumer have to read from the leader?
Answer: Not always. Since Kafka 2.4 (KIP-392), a consumer can set client.rack and read from a nearby follower if the broker uses a rack-aware replica.selector.class. A follower still only serves records below the high watermark, so the consumer never sees uncommitted data.
Q. Can the ISR ever be larger than the replication factor?
Answer: No. The ISR is always a subset of the replica list, so its size is between 1 and the replication factor. During a partition reassignment the replica list grows temporarily, and the ISR can grow with it.
Intermediate questions
Q. What is NotEnoughReplicasAfterAppendException?
Answer: The leader checks the ISR size before it writes a record. If the ISR shrinks below min.insync.replicas after the record was written but before it was replicated, the producer gets this error instead. The record may already be in the leader's log, so a retry can write it twice unless the producer is idempotent (the default since Kafka 3.0).
Q. Why did Kafka remove replica.lag.max.messages?
Answer: Older versions also removed a follower that fell more than a fixed number of messages behind. During traffic spikes, healthy followers briefly exceeded that count and were dropped and re-added constantly. Kafka 0.9 removed the setting, and lag is now measured only in time.
Q. How do you find partitions whose ISR is below min.insync.replicas?
Answer: Use the topic tool with a filter, or check the broker metric UnderMinIsrPartitionCount.
kafka-topics.sh --bootstrap-server b1:9092 --describe --under-min-isr-partitionsQ. What happens when unclean leader election is enabled and the ISR is empty?
Answer: Kafka may choose an out-of-sync replica as the new leader. The partition becomes writable again, but every record the old leader had that this replica lacked is lost. Consumers may also see offsets reused for different records.
Tricky and scenario questions
Q. Your topic has replication factor 3 and min.insync.replicas=2. One broker is down for maintenance, and another starts a long garbage collection pause. What happens?
Answer: If the paused broker leads the partition, nothing can be written to the partition until the pause ends or the controller fences the broker and elects a new leader. If it is a follower, acks=all writes wait for it. If the pause is longer than replica.lag.time.max.ms, the leader drops it from the ISR. The ISR then has only 1 replica, which is below 2, so writes fail until a broker returns.
Q. Why can consumer lag grow while producers are writing successfully with acks=1?
Answer: acks=1 only waits for the leader, but consumers can only read up to the high watermark. If a follower in the ISR is slow, the high watermark lags behind the leader's LEO. The producer is happy while consumers wait, as Example 1 shows.
Q. A topic shows Isr: 2 and Elr: 3. Broker 2 crashes, and broker 3 is started again. Which broker leads next?
Answer: It depends on how broker 3 stopped. If it shut down cleanly, the controller elects it from the ELR, because it holds every committed record. If it crashed, its unflushed data may be gone, so it loses its ELR place when it restarts, and the partition stays offline until broker 2 returns. You can reproduce both on Kafka 4.1 with docker stop and docker kill. Without ELR, the partition always waits for broker 2, unless unclean leader election is enabled.