KafkaIQ

3.2 ISR & High Watermark Mechanics

Module Overview

Goal: The ONE authoritative explanation of In-Sync Replicas (ISR) and High Watermark (HW) — Kafka's durability boundaries.

What You'll Learn:

  • ISR membership rules and transitions
  • High Watermark calculation algorithm
  • Commit visibility and consumer reads
  • Read consistency guarantees
  • ISR shrinks and expands
  • Edge cases and race conditions

What's NOT Covered Here:

  • Basic replication flow (covered in Module 3.1)
  • Follower fetch details (covered in Module 3.3)
  • Leader election (covered in Module 4.x)

Key Insight: ISR and HW are the core mechanisms that determine what data is "committed" and visible to consumers. Understanding these is critical to understanding Kafka's durability and consistency guarantees.

1. In-Sync Replica Set (ISR)

Definition

ISR = Set of replicas that are "caught up" with the leader

Membership criteria:
└─ Replica has fetched from leader within replica.lag.time.max.ms

Example:
├─ Partition: orders-0
├─ Replicas: [1, 2, 3]
├─ Leader: 1
├─ ISR: [1, 2]  # Broker 3 fell behind
└─ Out-of-sync: [3]

Only ISR members participate in durability guarantees

Why ISR Matters

WITH acks=all:

Producer waits for ISR members (not all replicas):
├─ If ISR = [1, 2, 3]: Wait for all 3
├─ If ISR = [1, 2]: Wait for only 2 (broker 3 slow/dead)
└─ If ISR = [1]: Wait for leader only (dangerous!)

Benefits:
├─ Availability: Slow replicas don't block writes
├─ Durability: Writes are safe on all "healthy" replicas
└─ Flexibility: ISR shrinks/expands dynamically

2. ISR Membership Rules

Joining ISR

FOLLOWER JOINS ISR WHEN:

1. Replica is newly created (starts in ISR)
2. Replica catches up after falling behind

Catch-up criteria:
├─ Follower LEO >= Leader HW
└─ AND replica has fetched within replica.lag.time.max.ms

Process:
1. Follower fetches continuously
2. Follower LEO approaches Leader LEO
3. Leader determines: Follower caught up
4. Leader adds follower to ISR
5. Leader sends AlterPartitionRequest to controller
6. Controller updates cluster metadata
7. ISR change propagated to all brokers

Leaving ISR

FOLLOWER LEAVES ISR WHEN:

Condition:
└─ (now - last fetch time) > replica.lag.time.max.ms

Process:
1. Leader detects: Follower hasn't fetched recently
2. Leader removes follower from ISR (local decision)
3. Leader sends AlterPartitionRequest to controller
4. Controller updates cluster metadata
5. ISR change propagated to all brokers

Reasons for lag:
├─ Follower broker slow (CPU, disk, network)
├─ Follower broker dead
├─ Network partition
└─ Replication too slow (high throughput)

ISR Shrink Example

TIMELINE:

T=0: ISR = [1, 2, 3]

T=5: Broker 3 experiences GC pause (10 seconds)

T=10: Leader detects: Broker 3 last fetch was 10s ago
      replica.lag.time.max.ms = 10000ms (10s)
      Leader decides: Remove broker 3 from ISR
      ISR = [1, 2]

T=15: Broker 3 recovers, resumes fetching

T=20: Broker 3 catches up (LEO >= HW)
      Leader decides: Add broker 3 back to ISR
      ISR = [1, 2, 3]

Result:
└─ Temporary ISR shrink during issue
└─ Automatic recovery when caught up

3. Configuration: replica.lag.time.max.ms

Critical Parameter

replica.lag.time.max.ms = 10000  # Default: 10 seconds

Meaning:
└─ Max time a follower can go without fetching before being removed from ISR

Impact:

Too low (e.g., 1000ms):
├─ Frequent ISR shrinks (transient issues trigger removal)
├─ ISR instability
└─ May reduce availability (fewer ISR members)

Too high (e.g., 60000ms):
├─ Slow followers remain in ISR longer
├─ acks=all waits longer for slow replicas
└─ Reduced write throughput

Recommended:
└─ 10-30 seconds (balance stability vs responsiveness)

4. High Watermark (HW)

Definition

High Watermark (HW):
└─ Offset of the last record fully replicated to all ISR members
└─ Records up to HW are "committed" and visible to consumers

Calculation:
└─ HW = min(LEO of all ISR members)

Example:
├─ Leader LEO: 1010
├─ Follower 1 LEO: 1009
├─ Follower 2 LEO: 1008
├─ ISR: [Leader, Follower 1, Follower 2]
└─ HW: 1008 (minimum of ISR LEOs)

