10.1 Error Handling & Failure Scenarios

Module Overview

Goal: Master error handling and failure recovery in Kafka Streams.

What You'll Learn:

• Types of errors (transient vs permanent)
• Uncaught exception handlers
• Deserialization errors
• Production exception handlers
• Stream processing exceptions
• Graceful shutdown
• Recovery strategies

What's NOT Covered Here:

• EOS internals (covered in Module 9.1)
• State restoration (covered in Module 6.3)
• Rebalancing (covered in Module 1.1)

Key Insight: Proper error handling is crucial for production resilience — distinguish between recoverable and non-recoverable errors and handle accordingly.

1. Error Types

Transient Errors

Transient = temporary, may succeed on retry

Examples:
├─ Network timeout
├─ Broker temporarily unavailable
├─ Resource temporarily exhausted
└─ Transient database connection issue

Handling:
└─ Retry (automatic or manual)

Permanent Errors

Permanent = fundamental problem, won't succeed on retry

Examples:
├─ Malformed data (parse error)
├─ Missing required field
├─ Business rule violation
└─ Authentication failure

Handling:
├─ Log error
├─ Skip record (or send to DLT)
└─ Continue processing

Fatal Errors

Fatal = application-level issue, cannot continue

Examples:
├─ Out of memory
├─ Disk full
├─ State store corruption (unrecoverable)
└─ Configuration error

Handling:
└─ Shut down application (restart by orchestrator)

2. Exception Handlers

Overview

Kafka Streams provides hooks for exception handling:

1. DeserializationExceptionHandler
   └─ Handle deserialization errors (input)

2. ProductionExceptionHandler
   └─ Handle production errors (output)

3. StreamsUncaughtExceptionHandler
   └─ Handle all other uncaught exceptions

Configure during application setup

3. Deserialization Exception Handler

The Problem

Input record cannot be deserialized:

Scenario:
├─ Producer writes invalid JSON
├─ Schema evolution breaks deserialization
├─ Corrupt data in Kafka
└─ Kafka Streams tries to deserialize → fails

Default behavior:
└─ Application crashes (FAIL fast)

Better approach:
└─ Skip bad record, log, continue

Built-in Handlers

// 1. LOG AND FAIL (default)
props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
    LogAndFailExceptionHandler.class);

// Application crashes on deserialization error

// 2. LOG AND CONTINUE
props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
    LogAndContinueExceptionHandler.class);

// Logs error, skips record, continues processing

Custom Handler

public class CustomDeserializationExceptionHandler 
    implements DeserializationExceptionHandler {

    @Override
    public DeserializationHandlerResponse handle(
        ProcessorContext context,
        ConsumerRecord<byte[], byte[]> record,
        Exception exception) {

        // Log error with context
        logger.error("Deserialization error: topic={}, partition={}, offset={}, error={}",
            record.topic(), record.partition(), record.offset(), exception.getMessage());

        // Send to Dead Letter Topic (DLT)
        try {
            deadLetterProducer.send(new ProducerRecord<>(
                "deserialization-errors",
                record.key(),
                record.value()
            ));
        } catch (Exception e) {
            logger.error("Failed to send to DLT", e);
        }

        // Decide: CONTINUE or FAIL
        return DeserializationHandlerResponse.CONTINUE;
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // Initialize (e.g., DLT producer)
    }
}

Configuration

props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
    CustomDeserializationExceptionHandler.class);

4. Production Exception Handler

The Problem

Output record cannot be written:

Scenario:
├─ Broker temporarily unavailable
├─ Topic doesn't exist
├─ Authorization failure
└─ Serialization error (output)

Default behavior:
└─ Application crashes

Better approach:
└─ Retry transient errors, fail on permanent errors

Built-in Handlers

// 1. DEFAULT PRODUCTION EXCEPTION HANDLER
props.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
    DefaultProductionExceptionHandler.class);

// Fails on any error (safe, but strict)

// (No built-in "continue" handler for production)

Custom Handler

