KafkaIQ

8.1 Error Handling & Resilience

Module Overview

Goal: Turn consumers into production-safe components through comprehensive error handling.

What You'll Learn:

  • Deserialization errors and handling
  • Poison pill records
  • Retry strategies
  • Dead Letter Topics (DLTs)
  • Seek & replay patterns
  • Partial failure handling
  • Circuit breakers for consumers

What's NOT Covered Here:

  • Delivery semantics (covered in Module 5.3)
  • Performance tuning (covered in Module 7.1)

Key Distinction: This is about recovering safely from errors, not guaranteeing semantics.

1. Types of Consumer Errors

Error Categories

┌────────────────────────────────────────────────┐
│           CONSUMER ERROR TYPES                 │
├────────────────────────────────────────────────┤
│                                                │
│  1. Deserialization Errors                     │
│     └─ Bad data format, schema mismatch        │
│                                                │
│  2. Processing Errors                          │
│     ├─ Transient (DB timeout, network)        │
│     └─ Permanent (invalid data, logic error)  │
│                                                │
│  3. Commit Errors                              │
│     └─ Rebalance, network issues              │
│                                                │
│  4. Rebalance Errors                           │
│     └─ Timeout, coordinator issues            │
│                                                │
└────────────────────────────────────────────────┘

2. Deserialization Errors

The Problem

// Consumer configuration
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

// Producer sends binary data (not string!)
producer.send(new ProducerRecord<>("orders", binaryData));

// Consumer tries to deserialize
ConsumerRecords<String, String> records = consumer.poll(...);
// ← SerializationException thrown!

Result: Consumer crashes, can't proceed past bad record.

Solution 1: Error-Handling Deserializer

public class ErrorHandlingDeserializer<T> implements Deserializer<T> {
    private Deserializer<T> delegate;
    
    @Override
    public T deserialize(String topic, byte[] data) {
        try {
            return delegate.deserialize(topic, data);
        } catch (Exception e) {
            logger.error("Deserialization failed for topic {}", topic, e);
            // Return null or sentinel value
            return null;
        }
    }
}

// Usage
props.put("value.deserializer", ErrorHandlingDeserializer.class.getName());
props.put("value.deserializer.delegate.class", StringDeserializer.class.getName());

Solution 2: Check for Null

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

for (ConsumerRecord<String, String> record : records) {
    if (record.value() == null) {
        // Deserialization failed
        logger.error("Null value at offset {}", record.offset());
        sendToDeadLetterTopic(record);
        continue; // Skip this record
    }
    
    processRecord(record);
}

Solution 3: Dedicated Error Handler

while (true) {
    try {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        
        for (ConsumerRecord<String, String> record : records) {
            processRecord(record);
        }
        
        consumer.commitSync();
    } catch (SerializationException e) {
        // Deserialization failed
        logger.error("Deserialization error", e);
        
        // Seek past bad record
        consumer.seek(
            new TopicPartition(e.getTopic(), e.getPartition()),
            e.getOffset() + 1
        );
    }
}

3. Poison Pill Records

What is a Poison Pill?

A record that consistently causes processing to fail:

  • Invalid data format
  • Triggers application bug
  • Violates business rules
  • Causes infinite loop

Problem: Consumer gets stuck on same record forever.

Detection Pattern

Map<TopicPartition, Long> lastProcessedOffset = new HashMap<>();
Map<TopicPartition, Integer> retryCount = new HashMap<>();
int maxRetries = 3;

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        TopicPartition tp = new TopicPartition(record.topic(), record.partition());
        
        // Check if stuck on same offset
        if (lastProcessedOffset.get(tp) != null && 
            lastProcessedOffset.get(tp) == record.offset()) {
            retryCount.put(tp, retryCount.getOrDefault(tp, 0) + 1);
            
            if (retryCount.get(tp) >= maxRetries) {
                // Poison pill detected!
                logger.error("Poison pill at offset {}", record.offset());
                sendToDeadLetterTopic(record);
                
                // Skip past it
                consumer.seek(tp, record.offset() + 1);
                retryCount.put(tp, 0);
                continue;
            }
        } else {
            retryCount.put(tp, 0);
        }
        
        try {
            processRecord(record);
            lastProcessedOffset.put(tp, record.offset());
        } catch (Exception e) {
            logger.error("Processing failed", e);
            // Will retry next poll
        }
    }
}

4. Retry Strategies

Strategy 1: Immediate Retry

int maxRetries = 3;

