6.1 Error Handling & Failure Scenarios

Module Overview

Goal: Turn theory into operational reality. Learn to handle producer errors effectively.

What You'll Learn:

• Retriable vs non-retriable errors
• Serialization errors and how to handle them
• Authorization and metadata errors
• Timeout scenarios and recovery
• Callback-based error handling patterns
• When to fail fast vs retry forever
• Production error handling strategies
• Monitoring and alerting for failures

What's NOT Covered Here:

• How to prevent failures (covered in reliability modules)
• Performance optimization (covered in Module 8)
• Internal mechanics (covered in Module 7.1)

Key Distinction: This module is about reacting to failure, not preventing it.

1. Error Categories

Three Types of Errors

┌─────────────────────────────────────────────────┐
│            PRODUCER ERROR TYPES                 │
├─────────────────────────────────────────────────┤
│                                                 │
│  1. RETRIABLE ERRORS                            │
│     → Producer can retry automatically          │
│     → Transient failures                        │
│     → Example: Network glitch                   │
│                                                 │
│  2. NON-RETRIABLE ERRORS                        │
│     → Cannot be fixed by retrying               │
│     → Permanent failures                        │
│     → Example: Invalid topic                    │
│                                                 │
│  3. FATAL ERRORS                                │
│     → Producer cannot continue                  │
│     → Must close and recreate                   │
│     → Example: ProducerFencedException          │
│                                                 │
└─────────────────────────────────────────────────┘

2. Retriable Errors

Common Retriable Errors

How Producer Handles Retriable Errors

// Configuration
props.put("retries", Integer.MAX_VALUE);       // Retry indefinitely
props.put("retry.backoff.ms", 100);            // Wait 100ms between retries
props.put("delivery.timeout.ms", 120000);      // Give up after 2 minutes

Behavior:

Time  Action
─────┼────────────────────────────────
  0ms  send(record)
 10ms  → Broker: NOT_LEADER_FOR_PARTITION
110ms  Retry 1 (after backoff)
210ms  Retry 2
310ms  Retry 3
410ms  Success! ✅

Producer automatically retries — No application code needed.

Example: Handling Retriable Errors

producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        // Success (may have been retried internally)
        logger.info("Sent successfully: offset={}", metadata.offset());
    } else {
        // All retries exhausted or non-retriable error
        if (exception instanceof RetriableException) {
            logger.error("Failed after retries", exception);
            // Alert operations team
        } else {
            logger.error("Non-retriable error", exception);
            // Handle differently (see next section)
        }
    }
});

3. Non-Retriable Errors

Common Non-Retriable Errors

Handling Non-Retriable Errors

producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        
        if (exception instanceof RecordTooLargeException) {
            logger.error("Record too large: size={}", record.value().length);
            // Option 1: Split record
            // Option 2: Store large data elsewhere, send reference
            // Option 3: Increase max.message.bytes
            
        } else if (exception instanceof UnknownTopicOrPartitionException) {
            logger.error("Topic doesn't exist: {}", record.topic());
            // Create topic (if auto-create disabled)
            
        } else if (exception instanceof AuthorizationException) {
            logger.error("Permission denied for topic: {}", record.topic());
            // Grant ACL permissions
            
        } else if (exception instanceof SerializationException) {
            logger.error("Serialization failed for record: {}", record);
            // Fix data or serializer
            
        } else {
            logger.error("Unexpected error", exception);
            // Generic handling
        }
    }
});

4. Fatal Errors

Fatal Errors (Must Close Producer)

Handling Fatal Errors

try {
    producer.beginTransaction();
    producer.send(record1);
    producer.send(record2);
    producer.commitTransaction();
    
} catch (ProducerFencedException e) {
    // Fatal: Another producer fenced this one
    logger.error("Producer fenced — closing", e);
    producer.close();
    // Exit application or recreate producer
    System.exit(1);
    
} catch (OutOfOrderSequenceException e) {
    // Fatal: Sequence corruption
    logger.error("Sequence error — closing", e);
    producer.close();
    System.exit(1);
    
} catch (AuthorizationException e) {
    // Fatal in transactions
    logger.error("Authorization failed — closing", e);
    producer.close();
    System.exit(1);
    
} catch (KafkaException e) {
    // Retriable
    producer.abortTransaction();
    // Can retry
}

