KafkaIQ12.1 Monitoring & Operating Kafka Streams Applications
Module Overview
Goal: Master monitoring and operations for Kafka Streams in production.
What You'll Learn:
- Key metrics to monitor (JMX)
- Application state and health
- Performance monitoring
- Consumer lag tracking
- Rebalance detection
- Alerting strategies
- Operational best practices
- Troubleshooting
What's NOT Covered Here:
- Configuration tuning (covered in Module 13.1)
- Error handling (covered in Module 10.1)
- State restoration details (covered in Module 6.3)
Key Insight: Production observability is critical — monitor state, lag, throughput, and errors to ensure reliability.
1. Monitoring Overview
Why Monitor?
Production requirements:
├─ Detect failures quickly
├─ Understand performance
├─ Capacity planning
├─ SLA compliance
└─ Troubleshooting
Key questions:
├─ Is the application healthy?
├─ Is it processing data?
├─ Is it falling behind?
├─ Are there errors?
└─ Is a rebalance happening?
Monitoring Layers
1. APPLICATION LEVEL:
└─ State, threads, errors
2. TASK LEVEL:
└─ Processing rate, lag, commits
3. STATE STORE LEVEL:
└─ State size, restore progress
4. TOPOLOGY LEVEL:
└─ Processing per node
5. INFRASTRUCTURE:
└─ CPU, memory, disk, network
2. Application State
KafkaStreams States
CREATED:
└─ Application initialized, not started
REBALANCING:
└─ Consumer group rebalancing (tasks reassigning)
RUNNING:
└─ Healthy, processing records
PENDING_SHUTDOWN:
└─ Shutdown initiated, cleaning up
NOT_RUNNING:
└─ Application stopped
ERROR:
└─ Unrecoverable error occurred
State Listener
KafkaStreams streams = new KafkaStreams(topology, props);
streams.setStateListener((newState, oldState) -> {
logger.info("State transition: {} -> {}", oldState, newState);
switch (newState) {
case RUNNING:
logger.info("Application is running");
// Expose metrics, enable health check
break;
case REBALANCING:
logger.warn("Rebalancing in progress");
// May affect latency, alert if frequent
break;
case ERROR:
logger.error("Application in error state");
// Alert immediately
break;
default:
break;
}
});
streams.start();
Health Check Endpoint
@GetMapping("/health")
public ResponseEntity<HealthStatus> health() {
State state = streams.state();
if (state == State.RUNNING) {
return ResponseEntity.ok(new HealthStatus("healthy", state.name()));
} else if (state == State.REBALANCING) {
return ResponseEntity.status(503)
.body(new HealthStatus("degraded", state.name()));
} else {
return ResponseEntity.status(503)
.body(new HealthStatus("unhealthy", state.name()));
}
}
3. JMX Metrics
Enabling JMX
JMX enabled by default
Java options:
-Dcom.sun.management.jmxremote
-Dcom.sun.management.jmxremote.port=9999
-Dcom.sun.management.jmxremote.authenticate=false
-Dcom.sun.management.jmxremote.ssl=false
Query via:
├─ JConsole
├─ VisualVM
├─ Prometheus JMX Exporter
└─ Custom monitoring tools
Metric Categories
kafka.streams:
├─ type=stream-metrics (application-level)
├─ type=stream-thread-metrics (per thread)
├─ type=stream-task-metrics (per task)
├─ type=stream-processor-node-metrics (per node)
└─ type=stream-state-metrics (per state store)
4. Application-Level Metrics
Key Metrics
# Application state
kafka.streams:type=stream-metrics,
client-id=*,
name=state
# Number of threads
kafka.streams:type=stream-metrics,
client-id=*,
name=alive-stream-threads
# Failed threads
kafka.streams:type=stream-metrics,
client-id=*,
name=failed-stream-threads
# Processing rate (records/sec)
kafka.streams:type=stream-metrics,
client-id=*,
name=process-rate
Example: Querying JMX
// Programmatically access metrics
KafkaStreams streams = new KafkaStreams(topology, props);
streams.start();
Map<MetricName, ? extends Metric> metrics = streams.metrics();
for (Map.Entry<MetricName, ? extends Metric> entry : metrics.entrySet()) {
MetricName name = entry.getKey();
Metric metric = entry.getValue();
if (name.name().equals("process-rate")) {
System.out.println("Process rate: " + metric.metricValue());
}
}
5. Thread-Level Metrics
Key Metrics
# Thread state (running, dead, etc.)
kafka.streams:type=stream-thread-metrics,
thread-id=*,
name=thread-state
# Commit latency (ms)
kafka.streams:type=stream-thread-metrics,
thread-id=*,
name=commit-latency-avg
# Commit rate (commits/sec)
kafka.streams:type=stream-thread-metrics,
thread-id=*,
name=commit-rate
# Poll latency (ms)
kafka.streams:type=stream-thread-metrics,
thread-id=*,
name=poll-latency-avg
Monitoring Thread Health
Alert if:
├─ thread-state != "RUNNING" (for extended period)
├─ commit-latency-avg > threshold (e.g., 5 seconds)
├─ poll-latency-avg > threshold (e.g., 2 seconds)
└─ commit-rate drops to 0 (no progress)
6. Task-Level Metrics
Key Metrics
# Records processed per task
kafka.streams:type=stream-task-metrics,
task-id=*,
name=process-rate
# Records lag (behind latest offset)
kafka.streams:type=stream-task-metrics,
task-id=*,
name=consumer-lag-max
# Restore progress (state restoration)
kafka.streams:type=stream-task-metrics,
task-id=*,
name=restore-consumer-records-read-total
# Enforced processing (suppression)
kafka.streams:type=stream-task-metrics,
task-id=*,
name=suppression-buffer-size-current
Consumer Lag
Consumer lag = how far behind the latest offset
Healthy:
└─ Lag = 0 or small (< 1000 records)
Unhealthy:
└─ Lag growing (falling behind)
Causes:
├─ Slow processing
├─ Insufficient resources
├─ Rebalancing
└─ Downstream bottleneck
Alert if:
└─ Lag > threshold for X minutes
7. State Store Metrics
Key Metrics
# Number of entries in state store
kafka.streams:type=stream-state-metrics,
task-id=*,
store-id=*,
name=num-entries
# State store size (bytes)
kafka.streams:type=stream-state-metrics,
task-id=*,
store-id=*,
name=size-in-bytes
# Restore lag (during restoration)
kafka.streams:type=stream-task-metrics,
task-id=*,
name=restore-consumer-records-lag-max
Monitoring State Growth
Track:
├─ num-entries over time
├─ size-in-bytes over time
Alert if:
├─ Unbounded growth (memory leak?)
├─ Approaching disk/memory limits
└─ Restore taking too long (large state)
8. Rebalance Detection
Rebalance Metrics
# Rebalance start
kafka.streams:type=stream-thread-metrics,
thread-id=*,
name=rebalance-latency-total
# Rebalance duration (ms)
kafka.streams:type=stream-thread-metrics,
thread-id=*,
name=last-rebalance-seconds-ago
# Time since last rebalance
kafka.streams:type=stream-thread-metrics,
thread-id=*,
name=time-since-last-rebalance-ms
Detecting Frequent Rebalances
Problem: Rebalances too frequent
Causes:
├─ max.poll.interval.ms too low
├─ session.timeout.ms too low
├─ Instances joining/leaving frequently
└─ Network instability
Detection:
└─ Count rebalances per hour
Alert if:
└─ > 10 rebalances/hour (indicates instability)
Rebalance Listener
streams.setGlobalStateRestoreListener(new StateRestoreListener() {
@Override
public void onRestoreStart(TopicPartition topicPartition, String storeName,
long startingOffset, long endingOffset) {
logger.info("Restore started: store={}, partition={}, records={}",
storeName, topicPartition, endingOffset - startingOffset);
}
@Override
public void onBatchRestored(TopicPartition topicPartition, String storeName,
long batchEndOffset, long numRestored) {
logger.debug("Batch restored: {}", numRestored);
}
@Override
public void onRestoreEnd(TopicPartition topicPartition, String storeName,
long totalRestored) {
logger.info("Restore complete: store={}, total={}", storeName, totalRestored);
}
});
9. Performance Monitoring
Throughput Metrics
# Records processed per second
kafka.streams:type=stream-metrics,
client-id=*,
name=process-rate
# Records produced per second
kafka.streams:type=stream-metrics,
client-id=*,
name=produce-rate
# Records consumed per second
kafka.streams:type=stream-metrics,
client-id=*,
name=consume-rate
Latency Metrics
# End-to-end latency (input to output)
kafka.streams:type=stream-processor-node-metrics,
task-id=*,
processor-node-id=*,
name=process-latency-avg
# Commit latency
kafka.streams:type=stream-thread-metrics,
thread-id=*,
name=commit-latency-avg
Resource Utilization
Monitor at OS level:
CPU:
└─ Track CPU % per instance
Memory:
├─ Heap usage (JVM)
├─ Off-heap (RocksDB)
└─ Total memory
Disk:
├─ Disk usage (state stores)
├─ Disk I/O (read/write)
└─ IOPS
Network:
├─ Bandwidth (in/out)
└─ Connections
10. Alerting Strategy
Critical Alerts (Page Immediately)
1. Application state = ERROR
└─ Application cannot recover
2. All threads dead
└─ No processing happening
3. Consumer lag growing rapidly
└─ Falling behind, data loss risk
4. Disk space < 10%
└─ State stores at risk
5. Failed stream threads > 0
└─ Threads crashing
Warning Alerts (Investigate)
1. Application state = REBALANCING (> 5 minutes)
└─ Long rebalance, may affect SLA
2. Consumer lag > threshold (e.g., 10000 records)
└─ Falling behind, needs attention
3. Commit latency > 5 seconds
└─ Slow commits, may indicate issue
4. Restore taking > 30 minutes
└─ Large state or slow network
5. Frequent rebalances (> 10/hour)
└─ Instability, investigate
Informational Alerts (FYI)
1. Application state = RUNNING
└─ Healthy, processing normally
2. State restore started
└─ Expected on startup/rebalance
3. Consumer lag = 0
└─ Fully caught up
11. Dashboards
Grafana Dashboard Example
Panels:
1. Application State
└─ Gauge: state (RUNNING, REBALANCING, ERROR)
2. Throughput
├─ Line chart: process-rate
├─ Line chart: consume-rate
└─ Line chart: produce-rate
3. Consumer Lag
└─ Line chart: consumer-lag-max (per task)
4. Thread Health
├─ Gauge: alive-stream-threads
└─ Gauge: failed-stream-threads
5. Commit Metrics
├─ Line chart: commit-latency-avg
└─ Line chart: commit-rate
6. State Store Size
├─ Line chart: num-entries
└─ Line chart: size-in-bytes
7. Rebalances
└─ Counter: rebalance count per hour
8. Resource Usage
├─ Line chart: CPU %
├─ Line chart: Memory %
└─ Line chart: Disk %
12. Logging Best Practices
Structured Logging
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
private static final Logger logger = LoggerFactory.getLogger(MyApp.class);
// Structured logging with context
logger.info("Processing record: key={}, topic={}, partition={}, offset={}",
record.key(), context.topic(), context.partition(), context.offset());
logger.error("Processing failed: key={}, error={}",
record.key(), exception.getMessage(), exception);
Log Levels
ERROR:
├─ Unrecoverable errors
├─ Application crashes
└─ State corruption
WARN:
├─ Recoverable errors
├─ Retries
├─ Degraded performance
└─ Frequent rebalances
INFO:
├─ Application lifecycle (start, stop, rebalance)
├─ State transitions
└─ Configuration
DEBUG:
├─ Record-level processing
├─ State store operations
└─ Internal details (development only)
Log Aggregation
Use centralized logging:
├─ ELK Stack (Elasticsearch, Logstash, Kibana)
├─ Splunk
├─ Datadog
└─ CloudWatch (AWS)
Benefits:
├─ Search across all instances
├─ Correlate events
├─ Alert on log patterns
└─ Long-term retention
13. Troubleshooting Common Issues
Issue 1: High Consumer Lag
Symptom: Lag growing continuously
Causes:
├─ Processing too slow (bottleneck)
├─ Insufficient parallelism (too few tasks/instances)
├─ Downstream service slow (external calls)
└─ Resource constraints (CPU, memory)
Solutions:
1. Optimize processing logic
2. Scale horizontally (more instances)
3. Increase partitions (more parallelism)
4. Optimize external calls (caching, batching)
5. Add more CPU/memory
Issue 2: Frequent Rebalances
Symptom: Rebalances every few minutes
Causes:
├─ max.poll.interval.ms too low
├─ Processing too slow
├─ Instances flapping (crash loops)
└─ Network instability
Solutions:
1. Increase max.poll.interval.ms
2. Optimize processing (faster polling)
3. Fix application crashes
4. Stabilize network
5. Use static membership (group.instance.id)
Issue 3: Application Stuck in REBALANCING
Symptom: Rebalancing for extended period
Causes:
├─ Large state restoration (hours)
├─ Standby replicas not configured
├─ Slow network to Kafka
└─ Insufficient broker resources
Solutions:
1. Use standby replicas (fast failover)
2. Optimize restore consumer config
3. Improve network bandwidth
4. Scale Kafka brokers
5. Reduce state store size (TTL, compaction)
Issue 4: High Memory Usage
Symptom: OOM errors or high heap usage
Causes:
├─ Large state stores (in-memory)
├─ Large cache (cache.max.bytes.buffering)
├─ Too many tasks per instance
└─ Memory leak (application code)
Solutions:
1. Use RocksDB (off-heap)
2. Reduce cache size
3. Scale horizontally (more instances)
4. Profile for memory leaks
5. Increase heap size (if needed)
Issue 5: Slow Processing
Symptom: Low throughput, high latency
Causes:
├─ Blocking I/O (external calls)
├─ Large batch sizes
├─ Slow state store (disk I/O)
└─ Inefficient processing logic
Solutions:
1. Async external calls
2. Optimize batch sizes
3. Use SSD for state stores
4. Optimize processing logic
5. Profile CPU usage
14. Capacity Planning
Estimating Resources
Factors:
1. Input rate (records/sec)
└─ Determines throughput needs
2. State size (GB)
└─ Determines disk/memory needs
3. Processing complexity
└─ Determines CPU needs
4. Partition count
└─ Determines max parallelism
5. Latency requirements
└─ Determines commit interval
Sizing Guidelines
SMALL WORKLOAD (< 10K records/sec):
├─ 2-4 instances
├─ 2-4 threads per instance
├─ 4 GB heap
├─ 50 GB disk (state stores)
└─ 4 vCPUs
MEDIUM WORKLOAD (10K-100K records/sec):
├─ 5-10 instances
├─ 4-8 threads per instance
├─ 8 GB heap
├─ 100-500 GB disk
└─ 8 vCPUs
LARGE WORKLOAD (> 100K records/sec):
├─ 10+ instances
├─ 8+ threads per instance
├─ 16 GB heap
├─ 500+ GB disk (SSD)
└─ 16+ vCPUs
15. Operational Best Practices
1. Monitor Application State
// Always set state listener
streams.setStateListener((newState, oldState) -> {
logger.info("State: {} -> {}", oldState, newState);
metricsRegistry.recordStateChange(newState);
});
2. Configure Standby Replicas
// For production, use standby replicas
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
// Fast failover, minimal downtime
3. Use Graceful Shutdown
// Always close gracefully
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
streams.close(Duration.ofSeconds(30));
}));
4. Set Uncaught Exception Handler
// Handle unexpected errors
streams.setUncaughtExceptionHandler((thread, throwable) -> {
logger.error("Uncaught exception: {}", throwable.getMessage(), throwable);
return StreamThreadExceptionResponse.SHUTDOWN_APPLICATION;
});
5. Enable Metrics Export
Use Prometheus JMX Exporter or similar:
1. Add JMX exporter agent
2. Export to Prometheus
3. Visualize in Grafana
4. Set up alerts
Example:
java -javaagent:jmx_exporter.jar=9090:config.yaml -jar my-app.jar
6. Test in Staging
Before production:
├─ Load test (simulate production load)
├─ Failure test (kill instances, broker failures)
├─ Rebalance test (add/remove instances)
└─ Restore test (measure restoration time)
Verify:
├─ Throughput meets requirements
├─ Latency within SLA
├─ Recovery time acceptable
└─ Resource usage reasonable
16. Mental Model Summary
Monitoring Layers
┌───────────────────────────────────────────────┐
│ APPLICATION STATE │
│ └─ RUNNING, REBALANCING, ERROR │
└───────────────┬───────────────────────────────┘
↓
┌───────────────▼───────────────────────────────┐
│ THREAD METRICS │
│ └─ Alive threads, commit latency, poll rate │
└───────────────┬───────────────────────────────┘
↓
┌───────────────▼───────────────────────────────┐
│ TASK METRICS │
│ └─ Process rate, consumer lag, restore │
└───────────────┬───────────────────────────────┘
↓
┌───────────────▼───────────────────────────────┐
│ STATE STORE METRICS │
│ └─ Size, entries, restore lag │
└───────────────────────────────────────────────┘
Monitor from top to bottom
Key Metrics Summary
HEALTH:
├─ Application state (RUNNING)
├─ Alive threads (> 0)
└─ Failed threads (= 0)
THROUGHPUT:
├─ Process rate (records/sec)
├─ Consume rate (records/sec)
└─ Produce rate (records/sec)
LAG:
├─ Consumer lag (< threshold)
└─ Growing? Alert!
LATENCY:
├─ Commit latency (ms)
├─ Processing latency (ms)
└─ Poll latency (ms)
STATE:
├─ State store size (entries, bytes)
└─ Restore progress (%)
17. Questions You Should Be Able to Answer
After this module, you should be able to answer:
✅ What are the KafkaStreams application states?
✅ What are the most important metrics to monitor?
✅ How do I detect consumer lag?
✅ How do I detect frequent rebalances?
✅ What JMX metrics should I track?
✅ When should I alert vs just log?
✅ How do I troubleshoot high consumer lag?
✅ How do I estimate resource requirements?
18. What's Next?
Now that you understand monitoring and operations, the final module will cover:
- Module 13.1: Advanced configuration & tool selection — performance tuning, when to use Kafka Streams vs alternatives
Summary
You learned:
- ✅ Monitor application state (RUNNING, REBALANCING, ERROR) with state listener
- ✅ Key JMX metrics: process-rate, consumer-lag-max, commit-latency-avg
- ✅ Consumer lag indicates if falling behind (alert if growing)
- ✅ Detect frequent rebalances (> 10/hour indicates instability)
- ✅ Track thread health (alive threads, failed threads)
- ✅ Monitor state store size (entries, bytes) for growth
- ✅ Set up critical alerts (ERROR state, high lag, disk space)
- ✅ Use graceful shutdown, standby replicas, and exception handlers
- ✅ Troubleshoot common issues: lag, rebalances, OOM, slow processing
- ✅ Capacity planning based on input rate, state size, and latency requirements
Next up: Module 13.1 — Advanced Configuration & Choosing the Right Tool — Final module covering performance tuning and when to use Kafka Streams.