for (ConsumerRecord<String, String> record : records) {
    boolean success = false;
    int attempts = 0;
    
    while (!success && attempts < maxRetries) {
        try {
            processRecord(record);
            success = true;
        } catch (TransientException e) {
            attempts++;
            logger.warn("Retry {} for offset {}", attempts, record.offset());
            Thread.sleep(1000); // Wait before retry
        }
    }
    
    if (!success) {
        sendToDeadLetterTopic(record);
    }
}

Pros: Simple, immediate feedback
Cons: Blocks poll loop, can cause timeouts

Strategy 2: Retry Queue

Queue<ConsumerRecord<String, String>> retryQueue = new LinkedBlockingQueue<>();

while (true) {
    // Process retry queue first
    while (!retryQueue.isEmpty()) {
        ConsumerRecord<String, String> record = retryQueue.poll();
        try {
            processRecord(record);
        } catch (Exception e) {
            if (shouldRetry(record)) {
                retryQueue.offer(record); // Re-add to queue
            } else {
                sendToDeadLetterTopic(record);
            }
        }
    }
    
    // Then process new records
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        try {
            processRecord(record);
        } catch (Exception e) {
            retryQueue.offer(record); // Add to retry queue
        }
    }
    
    consumer.commitSync();
}

Strategy 3: Retry Topic

// Main topic consumer
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        try {
            processRecord(record);
        } catch (TransientException e) {
            // Send to retry topic with delay
            sendToRetryTopic(record, delaySeconds: 60);
        } catch (PermanentException e) {
            sendToDeadLetterTopic(record);
        }
    }
    
    consumer.commitSync();
}

// Separate retry topic consumer
// (processes after delay)

5. Dead Letter Topics (DLT)

What is a DLT?

A topic for records that can't be processed after all retry attempts.

orders (main topic)
   ↓ (processing fails)
orders.retry (retry topic, optional)
   ↓ (still fails)
orders.dead-letter (DLT)

DLT Implementation

public class DeadLetterProducer {
    private final KafkaProducer<String, String> dlqProducer;
    private final String dlqTopic;
    
    public DeadLetterProducer(String sourceTopic) {
        this.dlqTopic = sourceTopic + ".dead-letter";
        // Configure producer
        this.dlqProducer = new KafkaProducer<>(props);
    }
    
    public void sendToDeadLetter(ConsumerRecord<String, String> record, Exception error) {
        ProducerRecord<String, String> dlqRecord = 
            new ProducerRecord<>(dlqTopic, record.key(), record.value());
        
        // Add headers with error info
        dlqRecord.headers().add("original-topic", record.topic().getBytes());
        dlqRecord.headers().add("original-partition", 
            String.valueOf(record.partition()).getBytes());
        dlqRecord.headers().add("original-offset", 
            String.valueOf(record.offset()).getBytes());
        dlqRecord.headers().add("error-message", error.getMessage().getBytes());
        dlqRecord.headers().add("error-timestamp", 
            String.valueOf(System.currentTimeMillis()).getBytes());
        
        dlqProducer.send(dlqRecord);
        logger.error("Sent to DLT: topic={}, partition={}, offset={}", 
            record.topic(), record.partition(), record.offset());
    }
}

DLT Consumer Pattern

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        try {
            processRecord(record);
        } catch (Exception e) {
            logger.error("Processing failed for offset {}", record.offset(), e);
            deadLetterProducer.sendToDeadLetter(record, e);
        }
    }
    
    consumer.commitSync();
}

6. Seek & Replay Patterns

Seek to Beginning

// Replay entire partition
TopicPartition partition = new TopicPartition("orders", 0);
consumer.assign(Collections.singletonList(partition));
consumer.seekToBeginning(Collections.singletonList(partition));

// Now poll() starts from offset 0

Seek to Specific Offset

// Skip problematic message
TopicPartition partition = new TopicPartition("orders", 0);
consumer.seek(partition, 12345); // Jump to offset 12345

Seek by Timestamp

// Replay from specific time
long timestamp = System.currentTimeMillis() - (24 * 60 * 60 * 1000); // 24 hours ago

Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();
for (TopicPartition partition : consumer.assignment()) {
    timestampsToSearch.put(partition, timestamp);
}

Map<TopicPartition, OffsetAndTimestamp> offsets = 
    consumer.offsetsForTimes(timestampsToSearch);

for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : offsets.entrySet()) {
    if (entry.getValue() != null) {
        consumer.seek(entry.getKey(), entry.getValue().offset());
    }
}

7. Partial Failure Handling

Problem: Batch Processing Failure

// ❌ BAD: All-or-nothing
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