5. Serialization Errors

When Serialization Fails

// Custom serializer
public class UserSerializer implements Serializer<User> {
    @Override
    public byte[] serialize(String topic, User user) {
        if (user == null) {
            return null;
        }
        
        if (user.getEmail() == null) {
            throw new SerializationException("Email cannot be null");
        }
        
        // ... serialization logic
    }
}

When serializer throws exception:

User user = new User(123, "Alice", null); // No email

producer.send(new ProducerRecord<>("users", user), (metadata, exception) -> {
    if (exception != null) {
        // exception is SerializationException
        logger.error("Serialization failed", exception);
    }
});

Important: Serialization happens before sending to broker.

Handling Serialization Errors

Pattern 1: Validate Before Sending

public void sendUser(User user) {
    // Validate
    if (user.getEmail() == null || user.getEmail().isEmpty()) {
        logger.error("Invalid user: missing email");
        throw new IllegalArgumentException("Email required");
    }
    
    // Send
    ProducerRecord<String, User> record = 
        new ProducerRecord<>("users", String.valueOf(user.getId()), user);
    producer.send(record);
}

✅ Best practice: Validate before sending.

Pattern 2: Catch in Callback

producer.send(record, (metadata, exception) -> {
    if (exception instanceof SerializationException) {
        logger.error("Failed to serialize: {}", record, exception);
        // Log to dead letter queue
        sendToDeadLetterQueue(record);
    }
});

Pattern 3: Fallback Serializer

public class SafeSerializer implements Serializer<User> {
    private final UserSerializer primarySerializer = new UserSerializer();
    private final ByteArraySerializer fallbackSerializer = new ByteArraySerializer();
    
    @Override
    public byte[] serialize(String topic, User user) {
        try {
            return primarySerializer.serialize(topic, user);
        } catch (Exception e) {
            logger.error("Primary serialization failed, using fallback", e);
            // Fallback: Store error message as bytes
            return ("ERROR: " + e.getMessage()).getBytes();
        }
    }
}

6. Timeout Scenarios

Timeout Configuration

props.put("delivery.timeout.ms", 120000);      // 2 minutes total
props.put("request.timeout.ms", 30000);        // 30 seconds per request
props.put("max.block.ms", 60000);              // 60 seconds for metadata

Scenario 1: request.timeout.ms

What it controls: How long to wait for broker response per request.

Time     Action
───────  ──────────────────────────────────
0ms      send(record)
10ms     Request sent to broker
30010ms  request.timeout.ms expires
         → Retry (if retries enabled)
30110ms  Retry sent
30200ms  Success ✅

Handling:

props.put("request.timeout.ms", 30000);
props.put("retries", Integer.MAX_VALUE);

// Producer automatically retries — no code needed

Scenario 2: delivery.timeout.ms

What it controls: Maximum time from send() to success or failure.

Time      Action
────────  ──────────────────────────────────
0ms       send(record)
          → Try, timeout, retry
30s       → Try, timeout, retry
60s       → Try, timeout, retry
90s       → Try, timeout, retry
120000ms  delivery.timeout.ms expires
          → TimeoutException in callback

Handling:

producer.send(record, (metadata, exception) -> {
    if (exception instanceof TimeoutException) {
        logger.error("Delivery timeout after 2 minutes", exception);
        // Alert operations team
        // Log to dead letter queue
        sendToDeadLetterQueue(record);
    }
});

Scenario 3: max.block.ms

What it controls: How long send() can block waiting for metadata or buffer space.

// Scenario A: Metadata not available
producer.send(new ProducerRecord<>("new-topic", "key", "value"));
// → Blocks up to max.block.ms waiting for metadata
// → If timeout, throws TimeoutException