public class CustomProductionExceptionHandler 
    implements ProductionExceptionHandler {

    @Override
    public ProductionExceptionHandlerResponse handle(
        ProducerRecord<byte[], byte[]> record,
        Exception exception) {

        // Check error type
        if (exception instanceof RetriableException) {
            // Transient error (network, timeout)
            logger.warn("Transient production error, will retry: {}", 
                exception.getMessage());
            return ProductionExceptionHandlerResponse.CONTINUE;
        } else if (exception instanceof org.apache.kafka.common.errors.SerializationException) {
            // Permanent error (bad data)
            logger.error("Serialization error, skipping record: {}", 
                exception.getMessage());
            // Send to DLT
            sendToDeadLetterTopic(record);
            return ProductionExceptionHandlerResponse.CONTINUE;
        } else {
            // Unknown error, fail application
            logger.error("Fatal production error: {}", exception.getMessage());
            return ProductionExceptionHandlerResponse.FAIL;
        }
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // Initialize
    }
}

Configuration

props.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
    CustomProductionExceptionHandler.class);

5. Uncaught Exception Handler

The Problem

Exceptions not caught by other handlers:

Examples:
├─ NullPointerException in processing logic
├─ Business logic exception
├─ State store exception
└─ Any runtime exception

Default behavior (Kafka Streams < 3.0):
└─ Thread dies, instance may become zombie

Default behavior (Kafka Streams 3.0+):
└─ Application shuts down

Handler

public class CustomUncaughtExceptionHandler 
    implements StreamsUncaughtExceptionHandler {

    @Override
    public StreamThreadExceptionResponse handle(Throwable exception) {

        logger.error("Uncaught exception in Kafka Streams: {}", 
            exception.getMessage(), exception);

        // Classify error
        if (exception instanceof OutOfMemoryError) {
            // Fatal: Shut down and let orchestrator restart
            return StreamThreadExceptionResponse.SHUTDOWN_APPLICATION;
        } else if (exception instanceof NullPointerException) {
            // Likely a bug, but attempt to continue on other threads
            return StreamThreadExceptionResponse.REPLACE_THREAD;
        } else {
            // Unknown, shut down to be safe
            return StreamThreadExceptionResponse.SHUTDOWN_APPLICATION;
        }
    }
}

Response Types

REPLACE_THREAD:
├─ Kill current thread
├─ Start new thread
└─ Continue processing (if threads available)

SHUTDOWN_CLIENT:
├─ Shut down Kafka Streams instance
└─ Graceful shutdown

SHUTDOWN_APPLICATION:
├─ Shut down entire application
└─ Most aggressive (use for fatal errors)

Configuration

KafkaStreams streams = new KafkaStreams(topology, props);

streams.setUncaughtExceptionHandler(new CustomUncaughtExceptionHandler());

streams.start();

6. Processing Errors (in Topology)

The Problem

Errors during record processing:

Example:
stream.mapValues(value -> {
    return value.toUpperCase();  // NPE if value is null
});

If not handled:
└─ Uncaught exception → thread dies

Strategy 1: Try-Catch in Lambda

stream.mapValues(value -> {
    try {
        return value.toUpperCase();
    } catch (Exception e) {
        logger.error("Error processing: {}", e.getMessage());
        return "ERROR";  // Fallback value
    }
});

Strategy 2: Branching

// Split valid and invalid records
Map<String, KStream<String, Order>> branches = stream.split()
    .branch((k, v) -> isValid(v), Branched.as("valid"))
    .branch((k, v) -> !isValid(v), Branched.as("invalid"))
    .noDefaultBranch();

// Process valid records
branches.get("valid")
    .mapValues(this::processOrder)
    .to("output-topic");

// Send invalid records to DLT
branches.get("invalid")
    .to("dead-letter-topic");

Strategy 3: Custom Processor