for (ConsumerRecord<String, String> record : records) {
    processRecord(record); // If one fails, all fail
}

consumer.commitSync(); // Never reached if error thrown

Solution: Per-Record Error Handling

// ✅ GOOD: Handle each record independently
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();

for (TopicPartition partition : records.partitions()) {
    List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
    
    for (ConsumerRecord<String, String> record : partitionRecords) {
        try {
            processRecord(record);
            
            // Track successful offset
            offsetsToCommit.put(
                partition,
                new OffsetAndMetadata(record.offset() + 1)
            );
        } catch (Exception e) {
            logger.error("Failed at offset {}", record.offset(), e);
            sendToDeadLetterTopic(record);
            
            // Still commit up to this point
            offsetsToCommit.put(
                partition,
                new OffsetAndMetadata(record.offset() + 1)
            );
        }
    }
}

consumer.commitSync(offsetsToCommit);

8. Circuit Breaker Pattern

Implementation

public class CircuitBreaker {
    private int failureCount = 0;
    private int failureThreshold = 5;
    private long resetTimeout = 60000; // 1 minute
    private long lastFailureTime = 0;
    private State state = State.CLOSED;
    
    enum State { CLOSED, OPEN, HALF_OPEN }
    
    public boolean allowRequest() {
        if (state == State.OPEN) {
            if (System.currentTimeMillis() - lastFailureTime > resetTimeout) {
                state = State.HALF_OPEN;
                return true;
            }
            return false; // Circuit open, reject request
        }
        return true;
    }
    
    public void recordSuccess() {
        failureCount = 0;
        state = State.CLOSED;
    }
    
    public void recordFailure() {
        failureCount++;
        lastFailureTime = System.currentTimeMillis();
        
        if (failureCount >= failureThreshold) {
            state = State.OPEN;
            logger.error("Circuit breaker opened after {} failures", failureCount);
        }
    }
}

// Usage
CircuitBreaker circuitBreaker = new CircuitBreaker();

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        if (!circuitBreaker.allowRequest()) {
            logger.warn("Circuit breaker open, skipping processing");
            sendToRetryTopic(record);
            continue;
        }
        
        try {
            processRecord(record);
            circuitBreaker.recordSuccess();
        } catch (Exception e) {
            circuitBreaker.recordFailure();
            sendToDeadLetterTopic(record);
        }
    }
    
    consumer.commitSync();
}

9. Graceful Degradation

Fallback Pattern

for (ConsumerRecord<String, String> record : records) {
    try {
        // Primary processing
        processPrimary(record);
    } catch (DatabaseException e) {
        try {
            // Fallback: cache
            processWithCache(record);
        } catch (CacheException e2) {
            // Last resort: queue for later
            sendToRetryQueue(record);
        }
    }
}

10. Error Handling Checklist

Production-Ready Error Handling

while (running) {
    try {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        
        for (ConsumerRecord<String, String> record : records) {
            try {
                // 1. Validate
                if (!isValid(record)) {
                    logger.warn("Invalid record at offset {}", record.offset());
                    sendToDeadLetterTopic(record);
                    continue;
                }
                
                // 2. Process with retry
                processWithRetry(record, maxRetries: 3);
                
            } catch (PoisonPillException e) {
                // 3. Handle poison pills
                logger.error("Poison pill detected", e);
                sendToDeadLetterTopic(record);
                
            } catch (TransientException e) {
                // 4. Handle transient errors
                logger.warn("Transient error, will retry", e);
                sendToRetryTopic(record);
                
            } catch (Exception e) {
                // 5. Catch-all
                logger.error("Unexpected error", e);
                sendToDeadLetterTopic(record);
            }
        }
        
        // 6. Commit after batch
        consumer.commitSync();
        
    } catch (WakeupException e) {
        // 7. Graceful shutdown
        break;
        
    } catch (Exception e) {
        // 8. Fatal error
        logger.error("Fatal error in consumer loop", e);
        break;
    }
}

10. Framework Integration: Spring Kafka

Overview

If you're using Spring Kafka (or similar frameworks), many error-handling patterns are built-in. This section shows framework-specific features that simplify error handling.

Note: This is optional — the patterns above work with any Kafka client. Frameworks provide convenience but aren't required.

Spring Kafka ErrorHandlingDeserializer

Spring Kafka (2.3+) provides a built-in error-handling deserializer that wraps any deserializer and gracefully handles failures.

Configuration

# application.properties
spring.kafka.consumer.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=my-group

# Use ErrorHandlingDeserializer as wrapper
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.ErrorHandlingDeserializer