// Scenario B: Buffer full
// Producer sending faster than broker can consume
producer.send(record); // Blocks waiting for buffer space

Handling:

try {
    producer.send(record);
} catch (TimeoutException e) {
    logger.error("Blocked for {} ms", props.get("max.block.ms"), e);
    // Possible causes:
    // 1. Topic doesn't exist and auto-create disabled
    // 2. Broker unreachable
    // 3. Buffer full (backpressure)
}

7. Callback-Based Error Handling

Callback Execution Context

⚠️ Important: Callbacks execute in producer's I/O thread.

producer.send(record, (metadata, exception) -> {
    // ⚠️ Running in producer I/O thread!
    // Keep this fast — don't block
});

Rules:

• ✅ Fast operations (logging, metrics)
• ❌ Slow operations (HTTP calls, database writes)
• ❌ Blocking operations

Pattern 1: Fast Logging

producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        logger.info("Sent: offset={}", metadata.offset());
    } else {
        logger.error("Failed to send", exception);
    }
});

✅ Good: Fast logging.

Pattern 2: Offload to Thread Pool

private final ExecutorService executor = Executors.newFixedThreadPool(4);

producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        // Offload slow work
        executor.submit(() -> {
            sendToDeadLetterQueue(record);  // Slow operation
            alertOperationsTeam(exception); // Slow operation
        });
    }
});

✅ Good: Offload slow work.

Pattern 3: Error Aggregation

private final ConcurrentHashMap<String, AtomicInteger> errorCounts = new ConcurrentHashMap<>();

producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        String errorType = exception.getClass().getSimpleName();
        errorCounts.computeIfAbsent(errorType, k -> new AtomicInteger()).incrementAndGet();
    }
});

// Separate thread logs every minute
scheduledExecutor.scheduleAtFixedRate(() -> {
    errorCounts.forEach((errorType, count) -> {
        logger.warn("Error type: {} count: {}", errorType, count.getAndSet(0));
    });
}, 1, 1, TimeUnit.MINUTES);

✅ Good: Aggregated error reporting.

8. Fail Fast vs Retry Forever

When to Fail Fast

props.put("retries", 0); // Don't retry
props.put("delivery.timeout.ms", 5000); // Fail after 5 seconds

Use When:

• Ultra-low latency required
• Data loss acceptable (metrics, logs)
• Client-side retry logic preferred

When to Retry Forever

props.put("retries", Integer.MAX_VALUE);
props.put("delivery.timeout.ms", 120000); // 2 minutes

Use When:

• Data loss unacceptable
• Transient failures expected
• Application can tolerate delays

Hybrid Approach

props.put("retries", Integer.MAX_VALUE);
props.put("delivery.timeout.ms", 120000);

producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        if (isCriticalData(record)) {
            // Critical data: Fail entire application
            logger.error("Critical data failed — shutting down", exception);
            System.exit(1);
        } else {
            // Non-critical: Log and continue
            logger.warn("Non-critical data failed", exception);
            metrics.increment("non_critical_failures");
        }
    }
});

9. Dead Letter Queue Pattern

What is a Dead Letter Queue?

A DLQ is a separate topic for messages that failed to process.

Primary Topic: "orders"
DLQ Topic: "orders-dlq"

Producer sends to "orders"
  ├─ Success → Stays in "orders"
  └─ Failure → Send to "orders-dlq"

Implementing DLQ

public class ProducerWithDLQ {
    private final KafkaProducer<String, String> producer;
    private final String primaryTopic;
    private final String dlqTopic;
    
    public ProducerWithDLQ(KafkaProducer<String, String> producer, 
                           String primaryTopic) {
        this.producer = producer;
        this.primaryTopic = primaryTopic;
        this.dlqTopic = primaryTopic + "-dlq";
    }
    
