6.2 Changelogs & Standby Replicas

Module Overview

Goal: Explain durability and availability of state stores.

What You'll Learn:

• Changelog topics and their purpose
• How changelogs work
• Standby replicas for high availability
• Write amplification implications
• Impact on performance
• Changelog configuration

What's NOT Covered Here:

• State store basics (covered in Module 6.1)
• Restore process (covered in Module 6.3)
• Exactly-once semantics (covered in Module 9.1)

Key Insight: Changelogs make state durable by backing it up to Kafka, enabling fault tolerance.

1. The State Durability Problem

Problem: Local State is Volatile

Scenario without changelogs:

Instance crashes:
├─ Local state lost (RocksDB on disk)
├─ Aggregation results gone
├─ Hours of processing lost
└─ Must reprocess from beginning

This is unacceptable for production

Solution: Changelog Topics

Changelog = Kafka topic that mirrors state store

Every state update:
├─ Written to local state store
└─ Also written to changelog topic

If instance crashes:
├─ State restored from changelog
├─ Recovery in minutes (not hours)
└─ No reprocessing needed

2. How Changelogs Work

The Backup Mechanism

Processing flow with changelog:

1. Record arrives
   ↓
2. Process and update state
   ├─ Write to RocksDB (local)
   └─ Write to changelog topic (Kafka)
   ↓
3. Commit offset
   └─ Only after changelog write confirms

State changes = events in Kafka
State can be reconstructed from changelog

Changelog Topic Structure

Changelog topic name:
{application.id}-{store-name}-changelog

Example:
my-app-user-counts-changelog

Characteristics:
├─ Compacted topic (log.cleanup.policy=compact)
├─ Same partition count as input topic
├─ Replication factor: 3 (configurable)
└─ Managed automatically by Kafka Streams

Visual Flow

┌────────────────────────────────────────────┐
│  KAFKA STREAMS INSTANCE                    │
│                                            │
│  ┌──────────────┐                          │
│  │ Input Record │                          │
│  └──────┬───────┘                          │
│         │                                  │
│         ↓                                  │
│  ┌──────────────┐                          │
│  │   Process    │                          │
│  └──────┬───────┘                          │
│         │                                  │
│         ↓                                  │
│  ┌──────────────┐      ┌────────────────┐ │
│  │ State Store  │─────→│ Changelog Topic│ │
│  │  (RocksDB)   │      │   (Kafka)      │ │
│  └──────────────┘      └────────────────┘ │
│                                            │
└────────────────────────────────────────────┘

Every state update goes to both

3. Changelog Configuration

Default Behavior

// Changelog enabled by default
KTable<String, Long> counts = stream
    .groupByKey()
    .count(Materialized.as("user-counts"));

// Changelog topic created automatically:
// my-app-user-counts-changelog

Explicit Configuration

// Configure changelog
KTable<String, Long> counts = stream
    .groupByKey()
    .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("user-counts")
        .withLoggingEnabled()  // Enable changelog (default)
        .withCachingEnabled()  // Cache before writing
    );

Disabling Changelog (NOT RECOMMENDED)

// Disable changelog (dangerous!)
KTable<String, Long> counts = stream
    .groupByKey()
    .count(Materialized.as("user-counts")
        .withLoggingDisabled()
    );

// ⚠️ State not backed up
// ⚠️ Lost on restart
// ⚠️ Only use for development/testing

Changelog Topic Configuration

// Configure changelog topic properties
Map<String, String> changelogConfig = new HashMap<>();
changelogConfig.put("min.insync.replicas", "2");
changelogConfig.put("compression.type", "lz4");

Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("user-counts")
    .withLoggingEnabled(changelogConfig);

4. Write Amplification

What is Write Amplification?

Write amplification = writing same data multiple times

Without changelog:
├─ 1 input record
└─ 1 write to state store

With changelog:
├─ 1 input record
├─ 1 write to state store
└─ 1 write to changelog topic
└─ Replication factor 3 = 3 copies in Kafka

Total writes: 1 + 1 + 3 = 5× amplification

Performance Impact

IMPLICATIONS:

1. Network I/O
   └─ More data sent to Kafka

2. Disk I/O
   └─ More writes on brokers

3. Throughput
   └─ Slower than stateless operations

4. Latency
   └─ Wait for changelog write confirmation

Trade-off: durability vs performance

Mitigation: Caching