HW Update Algorithm

LEADER UPDATES HW:

After each follower fetch:
1. Update follower's LEO (from FetchRequest offset)
2. Recalculate HW = min(all ISR member LEOs)
3. If HW advanced:
   ├─ Update leader's local HW
   └─ Include new HW in next FetchResponse to followers

Followers update HW:
└─ Upon receiving FetchResponse with HW
└─ Set local HW = leader's HW (from response)

HW Advancement Example

TIMELINE:

T=0:
├─ Leader appends records 1000-1010
├─ Leader LEO: 1010
├─ Follower 1 LEO: 999
├─ Follower 2 LEO: 998
└─ HW: 999 (min of ISR LEOs)

T=1: Follower 1 fetches
├─ Follower 1 LEO: 1010 (caught up)
├─ Leader recalculates: HW = min(1010, 1010, 998) = 998
└─ HW: 998 (no change yet)

T=2: Follower 2 fetches
├─ Follower 2 LEO: 1010 (caught up)
├─ Leader recalculates: HW = min(1010, 1010, 1010) = 1010
└─ HW: 1010 (advanced!)

Result:
└─ Records 999-1009 now visible to consumers

5. Consumer Visibility

Read-Your-Writes Semantics

CONSUMERS CAN ONLY READ UP TO HW:

Partition state:
├─ Leader LEO: 1010
├─ HW: 1008
└─ Records 1008, 1009 not yet committed

Consumer fetch at offset 1005:
├─ Can read: 1005, 1006, 1007
├─ Cannot read: 1008, 1009 (beyond HW)
└─ Fetch returns records 1005-1007, HW=1008

Why?
├─ Records beyond HW not yet fully replicated
├─ May be lost if leader fails
└─ Ensuring consumers see only "committed" data

Read Consistency Guarantee

KAFKA'S CONSISTENCY MODEL:

Read Committed Isolation:
└─ Consumers see only records up to HW
└─ HW = committed data
└─ Beyond HW = uncommitted (not visible)

Consistency:
├─ All consumers see same view (up to HW)
├─ Reads are consistent across replicas
└─ No dirty reads (uncommitted data)

Trade-off:
├─ Slightly stale reads (lag between LEO and HW)
└─ But strong consistency (committed data)

6. ISR and min.insync.replicas

Durability Configuration

min.insync.replicas = 2  # Broker or topic config

Meaning:
└─ Minimum ISR size required for writes to succeed

With acks=all:
├─ Producer write requires ISR size >= min.insync.replicas
├─ If ISR < min.insync.replicas:
│   └─ Producer receives NOT_ENOUGH_REPLICAS error
└─ Write rejected (not durable enough)

Example:
├─ Replicas: [1, 2, 3]
├─ ISR: [1] (brokers 2 and 3 down)
├─ min.insync.replicas: 2
└─ Write fails: NOT_ENOUGH_REPLICAS

Availability vs Durability

CONFIGURATION TRADE-OFF:

min.insync.replicas = 1:
├─ Writes succeed with only leader (ISR=[1])
├─ High availability (tolerates all followers down)
└─ Low durability (data on one replica)

min.insync.replicas = 2:
├─ Writes require leader + 1 follower (ISR=[1,2])
├─ Moderate availability (tolerates 1 follower down)
└─ High durability (data on 2 replicas)

min.insync.replicas = 3:
├─ Writes require all replicas (ISR=[1,2,3])
├─ Low availability (any failure blocks writes)
└─ Highest durability (data on 3 replicas)

Recommended:
└─ RF=3, min.insync.replicas=2
└─ Balances availability and durability

7. ISR Shrinks and Expands

ISR Shrink Metrics

JMX METRIC:

kafka.server:type=ReplicaManager,name=IsrShrinksPerSec

Meaning:
└─ Rate of ISR membership decreases

High shrink rate indicates:
├─ Followers frequently falling behind
├─ replica.lag.time.max.ms too aggressive
├─ Broker capacity issues (CPU, disk, network)
└─ Need to investigate and fix root cause

Monitoring:
└─ Alert if IsrShrinksPerSec > threshold (e.g., 10/min)

ISR Expand Metrics

JMX METRIC:

kafka.server:type=ReplicaManager,name=IsrExpandsPerSec

Meaning:
└─ Rate of ISR membership increases

High expand rate (after shrinks):
├─ Followers catching up after falling behind
├─ Recovery from temporary issues
└─ Positive sign (system healing)

Pattern:
└─ Shrinks followed by expands = transient issues
└─ Shrinks without expands = persistent failure

