KafkaIQ6.3 Restore & Recovery
Module Overview
Goal: Understand state restoration when instances fail or tasks migrate.
What You'll Learn:
- When restoration happens
- How state is restored from changelogs
- Restore phases and timing
- Monitoring restore progress
- Optimizing restore performance
- Handling restore failures
What's NOT Covered Here:
- State store internals (covered in Module 6.1)
- Changelog mechanics (covered in Module 6.2)
- Standby replicas (covered in Module 6.2)
Key Insight: State restoration is automatic but can be slow for large state — understanding the process helps you optimize recovery time.
1. When Does Restoration Happen?
Cold Start
Scenario: Application starts for the first time
├─ No local state exists
├─ Must restore from changelog
└─ Happens once per instance
Duration: Depends on changelog size
Task Reassignment
Scenario: Task moves to different instance
Causes:
├─ Rebalance (new instance added/removed)
├─ Instance crash/restart
├─ Manual scale-up/scale-down
└─ Partition reassignment
New instance must restore task's state
State Corruption
Scenario: Local state invalid or corrupted
Causes:
├─ Disk failure
├─ Process killed mid-write
├─ RocksDB corruption
└─ Manual state deletion
Kafka Streams detects corruption → triggers restore
Manual State Cleanup
# Delete local state directory
rm -rf /tmp/kafka-streams/my-app/0_0
# On restart, state restored from changelog
2. The Restoration Process
High-Level Flow
1. Instance assigned task
↓
2. Check local state directory
↓
3. If no state (or corrupted):
├─ Enter RESTORING state
├─ Create restore consumer
├─ Read changelog topic from beginning
├─ Rebuild state store
└─ Enter RUNNING state
↓
4. Resume processing input records
Visual Representation
┌──────────────────────────────────────────────┐
│ TASK ASSIGNMENT │
│ ├─ Instance 2 assigned Task 0 │
│ └─ No local state exists │
└──────────────────────────────────────────────┘
↓
┌──────────────────────────────────────────────┐
│ ENTER RESTORING STATE │
│ ├─ Create restore consumer │
│ ├─ Subscribe to changelog: │
│ │ my-app-user-counts-changelog │
│ └─ Seek to beginning │
└──────────────────────────────────────────────┘
↓
┌──────────────────────────────────────────────┐
│ READ CHANGELOG RECORDS │
│ ├─ Fetch records in batches │
│ ├─ Write to local state store (RocksDB) │
│ └─ Track progress │
└──────────────────────────────────────────────┘
↓
┌──────────────────────────────────────────────┐
│ RESTORATION COMPLETE │
│ ├─ State store fully rebuilt │
│ ├─ Close restore consumer │
│ └─ Enter RUNNING state │
└──────────────────────────────────────────────┘
↓
┌──────────────────────────────────────────────┐
│ RESUME PROCESSING │
│ └─ Process input records normally │
└──────────────────────────────────────────────┘
3. Restore Consumer
What is the Restore Consumer?
Restore consumer = internal consumer for reading changelogs
NOT the same as:
├─ Main consumer (reads input topics)
└─ Standby consumer (keeps standby replicas up-to-date)
Purpose:
└─ Read changelog topic to rebuild state
Configuration
// Restore consumer configs
props.put(StreamsConfig.RESTORE_CONSUMER_PREFIX +
ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000);
props.put(StreamsConfig.RESTORE_CONSUMER_PREFIX +
ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 52428800); // 50 MB
// Prefix all restore consumer configs with:
// "restore.consumer."
Restore vs Main Consumer
MAIN CONSUMER:
├─ Reads input topics
├─ Processes records one-by-one
└─ Commits offsets
RESTORE CONSUMER:
├─ Reads changelog topics
├─ Bulk loads state store
└─ No offset commits (reads entire changelog)
4. Restoration Phases
Phase 1: Discovery
Task assigned:
├─ Check if local state exists
├─ Compare local offsets with changelog
└─ Determine if restore needed
Decision:
├─ Local state up-to-date → skip restore
└─ Local state missing/stale → start restore
Phase 2: Initialization
Prepare for restore:
├─ Create restore consumer
├─ Subscribe to changelog topic
├─ Create/clear local state directory
└─ Initialize state store
Phase 3: Bulk Load
Read changelog in batches:
├─ Fetch records (default: 1000/poll)
├─ Write to state store
├─ Track progress
└─ Repeat until end of changelog
This is the slowest phase
Phase 4: Catchup
Near end of changelog:
├─ Catch up to latest offset
├─ Apply remaining updates
└─ Ensure state fully restored
Phase 5: Completion
Restoration complete:
├─ Close restore consumer
├─ Mark task as RUNNING
└─ Resume normal processing
5. Restore Time Estimation
Factors Affecting Restore Time
1. Changelog size (records/bytes)
└─ Larger state = longer restore
2. Network bandwidth
└─ Fetching changelog from Kafka
3. Disk I/O
└─ Writing to RocksDB
4. State store type
└─ RocksDB vs in-memory
5. Restore consumer config
└─ fetch.max.bytes, max.poll.records
6. Broker load
└─ Broker response time
Typical Restore Times
Small state (< 1 GB):
└─ 1-5 minutes
Medium state (1-10 GB):
└─ 5-20 minutes
Large state (10-100 GB):
└─ 20-60 minutes
Very large state (> 100 GB):
└─ Hours
Formula (Rough Estimate)
Restore time ≈ (Changelog size in bytes) / (Network throughput)
Example:
├─ Changelog: 10 GB
├─ Network: 100 MB/s
└─ Time: 10 GB / 100 MB/s = 100 seconds ≈ 2 minutes
(Add 20-50% overhead for writes)
6. Monitoring Restoration
Task State
Task states:
├─ CREATED: Task assigned, not started
├─ RESTORING: Restoring state from changelog
├─ RUNNING: Processing input records
└─ SUSPENDED: Temporarily paused (rebalance)
Monitor state transitions:
kafka.streams:type=stream-thread-metrics,
thread-id=*,
task-id=*,
name=task-restoring-count
Restore Progress Metrics
# Records read from changelog
kafka.streams:type=stream-task-metrics,
task-id=*,
name=restore-consumer-records-read-total
# Bytes read from changelog
kafka.streams:type=stream-task-metrics,
task-id=*,
name=restore-consumer-bytes-read-total
# Restore lag (records behind)
kafka.streams:type=stream-task-metrics,
task-id=*,
name=restore-consumer-records-lag-max
# Restore rate
kafka.streams:type=stream-task-metrics,
task-id=*,
name=restore-consumer-records-read-rate
Logging Restore Progress
Kafka Streams logs:
INFO Starting restore for task 0_0 (org.apache.kafka.streams.processor.internals.Task)
INFO Restoring state for store user-counts from beginning
INFO Restored 100000 records for store user-counts (10% complete)
INFO Restored 500000 records for store user-counts (50% complete)
INFO Restoration complete for task 0_0
Custom Monitoring
// StateRestoreListener to track progress
props.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp/kafka-streams");
KafkaStreams streams = new KafkaStreams(topology, props);
streams.setGlobalStateRestoreListener(new StateRestoreListener() {
@Override
public void onRestoreStart(TopicPartition topicPartition,
String storeName,
long startingOffset,
long endingOffset) {
System.out.println("Restore started: " + storeName);
System.out.println("Records to restore: " +
(endingOffset - startingOffset));
}
@Override
public void onBatchRestored(TopicPartition topicPartition,
String storeName,
long batchEndOffset,
long numRestored) {
System.out.println("Restored batch: " + numRestored + " records");
}
@Override
public void onRestoreEnd(TopicPartition topicPartition,
String storeName,
long totalRestored) {
System.out.println("Restore complete: " + storeName);
System.out.println("Total restored: " + totalRestored);
}
});
streams.start();
7. Optimizing Restore Performance
1. Increase Fetch Size
// Fetch more data per poll
props.put(StreamsConfig.RESTORE_CONSUMER_PREFIX +
ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 104857600); // 100 MB
props.put(StreamsConfig.RESTORE_CONSUMER_PREFIX +
ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 10485760); // 10 MB
2. Increase Batch Size
// Process more records per batch
props.put(StreamsConfig.RESTORE_CONSUMER_PREFIX +
ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 5000);
3. Use Standby Replicas
// Fast failover without full restore
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
// Standby already has state → promotes immediately
// No restore needed!
4. Enable Caching
// Fewer changelog records to restore
props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG,
10 * 1024 * 1024); // 10 MB
// Caching reduces changelog size → faster restore
5. Tune RocksDB
// RocksDB configuration for faster bulk loading
RocksDBConfigSetter configSetter = new RocksDBConfigSetter() {
@Override
public void setConfig(String storeName, Options options, Map<String, Object> configs) {
// Optimize for bulk writes during restore
options.setMaxWriteBufferNumber(4);
options.setWriteBufferSize(64 * 1024 * 1024); // 64 MB
options.setDisableAutoCompactions(true); // Disable during restore
}
};
props.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG,
configSetter.getClass());
6. Network Optimization
Ensure low latency to Kafka brokers:
├─ Co-locate Streams app and brokers (same data center)
├─ Use high-bandwidth network (10 Gbps+)
└─ Avoid cross-region restoration
8. Standby Replicas vs Full Restore
Restore Without Standby
Failover scenario:
Instance 1 crashes:
├─ Task reassigned to Instance 2
├─ Instance 2 has no state
├─ Full restore from changelog (20 minutes)
└─ Application unavailable for 20 minutes
User experience: Poor
Restore With Standby
Failover scenario:
Instance 1 crashes:
├─ Task reassigned to Instance 2 (standby)
├─ Instance 2 already has state
├─ Minor catchup (< 1 minute)
└─ Application available quickly
User experience: Good
Standby Configuration
// Enable standby replicas
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
Cost:
├─ 2× storage (primary + standby)
├─ 2× instances (minimum)
└─ More network I/O
Benefit:
├─ Fast failover
└─ High availability
9. Handling Restore Failures
Failure Scenarios
1. Changelog topic deleted
└─ State lost permanently
2. Insufficient disk space
└─ Restore fails mid-process
3. RocksDB corruption during restore
└─ Retry from beginning
4. Network partition
└─ Restore stalled
Automatic Retry
Kafka Streams retries restore automatically:
On failure:
├─ Log error
├─ Clean up partial state
├─ Retry restore from beginning
└─ Continue retrying indefinitely
Manual intervention needed if:
├─ Changelog deleted
├─ Insufficient resources
└─ Configuration issue
Handling Missing Changelog
Scenario: Changelog topic deleted manually
Result:
├─ Restore fails (topic not found)
├─ Application cannot start
└─ State lost permanently
Solution:
1. Reset application (lose state):
kafka-streams-application-reset.sh \
--application-id my-app \
--bootstrap-servers localhost:9092
2. Reprocess from beginning (rebuild state):
├─ Delete local state
├─ Application restores from input topics
└─ Slow but recovers state
Insufficient Disk Space
Symptom:
├─ Restore fails mid-process
└─ Error: "No space left on device"
Solution:
1. Free up disk space
2. Increase disk capacity
3. Delete old state stores
4. Reduce state store size (TTL, compaction)
10. Restore During Rebalance
What Happens?
Rebalance triggered:
├─ Tasks suspended (SUSPENDED state)
├─ Some tasks reassigned
└─ New instance restores assigned tasks
During restore:
├─ Processing stalled
├─ Consumer not polling
└─ Application partially unavailable
Minimizing Rebalance Impact
1. Use cooperative rebalancing (default in recent versions)
└─ Only affected tasks stopped
2. Use standby replicas
└─ Fast promotion, minimal restore
3. Avoid frequent rebalances
├─ Set higher session.timeout.ms
└─ Stable cluster (no frequent scaling)
11. State Directory Management
Default State Directory
Default location:
/tmp/kafka-streams/{application.id}/{task.id}
Example:
/tmp/kafka-streams/my-app/0_0/rocksdb/user-counts
Problems with /tmp:
├─ Cleared on reboot (full restore needed)
└─ Limited space
Custom State Directory
// Use dedicated disk with more space
props.put(StreamsConfig.STATE_DIR_CONFIG,
"/var/kafka-streams-state");
Benefits:
├─ Survives reboots
├─ More disk space
└─ Better performance (SSD)
Cleaning Up State
# Safe: Delete state for specific task
rm -rf /var/kafka-streams-state/my-app/0_0
# Dangerous: Delete all state
rm -rf /var/kafka-streams-state/my-app
# On restart, all state restored from changelog
12. Restore vs Reprocessing
Restore (from Changelog)
Source: Changelog topic
Process:
├─ Read compacted changelog
├─ Only latest value per key
└─ Fast (bulk load)
Use case:
└─ Normal failover/reassignment
Reprocessing (from Input)
Source: Input topic(s)
Process:
├─ Read all input records from beginning
├─ Recompute all aggregations
└─ Slow (full processing)
Use case:
├─ Changelog deleted
├─ Application reset
└─ Logic change (reprocessing needed)
Comparison
RESTORE (CHANGELOG):
✅ Fast (minutes)
✅ Automatic
✅ No logic execution
❌ Requires changelog
REPROCESSING (INPUT):
✅ Always possible
✅ Rebuilds lost changelogs
❌ Slow (hours/days)
❌ Must replay all input
13. Restore Best Practices
1. Use Standby Replicas for Critical Apps
// Production configuration
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
// Balance cost vs availability
2. Monitor Restore Progress
// Implement StateRestoreListener
streams.setGlobalStateRestoreListener(new MyRestoreListener());
// Log progress, send metrics to monitoring system
3. Optimize Restore Consumer Config
// Increase throughput
props.put(StreamsConfig.RESTORE_CONSUMER_PREFIX +
ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 104857600);
props.put(StreamsConfig.RESTORE_CONSUMER_PREFIX +
ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 5000);
4. Use Dedicated State Directory
// Avoid /tmp
props.put(StreamsConfig.STATE_DIR_CONFIG,
"/var/kafka-streams-state");
// Use SSD for better performance
4.5 Same-Host Restart Advantage (Cached State Store)
Key Optimization: When a Kafka Streams instance restarts on the same host, it can reuse existing local state, avoiding full restoration from the changelog!
How It Works:
SCENARIO 1: Restart on Same Host
────────────────────────────────────
1. Instance stops gracefully
└─ Local state remains at /var/kafka-streams-state/
2. Instance restarts on same host
├─ Check local state directory
├─ Find existing state for Task 0
├─ Validate state is valid
└─ Check offset vs changelog
3. Delta Catch-Up (Only New Records)
├─ Local offset: 1,000,000
├─ Changelog offset: 1,000,500
└─ Read only 500 records (delta)
4. Resume Processing
└─ Total downtime: Seconds!
SCENARIO 2: Restart on Different Host
────────────────────────────────────
1. Instance stops
2. Task reassigned to different host
└─ No local state exists
3. Full Restoration
├─ Read entire changelog (1,000,500 records)
└─ Rebuild state from scratch
4. Resume Processing
└─ Total downtime: Minutes!
Performance Comparison:
| Scenario | State Size | Restoration Time | Network I/O |
|---|---|---|---|
| Same host | 10 GB | 5-15 seconds | Only delta (~MB) |
| Different host | 10 GB | 2-5 minutes | Full state (10 GB) |
Best Practices for Same-Host Advantage:
1. Use Persistent Disks (Not Ephemeral)
├─ Docker: Mount volume to /var/kafka-streams-state
├─ Kubernetes: Use PersistentVolumeClaim
└─ Avoid /tmp (cleared on restart)
2. Configure Pod/Container Affinity
├─ Kubernetes: Use StatefulSets (not Deployments)
├─ Tie instance to specific node
└─ Ensures same-host restart
3. Set Dedicated State Directory
props.put(StreamsConfig.STATE_DIR_CONFIG,
"/persistent-disk/kafka-streams-state");
4. Graceful Shutdown
├─ Call streams.close() properly
├─ Allows state to flush cleanly
└─ State directory remains valid
Soccer Team Analogy:
Think of state restoration like substituting players:
SAME-HOST RESTART (Player returns after timeout):
├─ Player (task) leaves field temporarily
├─ Player comes back to same position
├─ Player remembers game state (score, strategy)
└─ Resumes immediately (no briefing needed)
DIFFERENT-HOST RESTART (New player substituted in):
├─ New player enters field
├─ Doesn't know game state (score, strategy)
├─ Coach must brief player (restore from changelog)
└─ Takes time before player is effective
Kubernetes Example (StatefulSet):
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: kafka-streams-app
spec:
serviceName: kafka-streams
replicas: 3
template:
spec:
containers:
- name: streams-app
image: my-streams-app:1.0
volumeMounts:
- name: state-store
mountPath: /var/kafka-streams-state
volumeClaimTemplates:
- metadata:
name: state-store
spec:
accessModes: ["ReadWriteOnce"]
resources:
requests:
storage: 50Gi # Persistent disk for state
Key Takeaway:
When possible, restart instances on the same host to leverage cached local state. This reduces restoration time from minutes to seconds and significantly improves availability during rolling restarts or failure recovery.
5. Enable Caching
// Reduce changelog size
props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG,
10 * 1024 * 1024);
// Faster restore (fewer records)
6. Never Delete Changelog Topics Manually
NEVER:
kafka-topics.sh --delete --topic my-app-user-counts-changelog
If deleted:
├─ State lost permanently (if no local state)
└─ Must reprocess from input topics
7. Plan for Restore Time
Expect restore time in:
├─ Cold starts (first deployment)
├─ Scaling events (new instances)
└─ Failover (without standby)
Capacity planning:
└─ Ensure instances can handle load during restore
14. Troubleshooting Restore Issues
Symptom: Restore Taking Too Long
Causes:
├─ Large changelog (100+ GB)
├─ Slow network
├─ Low fetch settings
└─ Broker under load
Solutions:
1. Increase restore consumer fetch configs
2. Use standby replicas
3. Scale Kafka brokers
4. Check network bandwidth
Symptom: Restore Stuck at 99%
Cause: Waiting for final records
Changelog compaction not complete:
├─ Old records still present
└─ Must read to end
Solution: Wait (usually resolves in minutes)
Symptom: Restore Keeps Failing
Logs:
ERROR Failed to restore state for task 0_0
Causes:
├─ RocksDB corruption
├─ Disk space issue
├─ Configuration problem
└─ Changelog topic issue
Solutions:
1. Check disk space
2. Check RocksDB logs
3. Verify changelog topic exists
4. Delete corrupted local state
15. Mental Model Summary
The Restore Lifecycle
┌──────────────────────────────────────┐
│ TASK ASSIGNED │
│ └─ Check local state │
└──────────────────────────────────────┘
↓
[State exists?]
↙ ↘
YES NO
↓ ↓
[SKIP RESTORE] [START RESTORE]
↓ ↓
↓ ┌───────────────────┐
↓ │ RESTORING │
↓ │ ├─ Read changelog │
↓ │ ├─ Rebuild state │
↓ │ └─ Track progress │
↓ └───────────────────┘
↓ ↓
└──────┬─────┘
↓
┌───────────────┐
│ RUNNING │
│ └─ Process │
└───────────────┘
Fast Restore vs Slow Restore
FAST RESTORE (Standby):
├─ State already present
├─ Minor catchup (~seconds)
└─ High availability
SLOW RESTORE (From Changelog):
├─ Read entire changelog
├─ Rebuild state (~minutes)
└─ Application unavailable
16. Questions You Should Be Able to Answer
After this module, you should be able to answer:
✅ When does state restoration happen?
✅ What is the restore consumer?
✅ What are the phases of restoration?
✅ How long does restore typically take?
✅ How do I monitor restore progress?
✅ How can I optimize restore performance?
✅ What's the benefit of standby replicas for restore?
✅ What happens if the changelog topic is deleted?
17. What's Next?
Now that you understand state restoration, the next module will cover:
- Module 7.1: Topology design & internal topics — understanding repartition topics and topology structure
- Module 11.1: Interactive queries — querying state during and after restore
- Module 12.1: Monitoring — observing restore metrics in production
Summary
You learned:
- ✅ State restored from changelog when task reassigned or state missing
- ✅ Restore has 5 phases: discovery, initialization, bulk load, catchup, completion
- ✅ Restore consumer reads changelog to rebuild state
- ✅ Restore time depends on changelog size, network, and disk I/O
- ✅ Standby replicas eliminate full restore (fast failover)
- ✅ Monitor restore with metrics and
StateRestoreListener - ✅ Optimize restore with fetch configs, caching, and RocksDB tuning
- ✅ Never delete changelog topics manually (state lost permanently)
Next up: Module 7.1 — Topology Design & Internal Topics — Learn how Kafka Streams structures processing logic and manages repartition topics.