    public void send(String key, String value) {
        ProducerRecord<String, String> record = 
            new ProducerRecord<>(primaryTopic, key, value);
        
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                // Send to DLQ
                sendToDLQ(key, value, exception);
            }
        });
    }
    
    private void sendToDLQ(String key, String value, Exception exception) {
        // Include error details in DLQ message
        String dlqValue = String.format(
            "{\"originalValue\":\"%s\",\"error\":\"%s\",\"timestamp\":%d}",
            value, exception.getMessage(), System.currentTimeMillis()
        );
        
        ProducerRecord<String, String> dlqRecord = 
            new ProducerRecord<>(dlqTopic, key, dlqValue);
        
        producer.send(dlqRecord, (metadata, dlqException) -> {
            if (dlqException != null) {
                // DLQ send failed — critical!
                logger.error("DLQ send failed", dlqException);
                // Fallback: Write to file, alert ops, etc.
            }
        });
    }
}

DLQ Best Practices

✅ Include error metadata:

{
  "originalKey": "user-123",
  "originalValue": "...",
  "originalTopic": "orders",
  "errorType": "SerializationException",
  "errorMessage": "Invalid JSON",
  "timestamp": 1674567890123,
  "attemptCount": 5
}

✅ Monitor DLQ size:

// Alert if DLQ accumulates
if (dlqMessageCount > threshold) {
    alertOps("DLQ threshold exceeded");
}

✅ Reprocess DLQ:

// Separate job to reprocess DLQ messages
consumer.subscribe(Collections.singletonList("orders-dlq"));
// Attempt to reprocess and send back to primary topic

10. Circuit Breaker Pattern

What is a Circuit Breaker?

Temporarily stop sending if broker is consistently failing.

┌─────────────────────────────────────┐
│        CIRCUIT BREAKER STATES       │
├─────────────────────────────────────┤
│                                     │
│  CLOSED → Sending normally          │
│     ↓                               │
│  Failures exceed threshold          │
│     ↓                               │
│  OPEN → Stop sending                │
│     ↓                               │
│  Wait (cooldown period)             │
│     ↓                               │
│  HALF-OPEN → Try sending again      │
│     ↓                               │
│  Success? → CLOSED                  │
│  Failure? → OPEN                    │
│                                     │
└─────────────────────────────────────┘

Implementing Circuit Breaker

public class CircuitBreakerProducer {
    private enum State { CLOSED, OPEN, HALF_OPEN }
    
    private State state = State.CLOSED;
    private int failureCount = 0;
    private long lastFailureTime = 0;
    
    private static final int FAILURE_THRESHOLD = 5;
    private static final long COOLDOWN_MS = 60000; // 1 minute
    
    public void send(ProducerRecord<String, String> record) {
        if (state == State.OPEN) {
            // Check if cooldown expired
            if (System.currentTimeMillis() - lastFailureTime > COOLDOWN_MS) {
                state = State.HALF_OPEN;
                logger.info("Circuit breaker: OPEN → HALF_OPEN");
            } else {
                logger.warn("Circuit breaker OPEN — rejecting send");
                throw new IllegalStateException("Circuit breaker open");
            }
        }
        
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                // Success
                if (state == State.HALF_OPEN) {
                    state = State.CLOSED;
                    failureCount = 0;
                    logger.info("Circuit breaker: HALF_OPEN → CLOSED");
                }
            } else {
                // Failure
                failureCount++;
                lastFailureTime = System.currentTimeMillis();
                
                if (failureCount >= FAILURE_THRESHOLD && state == State.CLOSED) {
                    state = State.OPEN;
                    logger.error("Circuit breaker: CLOSED → OPEN (failures={})", failureCount);
                }
            }
        });
    }
}

11. Monitoring and Alerting

Key Metrics to Monitor

Example: Metrics with Micrometer

import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;

public class MonitoredProducer {
    private final KafkaProducer<String, String> producer;
    private final Counter successCounter;
    private final Counter failureCounter;
    
