Handling Errors and Exceptions

1. Understanding Producer Exceptions

Exception Type	Description	Common Causes	Handling Strategy
SerializationException	Occurs when serialization fails for records	Invalid data format, unsupported data types	Validate input before sending; use proper serializers
RecordTooLargeException	Thrown when a record exceeds the configured max.message.bytes	Large payload size, misconfigured limits	Reduce message size or increase configuration limits
NotEnoughReplicasException	Producer fails to write due to insufficient replica availability	Broker unavailability, replication lag	Retry with exponential backoff; monitor cluster health
TimeoutException	Producer times out waiting for a response from the broker	Network issues, overloaded brokers	Adjust timeout settings; implement retry logic with jitter
RetriableException	Base class for exceptions that can be retried	Network or transient broker issues	Implement retry mechanism with exponential backoff
IllegalStateException	Producer is in an invalid state for the operation	Using a closed producer or incorrect usage sequence	Ensure proper lifecycle management of the producer instance

Example: Handling SerializationException

try {
    producer.send(new ProducerRecord<String, String>("topic", "key", "value"));
} catch (SerializationException e) {
    System.err.println("Failed to serialize message: " + e.getMessage());
    // Log error and potentially send to dead-letter queue
}

Note: Kafka producers are designed to be resilient and handle many transient errors automatically through built-in retry mechanisms.

2. Handling Timeout Exceptions

Exception Type	Description	Handling Strategy	Configuration
TimeoutException	Thrown when a Kafka operation exceeds its configured timeout.	Retry with exponential backoff or failover to alternative system.	`request.timeout.ms`
RebalanceInProgressException	Occurs when a consumer is rebalancing and cannot process messages.	Wait for rebalance completion or increase `max.poll.interval.ms`.	`max.poll.interval.ms`

Example: Configuring Timeout Settings

{
  "request.timeout.ms": 30000,
  "max.poll.interval.ms": 600000,
  "session.timeout.ms": 45000
}

Note: Adjust request.timeout.ms based on network latency and broker load.

Warning: Setting max.poll.interval.ms too high can cause consumer group instability.

3. Dealing with Serialization Errors

Error Type	Description	Handling Strategy	Recovery Options
SerializationException	Occurs when Kafka cannot serialize or deserialize messages due to incompatible schema or data format.	Validate message structure before sending; ensure proper serializer configuration.	Retry with corrected data format or use schema evolution techniques.
InvalidRecordException	Thrown when a record fails validation during processing (e.g., invalid key or value).	Implement strict input validation and logging.	Discard invalid records or send to dead-letter queue (DLQ).

Example: Handling Serialization Errors in Producer

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

KafkaProducer<String, MyObject> producer = new KafkaProducer<>(props);

try {
    producer.send(new ProducerRecord<>("my-topic", "key", myObject));
} catch (SerializationException e) {
    System.err.println("Serialization failed: " + e.getMessage());
    // Log error, send to DLQ or retry with fallback logic
}

Note: Ensure that your serializers are compatible with the data being sent to Kafka. Using incorrect or mismatched serializers can lead to runtime exceptions.

Best Practices for Serialization Error Handling

Use schema registries (e.g., Confluent Schema Registry) to manage and validate schemas.
Implement retry logic with exponential backoff for transient serialization issues.
Set up dead-letter queues (DLQs) to capture and inspect failed messages.
Log serialization errors with context (topic, key, value) for debugging.

4. Handling Authorization Failures

Property	Description	Default Value	Notes
kafka.security.authorizer	Class used to handle authorization checks	`kafka.security.authorizer.AclAuthorizer`	Can be customized for custom authorization logic
kafka.security.authorizer.class.name	Name of the authorizer class to use	`org.apache.kafka.server.authorizer.AclAuthorizer`	Used in newer versions of Kafka

Example: Configuring ACL-based Authorization

authorizer.class.name=org.apache.kafka.server.authorizer.AclAuthorizer

Note: When authorization fails, Kafka returns TOPIC_AUTHORIZATION_FAILED or CLUSTER_AUTHORIZATION_FAILED errors to the client.

Common Authorization Failure Scenarios

Client attempts to access a topic without proper ACL permissions
Client tries to produce or consume from a topic it doesn't have access to
Client attempts to create a topic without the required CREATE permission

Authorization vs Authentication

Authentication: Verifies identity (e.g., SASL/SSL)
Authorization: Determines what actions a user can perform (e.g., ACLs)