stream.process(() -> new Processor<String, Order, String, ProcessedOrder>() {
    @Override
    public void process(Record<String, Order> record) {
        try {
            ProcessedOrder result = processOrder(record.value());
            context().forward(record.withValue(result));
        } catch (Exception e) {
            logger.error("Error processing order {}: {}", 
                record.key(), e.getMessage());
            // Send to DLT
            context().forward(record, "dead-letter-topic");
        }
    }
});

7. Dead Letter Topics (DLT)

What is a DLT?

Dead Letter Topic (DLT) = topic for failed records

Purpose:
├─ Store records that cannot be processed
├─ Prevent application crash
├─ Allow later analysis/reprocessing
└─ Decouple error handling from main flow

Pattern:
├─ Catch error
├─ Send to DLT
└─ Continue processing

DLT Implementation

// Separate producer for DLT
Producer<String, String> dltProducer = new KafkaProducer<>(dltProps);

// In exception handler
public void sendToDeadLetterTopic(ConsumerRecord<byte[], byte[]> record, Exception error) {
    ProducerRecord<String, String> dltRecord = new ProducerRecord<>(
        "my-app-dlt",
        new String(record.key()),
        new String(record.value())
    );

    // Add headers with error info
    dltRecord.headers().add("error.message", error.getMessage().getBytes());
    dltRecord.headers().add("error.class", error.getClass().getName().getBytes());
    dltRecord.headers().add("original.topic", record.topic().getBytes());
    dltRecord.headers().add("original.partition", 
        String.valueOf(record.partition()).getBytes());
    dltRecord.headers().add("original.offset", 
        String.valueOf(record.offset()).getBytes());

    dltProducer.send(dltRecord);
}

DLT Best Practices

1. Include error context in headers
   └─ Error message, class, stack trace

2. Include original record metadata
   └─ Topic, partition, offset

3. Monitor DLT
   └─ Alert on high DLT volume

4. Process DLT asynchronously
   └─ Separate consumer analyzes/fixes records

5. Retention policy
   └─ Longer retention (weeks/months)

8. Graceful Shutdown

Why Graceful Shutdown?

Ungraceful shutdown:
├─ Kill -9 (SIGKILL)
├─ No cleanup
├─ In-flight transactions aborted
├─ State not flushed
└─ Long recovery on restart

Graceful shutdown:
├─ Finish current processing
├─ Commit transactions
├─ Flush state stores
├─ Close resources
└─ Fast restart

Implementation

KafkaStreams streams = new KafkaStreams(topology, props);

// Graceful shutdown on SIGTERM
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    logger.info("Shutdown signal received, closing Kafka Streams");
    streams.close(Duration.ofSeconds(30));  // Wait up to 30 seconds
    logger.info("Kafka Streams closed");
}));

streams.start();

// Keep main thread alive
streams.state();  // Or use CountDownLatch

Shutdown Timeout

// Close with timeout
streams.close(Duration.ofSeconds(30));

// If timeout exceeded:
├─ Force close
├─ May lose in-flight data
└─ Longer recovery

Recommendation: 30-60 seconds for large state

State Listener

streams.setStateListener((newState, oldState) -> {
    logger.info("State transition: {} -> {}", oldState, newState);

    if (newState == State.REBALANCING) {
        logger.warn("Rebalancing started");
    } else if (newState == State.RUNNING) {
        logger.info("Application running");
    } else if (newState == State.ERROR) {
        logger.error("Application in error state");
    }
});

9. Failure Scenarios & Recovery

Scenario 1: Broker Down

Failure:
├─ Kafka broker goes down
└─ Producer/consumer cannot connect

Kafka Streams behavior:
├─ Retries automatically (retries=Integer.MAX_VALUE)
├─ Waits for broker to return
└─ Resumes processing when available

Action: None (automatic recovery)

Scenario 2: Instance Crash

Failure:
├─ Kafka Streams instance crashes (OOM, kill -9)
└─ Tasks unprocessed

Kafka Streams behavior:
├─ Heartbeat stops
├─ Group coordinator detects timeout
├─ Rebalance triggered
├─ Tasks reassigned to other instances
└─ State restored from changelog