# Specify the actual deserializer as delegate
spring.kafka.consumer.properties.spring.deserializer.value.delegate.class=org.apache.kafka.common.serialization.StringDeserializer

# Optional: Custom error handler
spring.kafka.consumer.properties.spring.deserializer.value.deserializer.exception.handler=myDeserializationErrorHandler

Java Configuration

@Configuration
public class KafkaConsumerConfig {
    
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        
        // Use ErrorHandlingDeserializer
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
            ErrorHandlingDeserializer.class);
        
        // Delegate to actual deserializer
        props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, 
            JsonDeserializer.class);
        
        // Optional: trusted packages for JSON deserialization
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.mycompany.models");
        
        return new DefaultKafkaConsumerFactory<>(props);
    }
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> 
        kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory) {
        
        ConcurrentKafkaListenerContainerFactory<String, String> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        return factory;
    }
}

How It Works

Message arrives
      ↓
ErrorHandlingDeserializer (wrapper)
      ↓
Try to deserialize using delegate
      ↓
   Success?
   ├─ YES → Return deserialized object
   └─ NO  → Return null (or DeserializationException)
            Consumer can check for null and handle

In your consumer:

@KafkaListener(topics = "orders")
public void listen(ConsumerRecord<String, Order> record) {
    if (record.value() == null) {
        // Deserialization failed!
        logger.error("Deserialization error at offset {}", record.offset());
        
        // Access exception details from headers
        DeserializationException ex = 
            (DeserializationException) record.headers()
                .lastHeader(ErrorHandlingDeserializer.VALUE_DESERIALIZER_EXCEPTION_HEADER)
                .value();
        
        // Send to DLT or handle
        sendToDeadLetterTopic(record);
        return;
    }
    
    // Process normally
    processOrder(record.value());
}

Spring Kafka DefaultErrorHandler

Spring Kafka provides a DefaultErrorHandler for handling processing exceptions (not just deserialization).

Configuration

@Configuration
public class KafkaErrorHandlerConfig {
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> 
        kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory) {
        
        ConcurrentKafkaListenerContainerFactory<String, String> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        
        // Configure error handler with retry
        DefaultErrorHandler errorHandler = new DefaultErrorHandler(
            new DeadLetterPublishingRecoverer(kafkaTemplate(),
                (record, ex) -> {
                    // Custom DLT topic naming
                    return new TopicPartition(record.topic() + ".DLT", -1);
                }),
            new FixedBackOff(1000L, 3) // 3 retries with 1 second between
        );
        
        // Don't retry certain exceptions
        errorHandler.addNotRetryableExceptions(
            NullPointerException.class,
            IllegalArgumentException.class
        );
        
        factory.setCommonErrorHandler(errorHandler);
        return factory;
    }
    
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate(
        ProducerFactory<String, String> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }
}

Usage in Consumer

@Service
public class OrderConsumer {
    
    @KafkaListener(topics = "orders", groupId = "order-service")
    public void processOrder(ConsumerRecord<String, Order> record) {
        // If this throws an exception, DefaultErrorHandler handles it:
        // 1. Retries 3 times with 1 second backoff
        // 2. If still fails, sends to DLT
        // 3. Commits offset
        
        if (record.value().getAmount() < 0) {
            throw new IllegalArgumentException("Negative amount"); // Not retryable
        }
        
        // May throw transient exception (will retry)
        orderService.save(record.value());
    }
}

What Spring Kafka does for you:

  • ✅ Automatic retries with configurable backoff
  • ✅ Automatic DLT publishing on failure
  • ✅ Exception classification (retryable vs non-retryable)
  • ✅ Offset management (commits after success or DLT)
  • ✅ Headers added to DLT messages (original topic, offset, exception)

Spring Kafka Retry Topics

For more advanced retry scenarios, Spring Kafka supports retry topics with progressive delays.

Configuration

@Configuration
@EnableKafka
@EnableKafkaRetryTopic
public class KafkaRetryConfig {
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> 
        kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory) {
        
        ConcurrentKafkaListenerContainerFactory<String, String> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        return factory;
    }
}

@Service
public class OrderConsumer {
    
    @RetryableTopic(
        attempts = "4", // 1 original + 3 retries
        backoff = @Backoff(delay = 1000, multiplier = 2.0), // 1s, 2s, 4s
        autoCreateTopics = "true",
        include = {TransientException.class},
        exclude = {PermanentException.class}
    )
    @KafkaListener(topics = "orders")
    public void processOrder(Order order) {
        // Process order
        // If TransientException thrown, retries on:
        // - orders-retry-1000 (after 1 second)
        // - orders-retry-2000 (after 2 seconds)
        // - orders-retry-4000 (after 4 seconds)
        // If still fails, sent to orders-dlt
        
        orderService.save(order);
    }
    