// Cache state updates before writing to changelog
Materialized.as("user-counts")
    .withCachingEnabled();

// How it works:
// ├─ Buffer updates in memory
// ├─ Write to changelog periodically (commit.interval.ms)
// └─ Reduces changelog writes significantly

Example:
├─ Without cache: 1000 updates → 1000 changelog writes
└─ With cache:    1000 updates → 10 changelog writes

5. Changelog Compaction

Why Compaction?

Problem: Changelog grows forever

Without compaction:
├─ All updates kept
├─ Disk usage grows unbounded
└─ Restore takes longer

With compaction (default):
├─ Only latest value per key kept
├─ Bounded disk usage
└─ Faster restore

How Compaction Works

Changelog before compaction:
user1 → 1
user2 → 1
user1 → 2  ← Update
user1 → 3  ← Update
user2 → 2  ← Update

Changelog after compaction:
user1 → 3  ← Latest only
user2 → 2  ← Latest only

Old values removed

Compaction Configuration

Changelog topics use:
log.cleanup.policy=compact

Compaction runs periodically
└─ Controlled by broker settings
└─ min.cleanable.dirty.ratio
└─ segment.ms

6. Standby Replicas

The Availability Problem

Without standby replicas:

Task migrates to new instance:
├─ State must be restored from changelog
├─ Restore takes 5-30 minutes (for large state)
├─ Application unavailable during restore
└─ Poor user experience

Solution: Standby Replicas

Standby replica = warm backup on another instance

Configuration:
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);

Behavior:
├─ Primary instance: active processing
├─ Standby instance: maintains copy of state
└─ On failure: standby becomes active quickly

How Standby Replicas Work

Normal operation:

Instance 1 (PRIMARY):
├─ Task 0 → Partition 0
└─ State store (active)

Instance 2 (STANDBY):
├─ Task 0 replica
└─ State store (passive, kept up-to-date)

On Instance 1 failure:
├─ Instance 2 promotes standby to active
├─ No restore needed (state already present)
└─ Minimal downtime (~seconds, not minutes)

Standby Configuration

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

// Configure standby replicas
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);

// 0 = no standby (default)
// 1 = one standby per task
// 2 = two standbys per task (high availability)

Standby Replica Trade-offs

WITH STANDBY REPLICAS:
✅ Fast failover (seconds)
✅ High availability
✅ Better user experience
❌ More instances needed
❌ More storage (duplicate state)
❌ More network I/O (syncing standbys)

WITHOUT STANDBY REPLICAS:
✅ Fewer resources
✅ Lower cost
❌ Slow failover (minutes)
❌ Downtime during restore

7. Changelog vs Standby Replicas

Comparison

CHANGELOG:
├─ Purpose: Durability (survive crashes)
├─ Location: Kafka topic
├─ Recovery: Restore from topic (slow)
└─ Always enabled (default)

STANDBY REPLICA:
├─ Purpose: Availability (fast failover)
├─ Location: Another Streams instance
├─ Recovery: Promote standby (fast)
└─ Optional (disabled by default)

Both Work Together

Best practice: Use BOTH

Changelog:
└─ Ensures state never lost

Standby replicas:
└─ Ensures minimal downtime

Example:
┌─────────────────────────────────────────┐
│ Instance 1 (PRIMARY)                    │
│ ├─ Task 0 → State Store                 │
│ └─ Writes to changelog                  │
└─────────────────────────────────────────┘
           ↓ (changelog sync)
┌─────────────────────────────────────────┐
│ Kafka (CHANGELOG TOPIC)                 │
│ └─ Durability backup                    │
└─────────────────────────────────────────┘
           ↓ (standby sync)
┌─────────────────────────────────────────┐
│ Instance 2 (STANDBY)                    │
│ └─ Task 0 replica → State Store         │
└─────────────────────────────────────────┘

Standby Promotion Process (Failover)

What happens when the primary instance fails?

BEFORE FAILURE:
┌────────────────────────────────────┐
│ Instance 1 (PRIMARY)               │
│ ├─ Task 0: ACTIVE                  │
│ ├─ Processing records              │
│ └─ State: 1 GB                     │
└────────────────────────────────────┘

┌────────────────────────────────────┐
│ Instance 2 (STANDBY)               │
│ ├─ Task 0: STANDBY                 │
│ ├─ Replicating state               │
│ └─ State: 1 GB (up-to-date)        │
└────────────────────────────────────┘