8. Under-Replicated Partitions

Definition

UNDER-REPLICATED PARTITION:

Condition:
└─ ISR size < Replication Factor

Example:
├─ Replicas: [1, 2, 3] (RF=3)
├─ ISR: [1, 2] (broker 3 out-of-sync)
└─ Under-replicated: Yes

Causes:
├─ Follower lagging
├─ Broker down
├─ Network issues
└─ Insufficient replication capacity

Impact:
└─ Reduced fault tolerance (fewer in-sync replicas)

Monitoring

JMX METRIC:

kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions

Meaning:
└─ Number of partitions where ISR < RF

Target: 0
└─ Any non-zero value = problem

Alerts:
├─ Immediate alert if > 0
├─ Investigate which partitions affected
└─ Identify root cause (slow broker, network, etc.)

9. HW Truncation

Why Truncation is Needed

SCENARIO: Leader Failover

Before failover:
├─ Old leader LEO: 1010
├─ Follower A LEO: 1008
├─ Follower B LEO: 1009
├─ HW: 1008 (committed)

Old leader dies:
└─ Follower A becomes new leader

Issue:
├─ Old leader had records 1009 (uncommitted)
├─ New leader (Follower A) doesn't have 1009
└─ If old leader rejoins, 1009 is divergent

Solution:
└─ Old leader truncates to HW (1008) upon rejoining
└─ Discards uncommitted record 1009

Truncation Process

WHEN FOLLOWER REJOINS AFTER LEADER CHANGE:

1. Follower fetches from new leader
2. New leader responds with:
   ├─ Diverging epoch (leader epoch changed)
   └─ New HW

3. Follower detects epoch mismatch
   ├─ Realizes: Leader changed while I was offline
   └─ Fetches OffsetsForLeaderEpoch from new leader

4. New leader returns: Safe truncation offset

5. Follower truncates local log to safe offset
   └─ Discards any divergent records

6. Follower resumes replication from truncation point

10. Leader Epoch

Purpose

LEADER EPOCH:

Definition:
└─ Counter incremented each time partition leader changes

Purpose:
└─ Detect leader changes
└─ Determine safe truncation point

Structure:
├─ Epoch: Integer counter (0, 1, 2, ...)
├─ Start offset: First offset for this epoch
└─ Stored in: leader-epoch-checkpoint file

Epoch-Based Truncation

USING EPOCHS:

Old leader log:
├─ Epoch 0: Offsets 0-999
├─ Epoch 1: Offsets 1000-1050 (old leader)
└─ LEO: 1051

New leader log:
├─ Epoch 0: Offsets 0-999
├─ Epoch 2: Offsets 1000-1040 (new leader, different epoch!)
└─ LEO: 1041

Old leader rejoins:
1. Fetches OffsetsForLeaderEpoch(epoch=1)
2. New leader responds: "Epoch 1 doesn't exist, truncate to 1000"
3. Old leader truncates offsets 1000-1050
4. Old leader starts replicating from offset 1000 (epoch 2)

11. Key Takeaways

You've learned:

✅ ISR = replicas caught up within replica.lag.time.max.ms
✅ High Watermark = min(ISR member LEOs) — last committed offset
✅ Consumers read up to HW — committed data only
✅ ISR shrinks when followers lag or fail
✅ ISR expands when followers catch up
✅ min.insync.replicas controls durability threshold
✅ Under-replicated partitions = ISR < RF (needs investigation)
✅ HW truncation resolves divergence after leader change
✅ Leader epochs enable safe truncation
✅ ISR metrics (shrinks, expands, under-replicated) critical for monitoring

Summary

ISR and HW define Kafka's durability boundaries:

ISR Mechanics =
├─ Dynamic membership (shrinks/expands)
├─ Based on replica.lag.time.max.ms
├─ Only ISR members participate in acks=all
└─ Protects availability (slow replicas don't block)

High Watermark =
├─ HW = min(ISR member LEOs)
├─ Advances as followers catch up
├─ Defines "committed" data
└─ Consumer visibility boundary

Durability Configuration =
├─ RF=3, min.insync.replicas=2 (recommended)
├─ Balances availability and durability
├─ NOT_ENOUGH_REPLICAS if ISR too small
└─ Prevents under-replicated writes

Truncation =
├─ Leader epochs detect leader changes
├─ Followers truncate divergent data
└─ Ensures log consistency after failover

Remember: ISR and HW are the heart of Kafka's replication guarantees. Understanding these mechanics is essential for designing durable and available Kafka systems.

Next up: Module 3.3 — Follower Fetching & Data Consistency — Deep dive into follower behavior.