    public MonitoredProducer(KafkaProducer<String, String> producer, 
                             MeterRegistry registry) {
        this.producer = producer;
        this.successCounter = registry.counter("kafka.producer.success");
        this.failureCounter = registry.counter("kafka.producer.failure");
    }
    
    public void send(ProducerRecord<String, String> record) {
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                successCounter.increment();
            } else {
                failureCounter.increment();
                registry.counter("kafka.producer.failure", 
                    "error_type", exception.getClass().getSimpleName()).increment();
            }
        });
    }
}

12. Production Error Handling Template

Comprehensive Error Handling

public class ProductionProducer {
    private final KafkaProducer<String, String> producer;
    private final String dlqTopic;
    private final ExecutorService executor;
    
    public void send(ProducerRecord<String, String> record) {
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                // Success
                handleSuccess(metadata);
            } else {
                // Failure
                handleFailure(record, exception);
            }
        });
    }
    
    private void handleSuccess(RecordMetadata metadata) {
        logger.debug("Sent: topic={}, partition={}, offset={}", 
            metadata.topic(), metadata.partition(), metadata.offset());
        metrics.increment("producer.success");
    }
    
    private void handleFailure(ProducerRecord<String, String> record, Exception exception) {
        logger.error("Send failed: topic={}, key={}", 
            record.topic(), record.key(), exception);
        metrics.increment("producer.failure", "error_type", exception.getClass().getSimpleName());
        
        if (exception instanceof SerializationException) {
            // Data problem — send to DLQ immediately
            sendToDLQ(record, exception);
            
        } else if (exception instanceof TimeoutException) {
            // Timeout after retries — broker issue
            alertOps("Producer timeout", exception);
            sendToDLQ(record, exception);
            
        } else if (exception instanceof RecordTooLargeException) {
            // Record too large — log and DLQ
            logger.error("Record too large: {}", record.value().length());
            sendToDLQ(record, exception);
            
        } else if (exception instanceof RetriableException) {
            // All retries exhausted — likely broker down
            alertOps("Retriable error after max retries", exception);
            sendToDLQ(record, exception);
            
        } else {
            // Unknown error
            logger.error("Unknown error", exception);
            sendToDLQ(record, exception);
        }
    }
    
    private void sendToDLQ(ProducerRecord<String, String> record, Exception exception) {
        executor.submit(() -> {
            try {
                String dlqValue = createDLQPayload(record, exception);
                ProducerRecord<String, String> dlqRecord = 
                    new ProducerRecord<>(dlqTopic, record.key(), dlqValue);
                producer.send(dlqRecord);
            } catch (Exception e) {
                logger.error("DLQ send failed — CRITICAL", e);
                // Last resort: Write to disk
                writeToFile(record, exception);
            }
        });
    }
}

Summary

Key Takeaways

✅ Three error categories: Retriable, non-retriable, fatal
✅ Retriable errors: Producer auto-retries (configure with retries, delivery.timeout.ms)
✅ Non-retriable errors: Fix underlying issue (data, config, permissions)
✅ Fatal errors: Close producer and recreate
✅ Serialization errors: Validate before sending
✅ Timeouts: Configure request.timeout.ms, delivery.timeout.ms, max.block.ms
✅ Callbacks run in I/O thread: Keep them fast
✅ DLQ pattern: Send failed messages to separate topic
✅ Circuit breaker: Stop sending if broker consistently failing
✅ Monitor and alert: Track failures, timeouts, DLQ size

Error Handling Checklist

• Configure retries appropriately
• Handle retriable vs non-retriable errors differently
• Validate data before serialization
• Implement Dead Letter Queue
• Offload slow work from callbacks
• Monitor failure rates
• Alert on anomalies
• Test failure scenarios (network partition, broker down)
• Document error handling strategy

What's Next

• Module 7.1: Producer Internals
  Understand what happens inside the producer

• Module 8: Performance & Tuning
  Optimize producer for throughput and latency

Previous Module: ← 5.3 - Exactly-Once Semantics (Producer Side)
Next Module: 7.1 - Producer Internals →