FAILURE OCCURS:
Instance 1 crashes ❌

REBALANCE TRIGGERED:
├─ Group coordinator detects failure (session timeout)
├─ Rebalance begins (all instances pause)
└─ Tasks reassigned to remaining instances

STANDBY PROMOTION (Nearly Instant):
┌────────────────────────────────────┐
│ Instance 2                         │
│ ├─ Task 0: ACTIVE ← Promoted!      │
│ ├─ State already present (1 GB)    │
│ ├─ Catch up delta from changelog   │
│ │  (only records missed: ~seconds)  │
│ └─ Resume processing                │
└────────────────────────────────────┘

Total downtime: 5-15 seconds
(vs. minutes without standbys)

Promotion is nearly instant because:

1. State already exists locally on the standby instance
2. No full restoration needed (state is pre-loaded)
3. Only delta catch-up required (records since last sync)

Without Standby (for comparison):

FAILURE WITHOUT STANDBY:
Instance 1 crashes ❌

REBALANCE:
├─ Task 0 reassigned to Instance 2
└─ Instance 2 has NO state for Task 0

FULL RESTORATION:
├─ Read entire changelog from Kafka (1 GB)
├─ Rebuild state store from scratch
└─ Time: 2-5 minutes (or longer)

Total downtime: 2-5 minutes

Key Insight:

Standby replicas provide near-instantaneous failover by keeping a warm copy of state ready to take over. The promotion process is simply changing the task status from STANDBY to ACTIVE, followed by a small delta catch-up (typically seconds).

8. Monitoring Changelogs

Important Metrics

# Records written to changelog
kafka.streams:type=stream-task-metrics,
              task-id=*,
              name=changelog-records-write-rate

# Changelog topic lag
kafka.streams:type=stream-task-metrics,
              task-id=*,
              name=changelog-records-lag-max

# Restore progress
kafka.streams:type=stream-task-metrics,
              task-id=*,
              name=restore-consumer-records-read-rate

Checking Changelog Topics

# List changelog topics
kafka-topics.sh --bootstrap-server localhost:9092 --list | grep changelog

# Output:
# my-app-user-counts-changelog
# my-app-windowed-agg-changelog

# Describe changelog topic
kafka-topics.sh --bootstrap-server localhost:9092 \
  --describe --topic my-app-user-counts-changelog

# Check lag
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-app --describe

9. Changelog Best Practices

1. Always Enable Changelogs

// Good (default)
Materialized.as("my-store")
    .withLoggingEnabled()

// Bad (no durability)
Materialized.as("my-store")
    .withLoggingDisabled()

2. Use Caching

// Reduce changelog writes
Materialized.as("my-store")
    .withCachingEnabled()

// Configure cache size globally
props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 
    10 * 1024 * 1024);  // 10 MB per task

3. Configure Replication Factor

// Ensure changelog durability
Map<String, String> topicConfig = new HashMap<>();
topicConfig.put("min.insync.replicas", "2");

Materialized.as("my-store")
    .withLoggingEnabled(topicConfig);

4. Use Standby Replicas in Production

// For critical applications
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);

// Balance cost vs availability
// 1 standby = good for most use cases
// 2+ standbys = mission-critical only

10. Changelog Size Management

Factors Affecting Size

Changelog size depends on:

1. State store size
   └─ More state = larger changelog

2. Update frequency
   └─ More updates = more changelog writes

3. Compaction lag
   └─ Delay in compaction = larger changelog

4. Caching
   └─ More caching = smaller changelog

Reducing Changelog Size

1. Enable caching
   └─ Buffer updates before writing

2. Use appropriate commit interval
   └─ Less frequent commits = fewer writes

3. Ensure compaction runs
   └─ Check broker compaction settings

4. Use smaller values
   └─ Compact serialization format

11. Changelog Recovery

When Recovery Happens

State restored from changelog when:

1. Application first start (cold start)
   └─ No local state exists

2. Task reassignment
   └─ Task moved to different instance

3. State corruption
   └─ Local state invalid/corrupted

4. Manual state deletion
   └─ State directory deleted

Covered in detail in Module 6.3.

12. Troubleshooting Changelogs

Problem: Changelog Lag Growing

Symptom: Lag increases continuously

Causes:
├─ Broker under load
├─ Network issues
├─ Too many writes (disable caching)
└─ Insufficient broker capacity