    @DltHandler
    public void handleDlt(Order order, 
                          @Header(KafkaHeaders.ORIGINAL_TOPIC) String originalTopic,
                          @Header(KafkaHeaders.EXCEPTION_MESSAGE) String exceptionMessage) {
        // Handle DLT records
        logger.error("Order sent to DLT from {}: {}", originalTopic, exceptionMessage);
        alertService.notifyFailure(order, exceptionMessage);
    }
}

What it creates:

orders (main topic)
   ↓ (failure)
orders-retry-1000 (retry after 1 second)
   ↓ (failure)
orders-retry-2000 (retry after 2 seconds)
   ↓ (failure)
orders-retry-4000 (retry after 4 seconds)
   ↓ (failure)
orders-dlt (dead letter topic)

Spring Kafka vs Vanilla Kafka

Feature Vanilla Kafka Spring Kafka
Error-handling deserializer Manual implementation Built-in ErrorHandlingDeserializer
Retry logic Manual retry loop Built-in DefaultErrorHandler
Retry topics Manual topic creation + routing @RetryableTopic annotation
DLT publishing Manual producer + headers Automatic DeadLetterPublishingRecoverer
Exception classification Manual try-catch addNotRetryableExceptions()
Backoff strategies Manual Thread.sleep() FixedBackOff, ExponentialBackOff

When to use Spring Kafka:

  • ✅ Already using Spring Boot/Framework
  • ✅ Want declarative configuration (@KafkaListener)
  • ✅ Need retry topics with minimal code
  • ✅ Prefer convention over configuration

When to use Vanilla Kafka:

  • ✅ Not using Spring
  • ✅ Need full control over consumer loop
  • ✅ Want minimal dependencies
  • ✅ Building custom framework

Other Frameworks

Kafka Streams

Kafka Streams has built-in error handling:

StreamsConfig config = new StreamsConfig(props);
config.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, 
    LogAndContinueExceptionHandler.class);

KafkaStreams streams = new KafkaStreams(topology, config);

MicroProfile Reactive Messaging

@Incoming("orders")
@Outgoing("processed-orders")
@Retry(maxRetries = 3, delay = 1000)
@Fallback(fallbackMethod = "fallbackProcessOrder")
public Order processOrder(Order order) {
    // Process with automatic retry
    return orderService.process(order);
}

public Order fallbackProcessOrder(Order order) {
    // Fallback if all retries fail
    return orderService.processBestEffort(order);
}

Framework Decision Guide

Are you using Spring Boot?
├─ YES → Use Spring Kafka
│        ├─ @KafkaListener for consumers
│        ├─ ErrorHandlingDeserializer for deserialization errors
│        ├─ DefaultErrorHandler for processing errors
│        └─ @RetryableTopic for retry topics
│
└─ NO → Are you building event streaming?
        ├─ YES → Use Kafka Streams
        │        └─ Built-in error handlers
        │
        └─ NO → Use vanilla Kafka client
                └─ Implement patterns from sections 1-9

11. Best Practices Summary

  1. Always handle deserialization errors

    • Use error-handling deserializer (custom or framework)
    • Check for null values in consumer loop
    • Send bad records to DLT

  2. Classify exceptions

    • Transient: Retry (DB timeout, network issue)
    • Permanent: Don't retry (invalid data, logic error)

  3. Use Dead Letter Topics

    • For records that fail after all retries
    • Include error metadata in headers
    • Monitor and alert on DLT activity

  4. Implement idempotent processing

    • Retry logic means duplicates are possible
    • Use deduplication or idempotent operations

  5. Monitor error rates

    • Track deserialization errors
    • Track processing failures
    • Alert on DLT throughput spikes

  6. Test failure scenarios

    • Inject bad data in tests
    • Test retry exhaustion
    • Verify DLT routing

Summary

You learned:

  • ✅ Handle deserialization errors with error-handling deserializer
  • ✅ Detect and handle poison pill records
  • ✅ Implement retry strategies (immediate, queue, topic)
  • ✅ Use Dead Letter Topics for unprocessable records
  • ✅ Seek to replay or skip problematic offsets
  • ✅ Handle partial failures in batches
  • ✅ Use circuit breakers to prevent cascading failures
  • ✅ Implement graceful degradation with fallbacks

Next up: Monitoring, Lag & Observability (Module 9.1) — Learn how to monitor consumers in production.