9.1 Monitoring, Lag & Observability

Module Overview

Goal: Make consumers operable through comprehensive monitoring and observability.

What You'll Learn:

Consumer lag (what it means and what it doesn't)
JMX metrics for consumers
Rebalance detection and monitoring
Stuck consumer detection
Alerting strategies
Operational dashboards
Troubleshooting guide

Key Insight: You can't fix what you can't see — Monitoring is essential for production consumers.

1. Consumer Lag Explained

What is Consumer Lag?

Consumer Lag = Number of messages behind the latest offset

Partition:
[0][1][2][3][4][5][6][7][8][9]
                    ↑         ↑
              Consumer   Log End
              Offset: 6  Offset: 10

Lag = 10 - 6 = 4 messages

Formula:

Lag = Log End Offset - Consumer Offset

What Lag Tells You

Low Lag (0-100 messages)

Lag: 5 messages

✅ Good: Consumer keeping up with producers
✅ Near real-time processing

Medium Lag (100-10,000 messages)

Lag: 5,000 messages

⚠️ Warning: Consumer falling behind
⚠️ May indicate:

Slow processing
Insufficient consumers
Temporary spike in traffic

High Lag (10,000+ messages)

Lag: 500,000 messages

🚨 Critical: Consumer severely behind
🚨 Indicates:

Consumer stuck or slow
Undersized consumer group
Processing bottleneck
Data loss risk (if retention expires)

What Lag Doesn'tTell You

❌ Lag doesn't tell you:

Processing latency (time from produce to consume)
Whether consumer is stuck
If consumer is rebalancing
Root cause of lag

✅ You need additional metrics for complete picture.

2. Measuring Consumer Lag

Method 1: kafka-consumer-groups.sh

kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --group my-consumer-group

Output:

GROUP           TOPIC      PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG    CONSUMER-ID
my-group        orders     0          12345           12350           5      consumer-1-abc
my-group        orders     1          67890           68000           110    consumer-2-def
my-group        orders     2          11111           11111           0      consumer-3-ghi

Interpretation:

Partition 0: Lag 5 (good)
Partition 1: Lag 110 (watch)
Partition 2: Lag 0 (caught up)

Method 2: JMX Metrics

// Consumer exposes lag via JMX
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
ObjectName name = new ObjectName(
    "kafka.consumer:type=consumer-fetch-manager-metrics," +
    "client-id=consumer-1,topic=orders,partition=0"
);

Long lag = (Long) mbs.getAttribute(name, "records-lag");
System.out.println("Current lag: " + lag);

Method 3: Programmatic

// Get lag programmatically
Map<TopicPartition, Long> endOffsets = consumer.endOffsets(consumer.assignment());

for (TopicPartition partition : consumer.assignment()) {
    long currentOffset = consumer.position(partition);
    long endOffset = endOffsets.get(partition);
    long lag = endOffset - currentOffset;
    
    System.out.println(partition + " lag: " + lag);
}

3. Key Consumer Metrics

Critical Metrics (Must Monitor)

1. records-consumed-rate

Metric: kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*,name=records-consumed-rate
Unit: records/second

What it tells you: Consumption rate.

Alert on:

Sudden drop (consumer stuck)
Zero for extended period (consumer dead)

2. records-lag-max

Metric: kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*,topic=*,name=records-lag-max
Unit: number of records

What it tells you: Maximum lag across all partitions.

Alert on:

Exceeds threshold (e.g., 10,000)
Continuously increasing

3. commit-latency-avg

Metric: kafka.consumer:type=consumer-coordinator-metrics,client-id=*,name=commit-latency-avg
Unit: milliseconds

What it tells you: How long commits take.

Alert on:

High latency (> 100ms)
Timeouts

4. fetch-latency-avg

Metric: kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*,name=fetch-latency-avg
Unit: milliseconds

What it tells you: How long fetch requests take.

Alert on:

High latency (> 500ms)
Indicates broker or network issues

5. rebalance-total

Metric: kafka.consumer:type=consumer-coordinator-metrics,client-id=*,name=rebalance-total
Unit: count (cumulative)

What it tells you: How many rebalances occurred.

Alert on:

Frequent rebalances (rate > 1/minute)
Indicates instability

Important Metrics (Should Monitor)

6. fetch-rate

Metric: kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*,name=fetch-rate

What it tells you: Fetch requests per second.

7. bytes-consumed-rate

Metric: kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*,name=bytes-consumed-rate

What it tells you: Data consumption rate (MB/s).

8. assigned-partitions

Metric: kafka.consumer:type=consumer-coordinator-metrics,client-id=*,name=assigned-partitions

What it tells you: Number of partitions assigned.

Alert on:

Zero (consumer not assigned)
Imbalanced across consumers

4. Detecting Problems

Problem 1: Consumer Stuck

Symptoms:

records-consumed-rate = 0
Lag increasing
No errors in logs

Detection:

long lastConsumeTime = System.currentTimeMillis();

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(...);
    
    if (records.count() > 0) {
        lastConsumeTime = System.currentTimeMillis();
    } else {
        long timeSinceLastConsume = System.currentTimeMillis() - lastConsumeTime;
        if (timeSinceLastConsume > 60000) { // 1 minute
            logger.error("Consumer stuck! No records for 1 minute");
            alerting.sendAlert("Consumer stuck");
        }
    }
}

