KafkaIQ6.1 Consumer Lifecycle & Poll Loop Internals
Module Overview
Goal: Explain how the consumer actually runs — the mechanics behind poll().
What You'll Learn:
- poll() mechanics and what happens inside
max.poll.interval.msdeep divesession.timeout.msvs heartbeat timing- Backpressure from slow processing
- Pause/resume functionality
- Single-thread rule and thread safety
- Consumer lifecycle states
What's NOT Covered Here:
- Performance tuning recommendations (covered in Module 7.1)
- Error handling patterns (covered in Module 8.1)
- Rebalancing details (covered in Module 4.1)
Key Insight: Understanding poll() internals explains why configs exist, not just how to set them.
1. The Poll Loop Heartbeat
Basic Poll Loop
while (running) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
processRecord(record);
}
}
What poll() actually does:
- Send heartbeat (if needed)
- Check for rebalance
- Fetch records from broker
- Update position
- Return records to application
2. What Happens Inside poll()
Poll() Execution Flow
consumer.poll(Duration.ofMillis(100))
│
├─ 1. Ensure Coordinator Connection
│ └─ Find and connect to Group Coordinator
│
├─ 2. Check Group Membership
│ ├─ Am I in a rebalance?
│ └─ Do I need to rejoin?
│
├─ 3. Send Heartbeat (if interval elapsed)
│ └─ Tell coordinator "I'm alive"
│
├─ 4. Check poll() Interval
│ └─ Has max.poll.interval.ms elapsed?
│
├─ 5. Handle Rebalance (if triggered)
│ ├─ onPartitionsRevoked()
│ ├─ JoinGroup + SyncGroup
│ └─ onPartitionsAssigned()
│
├─ 6. Fetch Records
│ ├─ Send FetchRequest to brokers
│ ├─ Wait up to timeout
│ └─ Receive FetchResponse
│
├─ 7. Update Position
│ └─ Advance current position
│
└─ 8. Return Records
└─ ConsumerRecords<K, V>
Two Threads in Action
┌──────────────────────────────────────────────┐
│ APPLICATION THREAD │
│ │
│ while (true) { │
│ records = consumer.poll(...) ←─────────┼─┐
│ │ │
│ for (record : records) { │ │
│ process(record); │ │
│ } │ │
│ } │ │
└──────────────────────────────────────────────┘ │
│
┌──────────────────────────────────────────────┐ │
│ BACKGROUND THREAD (Heartbeat) │ │
│ │ │
│ while (consumer active) { │ │
│ sleep(heartbeat.interval.ms); │ │
│ sendHeartbeat(); ──────────────────────────┘
│ } │
└──────────────────────────────────────────────┘
Key: Heartbeats sent in background, poll() in foreground.
3. max.poll.interval.ms Explained
What It Controls
props.put("max.poll.interval.ms", 300000); // 5 minutes (default)
Definition: Maximum time between poll() calls before consumer is considered dead.
Why It Exists
Problem without it:
Consumer polls
Consumer processes for 10 hours (stuck!)
Heartbeats still sent (background thread)
Coordinator thinks consumer is alive
But consumer is not making progress!
Solution:
max.poll.interval.ms enforces:
"You must call poll() regularly or you're kicked out"
How It Works
Time Consumer Action Coordinator
────┼────────────────────────────┼─────────────────────
0s│ poll() #1 │ Timer reset
1s│ Processing... │ ⏱️ Timer: 1s
2s│ Processing... │ ⏱️ Timer: 2s
3s│ poll() #2 │ Timer reset
4s│ Processing... │ ⏱️ Timer: 1s
300s│ (no poll for 5 minutes) │ ⏱️ Timer: 300s
│ │ TIMEOUT!
│ │ Kick consumer
│ │ Trigger rebalance
Tuning max.poll.interval.ms
Too Short
props.put("max.poll.interval.ms", 10000); // 10 seconds
props.put("max.poll.records", 500);
// Processing takes 15 seconds per batch
for (record : records) { // 500 records
slowProcess(record); // 30ms each
}
// Total: 15 seconds → TIMEOUT!
Result: False timeouts, rebalance storms.
Too Long
props.put("max.poll.interval.ms", 3600000); // 1 hour
// Consumer actually stuck
while (true) {
records = consumer.poll(...);
infiniteLoop(); // BUG!
}
// Takes 1 hour to detect!
Result: Slow failure detection.
Right Size
max.poll.interval.ms >= (max.poll.records × processing_time_per_record) + buffer
Example:
max.poll.records = 500
Processing time = 100ms per record
Total processing = 500 × 100ms = 50 seconds
Buffer = 2x safety margin
max.poll.interval.ms = 100 seconds (100,000ms)
4. session.timeout.ms vs heartbeat.interval.ms
The Relationship
props.put("session.timeout.ms", 10000); // 10 seconds
props.put("heartbeat.interval.ms", 3000); // 3 seconds
Rule:
heartbeat.interval.ms < session.timeout.ms / 3
How They Work Together
Time Consumer Coordinator
────┼─────────────────────┼──────────────────────────
0s│ ❤️ Heartbeat │ Session timer reset
3s│ ❤️ Heartbeat │ Session timer reset
6s│ ❤️ Heartbeat │ Session timer reset
9s│ ❤️ Heartbeat │ Session timer reset
12s│ (heartbeat stops) │ ⏱️ Timer: 3s
15s│ │ ⏱️ Timer: 6s
18s│ │ ⏱️ Timer: 9s
21s│ │ ⏱️ Timer: 12s (>10s!)
│ │ TIMEOUT! Consumer dead
Key: session.timeout.ms is the grace period for missed heartbeats.
Two Timeout Dimensions
┌─────────────────────────────────────────────┐
│ session.timeout.ms │
│ ├─ Detects: Process crash │
│ ├─ Method: Heartbeat monitoring │
│ └─ Typical: 10 seconds │
│ │
│ max.poll.interval.ms │
│ ├─ Detects: Application hang │
│ ├─ Method: poll() call monitoring │
│ └─ Typical: 5 minutes │
└─────────────────────────────────────────────┘
Both must pass for consumer to stay alive.
5. Backpressure from Slow Processing
The Problem
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
slowDatabaseCall(record); // 500ms per record!
}
}
If max.poll.records = 1000:
Processing time = 1000 records × 500ms = 500 seconds (8+ minutes!)
max.poll.interval.ms = 300 seconds (5 minutes)
Result: TIMEOUT during processing!
Solution 1: Reduce max.poll.records
props.put("max.poll.records", 100); // Instead of 500
// Now processing = 100 × 500ms = 50 seconds (under 5 min limit)
Solution 2: Increase max.poll.interval.ms
props.put("max.poll.interval.ms", 600000); // 10 minutes
Trade-off: Slower failure detection.
Solution 3: Optimize Processing
// Before: Synchronous
for (record : records) {
slowDatabaseCall(record); // 500ms
}
// After: Batch processing
List<Record> batch = new ArrayList<>();
for (record : records) {
batch.add(record);
if (batch.size() >= 100) {
database.batchInsert(batch); // 1 second for 100
batch.clear();
}
}
Result: 10x faster processing.
6. Pause and Resume
Use Case
Problem: Consumer receiving data faster than you can process.
Solution: Temporarily pause consumption.
Pausing Partitions
// Pause specific partitions
Collection<TopicPartition> partitions = consumer.assignment();
consumer.pause(partitions);
// Consumer.poll() now returns empty (no new records)
while (backlogNotEmpty()) {
processBacklog();
}
// Resume when ready
consumer.resume(partitions);
Pause Behavior
Normal:
poll() → [records] → process → poll() → [records] → ...
Paused:
poll() → [] (empty) → wait → poll() → [] (empty) → ...
↑
No records fetched, but:
- Heartbeats still sent
- Rebalances still handled
- Consumer stays alive
Resumed:
poll() → [records] → process → ...
Key: Pause stops fetching, not heartbeats.
Practical Pause Pattern
int backlogSize = 0;
int maxBacklog = 10000;
while (running) {
// Check backlog
if (backlogSize > maxBacklog) {
consumer.pause(consumer.assignment());
} else if (backlogSize < maxBacklog / 2) {
consumer.resume(consumer.assignment());
}
// Poll (may be empty if paused)
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
queue.add(record);
backlogSize++;
}
}
7. Single-Thread Rule
The Rule
KafkaConsumer is NOT thread-safe.
One consumer = One thread
What's Allowed
// ✅ GOOD: Single thread
Thread consumerThread = new Thread(() -> {
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
while (running) {
ConsumerRecords<String, String> records = consumer.poll(...);
for (ConsumerRecord<String, String> record : records) {
processRecord(record);
}
}
consumer.close();
});
consumerThread.start();
What's NOT Allowed
// ❌ BAD: Multiple threads sharing consumer
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
Thread thread1 = new Thread(() -> {
consumer.poll(...); // Thread 1
});
Thread thread2 = new Thread(() -> {
consumer.poll(...); // Thread 2 (CONCURRENT!)
});
// Result: ConcurrentModificationException or worse!
Exception: Wakeup
Only wakeup() can be called from another thread:
final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
// Shutdown hook in another thread
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
consumer.wakeup(); // ✅ Safe from another thread
}));
// Main consumer thread
try {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(...);
// process...
}
} catch (WakeupException e) {
// Triggered by wakeup() call
} finally {
consumer.close();
}
8. Consumer Lifecycle States
State Machine
CREATED
↓ (subscribe/assign)
INITIALIZING
↓ (first poll)
JOINING
↓ (join complete)
STABLE
↓ (rebalance triggered)
PREPARING_REBALANCE
↓ (rebalance complete)
STABLE
↓ (close called)
CLOSING
↓
CLOSED
State Transitions
consumer = new KafkaConsumer<>(props); // CREATED
consumer.subscribe(topics); // INITIALIZING
consumer.poll(...); // JOINING → STABLE
// consuming... // STABLE
// rebalance happens // PREPARING_REBALANCE → STABLE
consumer.close(); // CLOSING → CLOSED
9. Graceful Shutdown
Proper Shutdown Pattern
final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
final AtomicBoolean running = new AtomicBoolean(true);
// Shutdown hook
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
running.set(false);
consumer.wakeup();
}));
try {
consumer.subscribe(Collections.singletonList("orders"));
while (running.get()) {
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
processRecord(record);
}
consumer.commitSync();
}
} catch (WakeupException e) {
// Expected during shutdown
if (running.get()) throw e; // Unexpected wakeup
} finally {
// Final commit before closing
try {
consumer.commitSync();
} finally {
consumer.close();
}
}
10. Mental Model Summary
The Poll Loop
poll() = Heartbeat + Rebalance Check + Fetch + Update Position
Critical Configs:
├─ max.poll.interval.ms (how long between polls)
├─ session.timeout.ms (how long without heartbeat)
├─ heartbeat.interval.ms (heartbeat frequency)
└─ max.poll.records (batch size)
Rules:
├─ One consumer = One thread
├─ Must call poll() regularly
├─ Pause/resume for backpressure
└─ Graceful shutdown with wakeup()
Key Takeaways
✅ poll() does more than fetch records
✅ Two timeouts: session (heartbeat) and poll (progress)
✅ Slow processing → reduce max.poll.records or increase max.poll.interval.ms
✅ Pause/resume for backpressure control
✅ KafkaConsumer is single-threaded
✅ Use wakeup() for shutdown
11. What's Next?
- Module 7.1: Configuration and performance tuning
- Module 8.1: Error handling patterns
- Module 9.1: Monitoring and observability
- Module 10.1: Practical patterns
Summary
You learned:
- ✅
poll()handles heartbeats, rebalancing, and fetching - ✅
max.poll.interval.msprevents stuck consumers - ✅
session.timeout.msdetects dead consumers - ✅ Backpressure solved by reducing batch size or increasing timeout
- ✅ Pause/resume provides flow control
- ✅ One consumer per thread (thread safety rule)
- ✅ Graceful shutdown with
wakeup()
Next up: Configuration & Performance Tuning (Module 7.1) — Learn how to tune consumers for optimal performance.