5. Managing Retriable Exceptions

Exception Type	Description	Retriable?	Recommended Action
org.apache.kafka.common.errors.RetriableException	Base class for exceptions that can be retried	YES	Retry with exponential backoff
org.apache.kafka.common.errors.TimeoutException	Thrown when a Kafka operation times out	YES	Retry with jittered backoff
org.apache.kafka.common.errors.RebalanceInProgressException	Thrown during consumer rebalancing	YES	Retry after rebalance completes
org.apache.kafka.common.errors.NotLeaderForPartitionException	Producer or consumer tries to operate on a non-leader partition	YES	Refresh metadata and retry
org.apache.kafka.common.errors.UnknownTopicOrPartitionException	Topic or partition does not exist	YES	Retry after topic creation or validation
org.apache.kafka.common.errors.TopicAuthorizationException	Insufficient authorization to access topic	NO	Check ACLs and retry (if permissions change)

Example: Retry Logic with Exponential Backoff

public class RetriableKafkaProducer {
    private static final int MAX_RETRIES = 3;
    private static final long BASE_DELAY_MS = 1000;

    public void sendWithRetry(String topic, String key, String value) {
        int retries = 0;
        while (retries <= MAX_RETRIES) {
            try {
                producer.send(new ProducerRecord<>(topic, key, value)).get();
                return;
            } catch (RetriableException e) {
                retries++;
                if (retries > MAX_RETRIES) {
                    throw new RuntimeException("Max retries exceeded", e);
                }
                long delay = (long) Math.pow(2, retries) * BASE_DELAY_MS;
                Thread.sleep(delay + new Random().nextInt(100));
            } catch (Exception e) {
                throw new RuntimeException("Non-retriable error", e);
            }
        }
    }
}

Note: Always implement circuit breaker patterns for long-running operations to prevent cascading failures.

Best Practices for Retriable Exceptions

Implement exponential backoff with jitter to prevent thundering herd
Use circuit breaker patterns for persistent issues
Log retry attempts with context for debugging
Set max retry limits to prevent infinite loops
Handle RebalanceInProgressException by pausing consumer processing

6. Handling Non-Retriable Errors

Error Type	Description	Handling Strategy	Example
SerializationException	Occurs when Kafka cannot serialize or deserialize messages.	Ensure proper serialization logic and schema compatibility.	`try { producer.send(new ProducerRecord<String, MyObject>(topic, key, value)); } catch (SerializationException e) { // Log and handle serialization errors }`
TimeoutException	Occurs when a Kafka operation times out (e.g., producer send timeout).	Adjust timeouts or retry with exponential backoff.	`producer.send(record).get(5, TimeUnit.SECONDS); // May throw TimeoutException`
RecordTooLargeException	Occurs when a message exceeds the configured `max.message.bytes`.	Validate message size before sending or increase limits.	`producer.send(new ProducerRecord<String, String>(topic, key, largeValue)); // May throw RecordTooLargeException`
FencedInstanceIdException	Thrown when a consumer instance is fenced due to a new consumer joining the group.	Restart consumer or rejoin group with updated instance ID.	`try { consumer.commitSync(); } catch (FencedInstanceIdException e) { // Reinitialize consumer or handle group rebalance }`

Example: Handling Serialization Errors

try {
    ProducerRecord<String, MyObject> record = new ProducerRecord<>(topic, key, value);
    producer.send(record).get();
} catch (SerializationException e) {
    logger.error("Serialization failed for record: {}", key, e);
    // Handle or log the error appropriately
}

Note: Non-retriable errors typically require application-level handling or consumer group reconfiguration.

7. Understanding Consumer Exceptions

Exception Type	Description	Handling Recommendations
org.apache.kafka.clients.consumer.RetriableConsumerException	Base exception for retriable consumer errors.	Implement exponential backoff and retry logic. These exceptions typically indicate transient issues like network hiccups or broker unavailability.
org.apache.kafka.common.errors.TimeoutException	Thrown when a consumer operation times out.	Adjust timeout configurations (e.g., `max.poll.interval.ms`). Consider increasing `session.timeout.ms` if rebalancing is frequent.
org.apache.kafka.common.errors.RebalanceInProgressException	Occurs when a consumer is rebalancing and cannot process new records.	Ensure `enable.auto.commit` is set to false if manual commits are preferred. Handle in a loop until rebalance completes.
org.apache.kafka.common.errors.UnknownTopicOrPartitionException	Thrown when the consumer attempts to read from a topic that does not exist.	Validate topic existence before consuming. Use `KafkaAdminClient` to check topic status.
org.apache.kafka.common.errors.FencedInstanceIdException	Occurs when a consumer instance is fenced out due to a new consumer joining the group with the same `group.instance.id`.	Ensure unique `group.instance.id` per consumer instance. Restart the consumer with a new ID if necessary.
org.apache.kafka.common.errors.IllegalStateException	Thrown when consumer state is invalid for an operation (e.g., calling `commitSync` after `commitAsync`).	Review consumer state management logic. Avoid mixing sync and async commit methods.