Problem 2: Rebalance Loop

Symptoms:

rebalance-total increasing rapidly
records-consumed-rate near zero
Logs full of "Joining group" messages

Detection:

# Check rebalance rate
kafka-consumer-groups.sh --describe --group my-group | grep -i rebalancing

Root Causes:

max.poll.interval.ms too short
Slow processing
Network issues
GC pauses

Problem 3: Unbalanced Partition Assignment

Symptoms:

Some consumers have many partitions
Some consumers have few/none
Uneven lag across consumers

Detection:

kafka-consumer-groups.sh --describe --group my-group

Consumer 1: 10 partitions
Consumer 2: 8 partitions
Consumer 3: 2 partitions  ← Underutilized!

Fix: Check assignment strategy, ensure enough partitions.

Problem 4: Lag Not Decreasing

Symptoms:

Lag constant or increasing
Consumer actively consuming
No errors

Diagnosis:

Produce rate: 1000 msgs/sec
Consume rate: 500 msgs/sec  ← Too slow!

Lag will increase by 500 msgs/sec indefinitely

Solutions:

Add more consumers
Optimize processing
Scale infrastructure

5. Alerting Strategies

Alert Thresholds

Alerts:
  LagTooHigh:
    metric: records-lag-max
    threshold: 10000
    duration: 5 minutes
    severity: warning
    
  LagCritical:
    metric: records-lag-max
    threshold: 100000
    duration: 1 minute
    severity: critical
    
  ConsumerStuck:
    metric: records-consumed-rate
    threshold: 0
    duration: 2 minutes
    severity: critical
    
  FrequentRebalances:
    metric: rebalance-rate-per-hour
    threshold: 10
    duration: 1 hour
    severity: warning
    
  HighCommitLatency:
    metric: commit-latency-avg
    threshold: 100
    duration: 5 minutes
    severity: warning

Alert Implementation

public class ConsumerMetrics {
    private final MeterRegistry registry;
    
    public ConsumerMetrics(MeterRegistry registry) {
        this.registry = registry;
    }
    
    public void recordLag(TopicPartition partition, long lag) {
        registry.gauge("consumer.lag", 
            Tags.of("topic", partition.topic(), "partition", String.valueOf(partition.partition())),
            lag);
        
        // Alert if lag exceeds threshold
        if (lag > 10000) {
            alerting.sendAlert("High lag: " + partition + " = " + lag);
        }
    }
    
    public void recordConsumption(int recordCount) {
        registry.counter("consumer.records.consumed").increment(recordCount);
    }
    
    public void recordRebalance() {
        registry.counter("consumer.rebalances").increment();
    }
}

6. Operational Dashboard

Essential Panels

Panel 1: Consumer Lag

┌────────────────────────────────────┐
│        Consumer Lag (by topic)     │
│                                    │
│  orders-0:    100 msgs             │
│  orders-1:    250 msgs             │
│  orders-2:    0 msgs               │
│                                    │
│  Total Lag:   350 msgs             │
└────────────────────────────────────┘

Panel 2: Consumption Rate

┌────────────────────────────────────┐
│     Records Consumed/sec           │
│                                    │
│ 1000 ┌─┐                          │
│      │ │  ┌─┐                     │
│  500 │ │  │ │  ┌─┐                │
│      │ │  │ │  │ │                │
│    0 └─┴──┴─┴──┴─┴────────────    │
│       Time →                       │
└────────────────────────────────────┘

Panel 3: Rebalance Events