Recovery time:
├─ Without standby: Minutes (restore from changelog)
└─ With standby: Seconds (standby promoted)

Action: Restart instance (orchestrator)

Scenario 3: State Store Corruption

Failure:
├─ RocksDB corruption
└─ Cannot read state

Kafka Streams behavior:
├─ Detects corruption
├─ Deletes corrupted state
├─ Restores from changelog
└─ Resumes processing

Recovery time: Minutes (restore time)

Action: Monitor and ensure changelog intact

Scenario 4: Rebalance Storm

Failure:
├─ Frequent rebalances
├─ Tasks constantly reassigned
└─ No progress

Causes:
├─ max.poll.interval.ms too low
├─ Processing too slow
├─ Network instability
└─ Instance flapping

Solutions:
1. Increase max.poll.interval.ms
2. Optimize processing
3. Fix network issues
4. Stabilize instances

10. Resilience Patterns

Pattern 1: Circuit Breaker

// Track failures
AtomicInteger failureCount = new AtomicInteger(0);
int FAILURE_THRESHOLD = 5;

stream.mapValues(value -> {
    if (failureCount.get() > FAILURE_THRESHOLD) {
        // Circuit open: Fast fail
        throw new CircuitBreakerOpenException();
    }

    try {
        Result result = externalService.call(value);
        failureCount.set(0);  // Reset on success
        return result;
    } catch (Exception e) {
        failureCount.incrementAndGet();
        throw e;
    }
});

Pattern 2: Fallback

stream.mapValues(value -> {
    try {
        return enrichWithExternalData(value);
    } catch (Exception e) {
        logger.warn("External enrichment failed, using fallback");
        return enrichWithCache(value);  // Fallback to cache
    }
});

Pattern 3: Retry with Backoff

public Result processWithRetry(Order order) {
    int maxRetries = 3;
    int retryDelay = 100;  // ms

    for (int i = 0; i < maxRetries; i++) {
        try {
            return externalService.process(order);
        } catch (TransientException e) {
            if (i == maxRetries - 1) throw e;  // Last retry
            Thread.sleep(retryDelay * (i + 1));  // Exponential backoff
        }
    }
    throw new RuntimeException("Max retries exceeded");
}

11. Monitoring for Failures

Key Metrics

# Failed records (deserialization)
kafka.streams:type=stream-metrics,
              client-id=*,
              name=failed-stream-threads

# Errors per second
kafka.streams:type=stream-task-metrics,
              task-id=*,
              name=error-rate

# Thread state
kafka.streams:type=stream-thread-metrics,
              thread-id=*,
              name=thread-state

# Commit failures
kafka.streams:type=stream-thread-metrics,
              thread-id=*,
              name=commit-latency-max

Logging

// Structured logging for debugging
logger.error("Processing error: topic={}, partition={}, offset={}, key={}, error={}",
    context.topic(),
    context.partition(),
    context.offset(),
    record.key(),
    exception.getMessage(),
    exception);

Alerting

Alert on:
├─ DLT volume spike (> threshold)
├─ Application state = ERROR
├─ High deserialization error rate
├─ Frequent restarts
└─ Consumer lag growing

12. Testing Error Handling

Unit Testing with TopologyTestDriver

@Test
public void testDeserializationError() {
    // Create topology with custom exception handler
    Topology topology = buildTopology();
    TopologyTestDriver testDriver = new TopologyTestDriver(topology, props);

    TestInputTopic<String, String> inputTopic = 
        testDriver.createInputTopic("input", stringSerde, stringSerde);

    // Send invalid record
    inputTopic.pipeInput("key", "invalid-json");

    // Verify: No crash, record skipped
    // (Check DLT or logs)
}

Integration Testing