Example: Handling RetriableConsumerException

try {
    consumer.poll(Duration.ofMillis(1000));
} catch (RetriableConsumerException e) {
    // Log and retry with exponential backoff
    logger.warn("Retriable error occurred: {}", e.getMessage());
    Thread.sleep(1000 * (int) Math.pow(2, retryCount));
    retryCount++;
}

Note: Always wrap consumer operations in try-catch blocks to handle specific exceptions and ensure graceful degradation.

8. Handling Offset Out of Range

Offset Type	Description	Handling Strategy	Notes
OUT_OF_RANGE	Consumer attempts to read from an offset that no longer exists in the topic	Reset to earliest/latest or manual adjustment	Occurs when the consumer's committed offset is outside the valid range of offsets in the topic
OFFSET_RESET	Consumer resets to earliest or latest offset when out of range	Use `auto.offset.reset` property	Defaults to `latest`, but can be set to `earliest`

Example: Configuring Offset Reset Behavior

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("auto.offset.reset", "earliest"); // Reset to earliest if out of range
props.put("enable.auto.commit", "false");

Note: When auto.offset.reset is set to earliest, the consumer will read from the beginning of the topic if it's trying to access an offset that has been deleted or is no longer available.

Warning: Setting auto.offset.reset to earliest can cause reprocessing of messages if the consumer group has not consumed all messages in the topic.

9. Dealing with Rebalance Exceptions

Exception Type	Description	Handling Strategy
RebalanceInProgressException	Thrown when a consumer attempts to perform an operation while a rebalance is in progress.	Retry the operation after the rebalance completes. Avoid manual partition assignment during rebalances.
IllegalGenerationException	Occurs when a consumer tries to rejoin the group with an outdated generation ID.	Rejoin the consumer group with a fresh generation ID. This usually happens after a consumer crash or long pause.
CommitFailedException	Occurs when a consumer fails to commit offsets, often due to session timeout or group rebalance.	Implement retry logic with exponential backoff. Consider using auto-commit or manual offset management.

Example: Handling Rebalance Exceptions in Consumer

try {
    consumer.commitSync();
} catch (RebalanceInProgressException e) {
    // Retry after rebalance completes
    Thread.sleep(1000);
    consumer.commitSync();
} catch (CommitFailedException e) {
    // Handle commit failure with retry logic
    if (retryCount < maxRetries) {
        retryCount++;
        consumer.commitSync();
    } else {
        // Log and handle failure
        logger.error("Failed to commit offsets after retries", e);
    }
}

Note: Rebalance exceptions are typically transient and can be resolved by retrying the operation once the rebalance is complete.

10. Configuring Error Handlers

Handler Type	Description	Configuration	Notes
DefaultErrorHandler	Basic error handler for uncaught exceptions	`default.error.handler`	Used when no custom handler is configured
DeadLetterQueueErrorHandler	Routes failed messages to a dead letter queue	`dlq.error.handler`	Prevents message loss in case of persistent errors
RetryableErrorHandler	Attempts to retry failed messages based on error type	`retryable.error.handler`	Supports configurable retry policies and backoff strategies

Example: Dead Letter Queue Configuration

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.ERROR_HANDLER_CONFIG, DeadLetterQueueErrorHandler.class);

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

Example: Retryable Error Handler Setup

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.ERROR_HANDLER_CONFIG, RetryableErrorHandler.class);
props.put(RetryableErrorHandler.RETRY_ATTEMPTS_CONFIG, 3);
props.put(RetryableErrorHandler.BACKOFF_MS_CONFIG, 1000);

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

Note: Error handlers can be customized to implement specific retry logic or logging behavior.

Warning: Misconfiguring error handlers can lead to infinite loops or message loss.