┌────────────────────────────────────┐
│        Rebalance Events            │
│                                    │
│  Last 24 hours: 3 rebalances       │
│  - 10:23 AM (duration: 5s)         │
│  - 02:15 PM (duration: 8s)         │
│  - 11:47 PM (duration: 4s)         │
└────────────────────────────────────┘

Panel 4: Consumer Group Health

┌────────────────────────────────────┐
│      Consumer Group Status         │
│                                    │
│  Group: my-group                   │
│  State: Stable                     │
│  Members: 5                        │
│  Partitions: 10                    │
│  Assignment: Balanced              │
└────────────────────────────────────┘

7. Troubleshooting Guide

Issue: High Lag

Steps:

Check consumption rate: Is consumer consuming?
Check produce rate: Is producer overwhelming consumer?
Compare rates: produce_rate > consume_rate means lag will grow
Check processing time: Slow processing?
Check consumer count: Need more consumers?

Fix:

Add more consumers (if consumers < partitions)
Optimize processing code
Increase max.poll.records (if processing fast)
Scale infrastructure

Issue: Frequent Rebalances

Steps:

Check logs for rebalance reasons
Check max.poll.interval.ms: Processing too slow?
Check session.timeout.ms: Heartbeats timing out?
Check GC logs: Long GC pauses?
Check network: Connectivity issues?

Fix:

Increase max.poll.interval.ms
Reduce max.poll.records
Tune JVM GC
Use static membership
Use cooperative rebalancing

Issue: Consumer Not Consuming

Steps:

Check if consumer is alive: jps or ps
Check logs for errors
Check partition assignment: kafka-consumer-groups.sh
Check for deserialization errors
Check for poison pills

Fix:

Restart consumer
Fix deserialization issues
Skip poison pill records
Check configuration

8. Monitoring Tools

Kafka Manager / CMAK

Web UI for managing Kafka clusters:

Consumer group monitoring
Lag visualization
Partition reassignment

Prometheus + Grafana

# Prometheus scrape config
scrape_configs:
  - job_name: 'kafka-consumers'
    static_configs:
      - targets: ['consumer-host:9090']

# PromQL queries
# Total lag
sum(kafka_consumer_records_lag_max)

# Consumption rate
rate(kafka_consumer_records_consumed_total[5m])

# Rebalance rate
rate(kafka_consumer_rebalance_total[1h])

Datadog / New Relic

Commercial APM tools with Kafka integrations:

Auto-discovery of consumers
Pre-built dashboards
Smart alerting
Anomaly detection

9. Logging Best Practices

What to Log

// Consumer startup
logger.info("Starting consumer: group={}, topics={}", groupId, topics);

// Rebalance events
consumer.subscribe(topics, new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        logger.info("Partitions revoked: {}", partitions);
    }
    
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        logger.info("Partitions assigned: {}", partitions);
    }
});

// Processing errors
try {
    processRecord(record);
} catch (Exception e) {
    logger.error("Processing failed: topic={}, partition={}, offset={}", 
        record.topic(), record.partition(), record.offset(), e);
}

// Lag tracking
logger.info("Current lag: {}", calculateLag());

// Shutdown
logger.info("Shutting down consumer");

10. Production Readiness Checklist

Monitoring Checklist

Consumer lag monitored per partition
Consumption rate tracked
Rebalance events logged and alerted
Commit latency monitored
Fetch latency monitored
JMX metrics exposed
Prometheus/Grafana dashboard configured
Alerts configured for critical metrics
Runbooks created for common issues
On-call team trained on troubleshooting

Dashboard Checklist

Consumer lag visualization
Consumption rate graph
Rebalance event timeline
Consumer group health status
Partition assignment view
Error rate tracking
Latency percentiles (p50, p95, p99)

Summary

You learned:

✅ Consumer lag = log end offset - consumer offset
✅ Monitor records-lag-max, records-consumed-rate, rebalance-total
✅ Detect stuck consumers, rebalance loops, unbalanced assignment
✅ Set up alerts for high lag, stuck consumers, frequent rebalances
✅ Build operational dashboards with essential panels
✅ Troubleshoot common issues systematically
✅ Use monitoring tools (Prometheus, Grafana, Kafka Manager)
✅ Log critical events (rebalances, errors, lag)

Next up: Practical Consumer Patterns & Real-World Scenarios (Module 10.1) — Bring everything together with real-world patterns.