Debug Kafka Streams Failures with AI Log Analysis in 20 Minutes

// Add to your Kafka Streams application
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

StreamsBuilder builder = new StreamsBuilder();
KStream<String, Event> events = builder.stream("input-topic");

events
    .peek((key, value) -> 
        log.debug("Processing key={} value={}", key, value))
    .mapValues(value -> {
        try {
            return transform(value);
        } catch (Exception e) {
            // Structured error with context
            log.error("Transform failed key={} error={} stacktrace={}", 
                key, e.getMessage(), e.getStackTrace());
            return null; // Dead letter handling
        }
    })
    .filter((k, v) -> v != null);

# Aggregate logs from all Kafka Streams instances
kubectl logs -l app=streams-processor --tail=10000 > streams-all.log

# Or from multiple VMs
for host in streams-{1..3}; do
    ssh $host "tail -10000 /var/log/kafka-streams.log" >> streams-all.log
done

# Include consumer group metadata
kafka-consumer-groups --bootstrap-server localhost:9092 \
    --group my-streams-app --describe >> streams-all.log

#!/usr/bin/env python3
# analyze_streams_logs.py
import ollama
import re
from collections import Counter

def extract_errors(log_file):
    """Parse structured logs into analyzable format"""
    errors = []
    with open(log_file) as f:
        for line in f:
            if 'ERROR' in line or 'WARN' in line:
                # Extract key=value pairs
                context = {
                    match.group(1): match.group(2) 
                    for match in re.finditer(r'(\w+)=([^\s]+)', line)
                }
                context['raw'] = line
                errors.append(context)
    return errors

def analyze_with_ai(errors):
    """Send error patterns to LLM for root cause analysis"""
    # Summarize error distribution
    error_types = Counter(e.get('error', 'unknown') for e in errors)
    sample_errors = errors[:50]  # First 50 for context
    
    prompt = f"""
You are debugging a Kafka Streams application. Analyze these errors:

ERROR DISTRIBUTION:
{error_types}

SAMPLE ERRORS (with context):
{chr(10).join(e['raw'] for e in sample_errors)}

TASK:
1. Identify the root cause (deserialization, state store, rebalancing, network)
2. Find the first failing component (which topic/transformation)
3. Suggest specific fix with code changes
4. Rate confidence: high/medium/low

FORMAT: JSON
"""
    
    response = ollama.chat(
        model='llama3.1:70b',
        messages=[{'role': 'user', 'content': prompt}],
        format='json'  # Structured output
    )
    
    return response['message']['content']

# Run analysis
errors = extract_errors('streams-all.log')
print(f"Found {len(errors)} errors across {len(set(e.get('partition') for e in errors))} partitions")

diagnosis = analyze_with_ai(errors)
print("\nAI DIAGNOSIS:")
print(diagnosis)

pip install ollama --break-system-packages
ollama pull llama3.1:70b  # 40GB download, needs 64GB RAM

{
  "root_cause": "Deserialization failure in windowed aggregation",
  "first_failure": "partition=2 offset=45234 on orders-topic",
  "explanation": "Avro schema mismatch - producer sent schema v2, consumer expects v1",
  "fix": "Update SerDe to handle schema evolution with SchemaRegistry",
  "confidence": "high"
}

// Before: No schema evolution handling
JsonSerde<Event> serde = new JsonSerde<>(Event.class);

// After: Handle schema changes gracefully
SpecificAvroSerde<Event> serde = new SpecificAvroSerde<>();
Map<String, String> config = Map.of(
    "schema.registry.url", "http://schema-registry:8081",
    "specific.avro.reader", "true"  // Use reader schema
);
serde.configure(config, false);

# Replay from offset where failure started
kafka-streams-application-reset --application-id my-streams-app \
    --input-topics orders-topic \
    --to-offset 45230  # Before failure point

# Run with verbose logging
export KAFKA_OPTS="-Dlog4j.logger.org.apache.kafka.streams=DEBUG"
java -jar streams-app.jar

# Send test event that previously failed
echo '{"schema":2,"orderId":"123"}' | \
    kafka-console-producer --topic orders-topic \
    --bootstrap-server localhost:9092

# Check consumer lag is decreasing
kafka-consumer-groups --bootstrap-server localhost:9092 \
    --group my-streams-app --describe

# streams_monitor.py - Run every 5 minutes
import ollama
import subprocess
import json

def check_streams_health():
    # Get recent errors
    logs = subprocess.check_output([
        'kubectl', 'logs', '-l', 'app=streams-processor', 
        '--tail=1000', '--since=5m'
    ]).decode()
    
    if 'ERROR' not in logs and 'WARN' not in logs:
        return {"status": "healthy"}
    
    # AI quick analysis
    response = ollama.chat(
        model='llama3.1:8b',  # Faster model for monitoring
        messages=[{
            'role': 'user', 
            'content': f'Kafka Streams logs from last 5min:\n{logs}\n\nIs this critical? Reply: yes/no + 1 sentence'
        }]
    )
    
    alert = response['message']['content']
    if alert.lower().startswith('yes'):
        # Send to PagerDuty/Slack
        print(f"ALERT: {alert}")
    
    return {"status": "degraded", "alert": alert}

if __name__ == '__main__':
    health = check_streams_health()
    print(json.dumps(health, indent=2))

apiVersion: batch/v1
kind: CronJob
metadata:
  name: streams-ai-monitor
spec:
  schedule: "*/5 * * * *"
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: monitor
            image: python:3.11-slim
            command: ["python", "/scripts/streams_monitor.py"]
            volumeMounts:
            - name: scripts
              mountPath: /scripts
          restartPolicy: OnFailure

{
  "root_cause": "Checkout service deployed with modified Order schema",
  "impact": "Partition 8 stuck, ~20% of orders not processing",
  "fix": "Rollback checkout service OR update streams app schema",
  "confidence": "high",
  "evidence": "partition=8 has 100% of errors, other partitions healthy"
}