@Test
public void testBrokerFailure() throws Exception {
    // Start Kafka Streams
    KafkaStreams streams = new KafkaStreams(topology, props);
    streams.start();

    // Simulate broker failure
    kafkaContainer.stop();

    // Wait (Streams should retry)
    Thread.sleep(10000);

    // Restart broker
    kafkaContainer.start();

    // Verify: Streams recovers and processes
    // (Check output topic)
}

13. Best Practices

1. Use LogAndContinue for Deserialization

// Don't crash on bad data
props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
    LogAndContinueExceptionHandler.class);

2. Implement Dead Letter Topics

// Capture failed records for analysis
// Don't lose data silently

3. Set Uncaught Exception Handler

// Always set handler (default changed in 3.0+)
streams.setUncaughtExceptionHandler(new CustomUncaughtExceptionHandler());

4. Use Graceful Shutdown

// Always close gracefully
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    streams.close(Duration.ofSeconds(30));
}));

5. Monitor and Alert

Track:
├─ Error rates
├─ DLT volume
├─ Application state
└─ Recovery time

Alert on anomalies

6. Test Failure Scenarios

Test:
├─ Deserialization errors
├─ Processing exceptions
├─ Broker failures
├─ Instance crashes
└─ Rebalances

Verify recovery behavior

14. Mental Model Summary

Exception Handler Hierarchy

┌─────────────────────────────────────────────┐
│  DeserializationExceptionHandler            │
│  └─ Input deserialization errors            │
└─────────────────────────────────────────────┘
          ↓ (not handled)
┌─────────────────────────────────────────────┐
│  Processing (your code)                     │
│  └─ Try-catch in lambda/processor           │
└─────────────────────────────────────────────┘
          ↓ (not handled)
┌─────────────────────────────────────────────┐
│  ProductionExceptionHandler                 │
│  └─ Output production errors                │
└─────────────────────────────────────────────┘
          ↓ (not handled)
┌─────────────────────────────────────────────┐
│  StreamsUncaughtExceptionHandler            │
│  └─ All other uncaught exceptions           │
└─────────────────────────────────────────────┘
          ↓ (not handled)
┌─────────────────────────────────────────────┐
│  Application crash / Thread dies            │
└─────────────────────────────────────────────┘

Error Handling Decision Tree

Error occurs:
├─ Transient (network, timeout)?
│  └─ Retry (automatic or manual)
│
├─ Permanent (bad data, parse error)?
│  ├─ Log error
│  ├─ Send to DLT
│  └─ Continue processing
│
└─ Fatal (OOM, corruption)?
   └─ Shut down, let orchestrator restart

15. Questions You Should Be Able to Answer

After this module, you should be able to answer:

✅ What are the three types of exception handlers in Kafka Streams?
✅ What's the difference between transient and permanent errors?
✅ How do I handle deserialization errors without crashing?
✅ What is a Dead Letter Topic (DLT)?
✅ How do I implement graceful shutdown?
✅ What happens when an instance crashes?
✅ How do I monitor for failures?
✅ When should I use REPLACE_THREAD vs SHUTDOWN_APPLICATION?

16. What's Next?

Now that you understand error handling, the next modules will cover:

• Module 11.1: Interactive queries & testing — querying state and testing topologies
• Module 12.1: Monitoring — operational metrics and dashboards
• Module 13.1: Advanced configuration — performance tuning

Summary

You learned:

• ✅ Three exception handlers: Deserialization, Production, Uncaught
• ✅ Transient errors → retry, Permanent errors → skip/DLT, Fatal errors → shutdown
• ✅ LogAndContinueExceptionHandler for deserialization (don't crash on bad data)
• ✅ Dead Letter Topics (DLT) capture failed records for analysis
• ✅ Graceful shutdown with streams.close(timeout) and shutdown hook
• ✅ Automatic recovery from broker/instance failures via rebalancing
• ✅ Monitor error rates, DLT volume, and application state
• ✅ Test failure scenarios (bad data, crashes, broker down) to verify resilience

Next up: Module 11.1 — Interactive Queries & Testing — Learn how to query Kafka Streams state and test your topologies effectively.