Solutions:
1. Enable/increase caching
2. Scale Kafka brokers
3. Reduce state update frequency
4. Check network connectivity

Problem: Changelog Topic Missing

Symptom: Application fails to start

Cause: Changelog topic deleted manually

Solution:
1. Kafka Streams will recreate topic automatically
2. But state must be restored from scratch (slow)
3. Never delete changelog topics manually!

Problem: High Network I/O

Symptom: Network saturation

Cause: Too many changelog writes

Solutions:
1. Enable caching:
   .withCachingEnabled()
   
2. Increase cache size:
   props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 
       50 * 1024 * 1024);  // 50 MB
   
3. Increase commit interval:
   props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 30000);

13. Advanced Changelog Configuration

Custom Topic Configuration

// Fine-tune changelog topic
Map<String, String> topicConfig = new HashMap<>();
topicConfig.put("min.insync.replicas", "2");
topicConfig.put("compression.type", "lz4");
topicConfig.put("segment.ms", "3600000");  // 1 hour
topicConfig.put("retention.ms", "-1");     // Infinite
topicConfig.put("min.cleanable.dirty.ratio", "0.5");

Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("my-store")
    .withLoggingEnabled(topicConfig);

Per-Store Configuration

// Different configs for different stores
Materialized.as("hot-store")
    .withLoggingEnabled(Map.of(
        "min.insync.replicas", "3",  // High durability
        "compression.type", "none"    // Low latency
    ));

Materialized.as("cold-store")
    .withLoggingEnabled(Map.of(
        "min.insync.replicas", "2",
        "compression.type", "zstd"    // High compression
    ));

14. Mental Model Summary

Changelog Purpose

DURABILITY:
├─ Local state backed up to Kafka
├─ Survives instance crashes
└─ Enables recovery

MECHANISM:
├─ Every state update → changelog write
├─ Changelog topic = compacted
└─ State reconstructed from changelog

TRADE-OFF:
├─ Durability ✅
├─ Write amplification ❌
└─ Mitigate with caching

Standby Replicas Purpose

AVAILABILITY:
├─ Warm backup on another instance
├─ Fast failover (seconds)
└─ Minimal downtime

MECHANISM:
├─ Primary: active processing
├─ Standby: passive, kept in sync
└─ On failure: standby promoted

TRADE-OFF:
├─ Fast failover ✅
├─ Extra resources ❌
└─ Use for critical applications

The Complete Picture

┌─────────────────────────────────────────────┐
│  Instance 1 (ACTIVE)                        │
│  ├─ Process records                         │
│  ├─ Update state store                      │
│  └─ Write to changelog                      │
└─────────────────────────────────────────────┘
                ↓
┌─────────────────────────────────────────────┐
│  Kafka Cluster (CHANGELOG)                  │
│  ├─ Durability backup                       │
│  ├─ Compacted topic                         │
│  └─ Replicated (RF=3)                       │
└─────────────────────────────────────────────┘
                ↓
┌─────────────────────────────────────────────┐
│  Instance 2 (STANDBY)                       │
│  ├─ Maintain state replica                  │
│  ├─ Sync from changelog                     │
│  └─ Ready for promotion                     │
└─────────────────────────────────────────────┘

15. Questions You Should Be Able to Answer

After this module, you should be able to answer:

✅ What is a changelog topic?
✅ Why are changelogs needed?
✅ What is write amplification?
✅ What are standby replicas?
✅ When should I use standby replicas?
✅ How does caching reduce changelog writes?
✅ What's the difference between changelog and standby replica?
✅ How do I monitor changelog lag?

16. What's Next?

Now that you understand changelogs and standby replicas, the next module will cover:

• Module 6.3: Restore & recovery — what happens when state needs to be rebuilt
• Module 11.1: Interactive queries — querying distributed state
• Module 12.1: Monitoring — operational metrics

Summary

You learned:

• ✅ Changelog topics back up state to Kafka for durability
• ✅ Every state update written to both local store and changelog
• ✅ Write amplification is the cost of durability (mitigated by caching)
• ✅ Standby replicas provide fast failover for high availability
• ✅ Caching buffers updates to reduce changelog writes
• ✅ Compaction keeps changelog size bounded (latest value per key)
• ✅ Changelogs (durability) + standbys (availability) work together
• ✅ Configuration: enable logging, use caching, set standby count

Next up: Module 6.3 — Restore & Recovery — Learn how state is rebuilt from changelogs when failures occur.