KafkaIQ

3.3 Follower Fetching & Data Consistency

Module Overview

Goal: Deep dive into follower replica behavior and consistency mechanics.

What You'll Learn:

  • Follower fetch loop internals
  • Lag detection and recovery
  • Divergence handling and log truncation
  • Consistency protocols (leader epochs)
  • Fetch configurations and tuning
  • Edge cases in replication

What's NOT Covered Here:

  • Basic replication model (covered in Module 3.1)
  • ISR rules and HW calculation (covered in Module 3.2)

Key Insight: Followers are not passive — they actively manage log consistency, detect divergence, and recover from failures using sophisticated protocols.

1. Follower Fetch Loop

Core Algorithm

FOLLOWER MAIN LOOP (simplified):

while (broker is alive) {
    for each partition where this broker is follower {
        1. Determine fetch offset (current LEO)
        2. Build FetchRequest
        3. Send to leader broker
        4. Wait for FetchResponse
        5. Validate response
        6. Append records to local log
        7. Update local LEO
        8. Update local HW (from response)
        9. Handle errors if any
    }
    Sleep briefly (if no data fetched)
}

Continuous process, runs throughout broker lifetime

Fetch Request Construction

FETCH REQUEST DETAILS:

FetchRequest:
├─ replica id: This follower's broker ID
├─ max wait ms: replica.fetch.wait.max.ms (500ms default)
├─ min bytes: replica.fetch.min.bytes (1 byte default)
└─ fetch data:
    └─ For each partition:
        ├─ topic: "orders"
        ├─ partition: 0
        ├─ fetch offset: follower's current LEO
        ├─ max bytes: replica.fetch.max.bytes
        └─ log start offset: for leader validation

Sent to: Leader broker for this partition

2. Fetch Response Processing

Response Contents

FETCH RESPONSE:

FetchResponse:
└─ partition data:
    ├─ partition: orders-0
    ├─ high watermark: 1050 (leader's HW)
    ├─ last stable offset: 1050 (for transactions)
    ├─ log start offset: 0
    ├─ error code: NONE (or error)
    ├─ records: RecordBatch[] (data to replicate)
    └─ aborted transactions: [] (if any)

Follower Processing Steps

UPON RECEIVING FETCH RESPONSE:

1. Check error code
   ├─ If NONE: Proceed
   ├─ If NOT_LEADER_FOR_PARTITION:
   │   └─ Refresh metadata, retry with new leader
   ├─ If OFFSET_OUT_OF_RANGE:
   │   └─ Fetch OffsetsForLeaderEpoch, truncate if needed
   └─ Other errors: Log and retry

2. Validate records
   ├─ Check offset sequence
   ├─ Verify CRCs
   └─ Ensure no gaps

3. Append to local log
   ├─ Write to active segment
   ├─ Update offset and time indexes
   └─ Fsync (based on flush policy)

4. Update follower LEO
   └─ LEO = last appended offset + 1

5. Update follower HW
   └─ HW = min(leader's HW, follower's LEO)

6. Acknowledge internally
   └─ Ready for next fetch

3. Lag Detection

Measuring Lag

LAG METRICS:

1. Record lag (messages):
   └─ Leader LEO - Follower LEO

2. Time lag (seconds):
   └─ Leader timestamp - Follower timestamp of last fetch

Leader tracks both:
├─ Per-follower LEO (from fetch offset)
├─ Per-follower last fetch time
└─ Determines ISR membership

Follower-Side Lag Awareness

FOLLOWER KNOWS ITS LAG:

From FetchResponse:
├─ Leader HW: 1050
├─ Records returned: offsets 1000-1049
└─ Follower calculates: Still catching up (LEO < HW)

Follower behavior:
├─ If lagging: Fetch immediately (no wait)
├─ If caught up: Respect replica.fetch.wait.max.ms
└─ Aggressive fetching when behind

4. Divergence Detection

How Divergence Occurs

SCENARIO:

1. Leader writes: offsets 1000-1010
2. Follower fetches up to 1008
3. Leader fails before follower gets 1009-1010
4. New leader elected (doesn't have 1009-1010)
5. Old leader rejoins as follower
   └─ Has divergent records 1009-1010

Problem:
└─ Old leader's 1009-1010 ≠ new leader's log
└─ Must be truncated

Leader Epoch Protocol

LEADER EPOCH TRACKING:

Each leader change increments epoch:
├─ Leader 1, Epoch 0: Offsets 0-999
├─ Leader 1, Epoch 1: Offsets 1000-1010
└─ Leader 2, Epoch 2: Offsets 1011-... (new leader)

leader-epoch-checkpoint file:
├─ Epoch 0 → Offset 0
├─ Epoch 1 → Offset 1000
└─ Epoch 2 → Offset 1011

Follower uses epochs to detect divergence

OffsetsForLeaderEpoch Request

WHEN FOLLOWER SUSPECTS DIVERGENCE:

1. Follower sends OffsetsForLeaderEpochRequest
   └─ "What offset started epoch X?"

2. Leader responds with epoch boundaries

3. Follower compares with local epochs
   ├─ If match: Continue replicating
   └─ If mismatch: Truncate divergent data

Example:
├─ Follower has: Epoch 1 → Offset 1000
├─ Leader responds: Epoch 1 doesn't exist (now Epoch 2)
└─ Follower truncates everything from offset 1000 onward

5. Log Truncation

Truncation Process

WHEN FOLLOWER MUST TRUNCATE:

1. Determine truncation offset
   └─ From OffsetsForLeaderEpoch response

2. Truncate local log
   ├─ Remove segments beyond truncation point
   ├─ Truncate active segment if needed
   └─ Rebuild indexes

3. Update follower LEO
   └─ LEO = truncation offset

4. Resume replication
   └─ Fetch from truncation offset

5. Catch up to leader
   └─ Eventually rejoin ISR

Truncation Example

TIMELINE:

T=0:
├─ Old leader: [Epoch 1: 1000-1010], LEO: 1011
├─ Follower A: [Epoch 1: 1000-1008], LEO: 1009
└─ Follower B: [Epoch 1: 1000-1009], LEO: 1010

T=1: Old leader fails, Follower A becomes new leader (Epoch 2)
├─ New leader (A): [Epoch 2: 1009-...], LEO: 1009
└─ Follower B still has: offsets 1009 from Epoch 1 (divergent!)

T=2: Old leader rejoins as follower
├─ Fetches from new leader
├─ Detects: Epoch 2 started at offset 1009
├─ Realizes: My offsets 1009-1010 (Epoch 1) are divergent
├─ Truncates to offset 1009
└─ Resumes replication from 1009 (Epoch 2)

Result:
└─ Both old leader and Follower B truncated to offset 1009
└─ All replicas consistent with new leader

6. Fetch Configurations

replica.fetch.max.bytes

CONFIGURATION:

replica.fetch.max.bytes = 1048576  # 1 MB default

Meaning:
└─ Max bytes follower fetches per request (per partition)

Impact:

Too low:
├─ More fetch requests (higher overhead)
├─ Slower replication (small batches)
└─ Higher network round-trip costs

Too high:
├─ Larger memory buffers
├─ Longer processing time per batch
└─ May delay replication of other partitions

Recommended:
└─ 1-10 MB (balance throughput and latency)

replica.fetch.wait.max.ms

CONFIGURATION:

replica.fetch.wait.max.ms = 500  # 500ms default

Meaning:
└─ Max time leader waits before responding if no data available

Behavior:

If leader has data:
└─ Respond immediately

If leader has no new data:
├─ Wait up to replica.fetch.wait.max.ms
├─ If data arrives: Respond immediately
└─ If timeout: Respond with empty result

Impact:

Low value (e.g., 100ms):
├─ Lower replication latency (frequent checks)
├─ More fetch requests (higher CPU)
└─ Better for low-latency requirements

High value (e.g., 2000ms):
├─ Higher replication latency
├─ Fewer fetch requests (lower CPU)
└─ Risk: Follower appears lagging

replica.fetch.min.bytes

CONFIGURATION:

replica.fetch.min.bytes = 1  # 1 byte default

Meaning:
└─ Min bytes of data before leader responds

Behavior:

If available data >= min bytes:
└─ Respond immediately

If available data < min bytes:
├─ Wait up to replica.fetch.wait.max.ms
└─ Then respond with whatever is available

Impact:

Low value (1 byte):
├─ Leader responds with any data (even small)
├─ Lower latency
└─ More fetch requests

High value (e.g., 10 KB):
├─ Leader waits for more data (batching)
├─ Higher latency
└─ Fewer, larger fetch requests

7. Fetch Throttling

Replication Quota

THROTTLING CONFIGURATION:

follower.replication.throttled.rate = 10485760  # 10 MB/s

Applied to:
└─ Follower fetch bandwidth

Mechanism:
├─ Follower tracks bytes fetched per second
├─ If rate exceeds quota: Delay next FetchRequest
└─ Effectively limits replication speed

Use cases:
├─ Prevent replication from saturating network
├─ Prioritize client traffic over replication
└─ Controlled partition reassignment

8. Consistency Guarantees

Read-After-Write Consistency

SCENARIO:

1. Producer writes record to leader
   └─ acks=all, write committed (HW advanced)

2. Consumer reads from follower (if enabled)
   └─ Follower HW may lag behind leader HW

Potential issue:
├─ Leader HW: 1010 (record visible)
├─ Follower HW: 1008 (record NOT visible)
└─ Consumer reading from follower sees stale data

Kafka's default:
└─ Consumers read from leader only (avoids issue)

KIP-392 (optional):
└─ Enable follower fetching for consumers
└─ Must tolerate slightly stale reads

Monotonic Reads

GUARANTEE:

Kafka ensures:
└─ Consumer never sees offset X, then fails to see X later

How:
├─ HW never decreases (except unclean leader election)
├─ Once record visible (HW advanced), stays visible
└─ Monotonic read consistency

9. Edge Cases

Follower Restarts

WHEN FOLLOWER RESTARTS:

1. Load partition logs from disk
2. Determine current LEO (last appended offset)
3. Send FetchRequest from LEO
4. Leader responds with records from LEO onward
5. Follower appends and catches up

Issue:
├─ Follower may have divergent data (if leader changed during downtime)
└─ Use OffsetsForLeaderEpoch to detect and truncate

Leader Restarts

WHEN LEADER RESTARTS:

1. Leader failover occurs (new leader elected)
2. Old leader restarts as follower
3. Fetches from new leader
4. Detects epoch change
5. Truncates divergent data
6. Replicates from new leader
7. Eventually catches up, rejoins ISR

Network Partitions

NETWORK PARTITION SCENARIO:

1. Follower loses connectivity to leader
2. Leader marks follower out-of-sync (after replica.lag.time.max.ms)
3. ISR shrinks (follower removed)
4. Network heals
5. Follower resumes fetching
6. Follower catches up
7. ISR expands (follower re-added)

Key point:
└─ ISR membership dynamically adjusts to network health

10. Key Takeaways

You've learned:

✅ Follower fetch loop continuously pulls data from leader
✅ FetchRequest includes follower LEO — leader tracks follower progress
✅ FetchResponse includes leader HW — follower updates local HW
✅ Lag measured by time and records — determines ISR membership
✅ Divergence detected via leader epochs — OffsetsForLeaderEpoch protocol
✅ Truncation resolves divergence — followers discard conflicting data
✅ Fetch configs control latency vs throughput — wait.max.ms, max.bytes, min.bytes
✅ Fetch throttling limits replication bandwidth — protects network
✅ Monotonic reads guaranteed — HW never decreases (except unclean election)
✅ Followers recover automatically — after restarts or network issues

Summary

Follower replicas are active participants in data consistency:

Fetch Mechanics =
├─ Pull-based (follower initiates)
├─ Continuous loop (fetch, append, repeat)
├─ LEO and HW updated per fetch
└─ Aggressive fetching when lagging

Consistency Protocols =
├─ Leader epochs detect leader changes
├─ OffsetsForLeaderEpoch identifies divergence
├─ Truncation removes conflicting data
└─ Eventually consistent across ISR

Configuration Tuning =
├─ replica.fetch.max.bytes (batch size)
├─ replica.fetch.wait.max.ms (latency vs throughput)
├─ replica.fetch.min.bytes (batching threshold)
└─ follower.replication.throttled.rate (bandwidth limit)

Remember: Followers aren't just copying data — they actively maintain consistency using leader epochs, detect divergence, truncate when necessary, and recover from failures automatically.

Next up: Module 4.1 — Partition Leadership — Defining leader and follower responsibilities.