Implementing Retry Mechanisms

1. Understanding Retry Behavior

Kafka's retry behavior is managed through configuration properties and consumer/group management. When a consumer fails to process a message, it can retry the operation based on specific configurations.

Property	Default Value	Description
max.poll.interval.ms	`300000`	Maximum delay between subsequent calls to poll(). If the consumer takes longer than this, it is considered dead and will be removed from the group.
max.poll.records	`500`	Maximum number of records returned in a single call to poll().
retry.backoff.ms	`100`	Backoff time in milliseconds between retries.
enable.auto.commit	`true`	Automatically commit offsets after each poll().
auto.commit.interval.ms	`5000`	Time in milliseconds between automatic offset commits.

Example: Configuring Retry Behavior

{"bootstrap.servers": "localhost:9092", "group.id": "my-group", "enable.auto.commit": true, "max.poll.interval.ms": 60000, "retry.backoff.ms": 100}

Note: The max.poll.interval.ms setting is crucial for ensuring that consumers do not get kicked out of the consumer group due to long processing times. It should be set higher than the expected time for message processing.

Warning: Setting enable.auto.commit to false requires manual offset management, which can lead to duplicate processing if not handled correctly.

2. Configuring Retry Count

Property	Default Value	Description
retries	`3`	Number of times to retry a failed operation before giving up
retry.backoff.ms	`1000`	Initial backoff time in milliseconds between retries
retry.backoff.max.ms	`10000`	Maximum backoff time in milliseconds between retries
retry.jitter	`false`	Whether to add jitter to backoff times to prevent thundering herd

Example: Producer Retry Configuration

{
  "retries": 5,
  "retry.backoff.ms": 2000,
  "retry.backoff.max.ms": 30000,
  "retry.jitter": true
}

Note: Increasing retry count can lead to increased load on the system and potential data duplication if not handled properly.

3. Setting Retry Backoff

Property	Default Value	Description
retry.backoff.ms	`1000`	Initial backoff time in milliseconds between retries
retry.backoff.max.ms	`10000`	Maximum backoff time in milliseconds between retries
retry.jitter	`false`	Whether to apply jitter to backoff times to prevent thundering herd

Example: Configuring Retry Backoff

{
  "retry.backoff.ms": 2000,
  "retry.backoff.max.ms": 30000,
  "retry.jitter": true
}

Note: Increasing retry.backoff.ms helps reduce load on the broker during transient failures.

Warning: Setting retry.backoff.max.ms too high may cause long delays in message processing.

4. Configuring Delivery Timeout

Property	Default Value	Description
delivery.timeout.ms	`120000`	Maximum time to await delivery confirmation before considering the message failed.
max.block.ms	`60000`	Maximum time to block on send operations if the buffer is full.

Example: Setting Delivery Timeout

{
  "delivery.timeout.ms": 30000,
  "max.block.ms": 5000
}

Note: The delivery.timeout.ms setting should be higher than the max.block.ms to allow for retries and backoff strategies.

5. Understanding Retriable Exceptions

Exception Type	Description	Retriable?	Notes
org.apache.kafka.common.errors.RetriableException	Base class for exceptions that can be retried	`true`	All subclasses are retriable by default
org.apache.kafka.common.errors.TimeoutException	Thrown when a Kafka operation times out	`true`	Typically retriable, especially in producer/consumer contexts
org.apache.kafka.common.errors.RebalanceInProgressException	Thrown when a consumer is rebalancing	`true`	Should be retried after rebalance completes
org.apache.kafka.common.errors.UnknownTopicOrPartitionException	Thrown when a topic or partition is unknown	`true`	May be retried if the topic is created or partition is added
org.apache.kafka.common.errors.NotLeaderForPartitionException	Thrown when the broker is not the leader for a partition	`true`	Should be retried after leader election
org.apache.kafka.common.errors.TopicAuthorizationException	Thrown when authorization fails for a topic	`false`	Not retriable due to permission issues
org.apache.kafka.common.errors.InvalidTopicException	Thrown when a topic name is invalid	`false`	Not retriable due to malformed topic name

Note: Retriable exceptions are typically handled by Kafka's built-in retry mechanisms, but custom logic can be implemented to handle specific cases.

6. Handling Non-Retriable Errors

Error Type	Description	Handling Strategy	Notes
SerializationError	Occurs when a message cannot be serialized or deserialized.	Log and discard; notify system administrators.	Usually indicates a schema mismatch or invalid data format.
TimeoutError	Producer or consumer times out during message processing.	Retry with exponential backoff; escalate if persistent.	May be due to network issues or overloaded brokers.
AuthorizationError	Producer or consumer lacks permissions to access topic or resource.	Fail fast; log and alert for configuration review.	Indicates misconfigured ACLs or credentials.
InvalidTopicError	Attempt to produce or consume from a non-existent topic.	Fail fast; do not retry unless topic is recreated.	Requires manual intervention to create or correct topic name.
FencedInstanceException	Consumer instance is fenced out due to rebalancing or concurrent consumer groups.	Stop consumer; restart with new session ID.	Indicates concurrent consumer group usage or stale session state.

Example: Handling Serialization Errors

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092']
});

const producer = kafka.producer();

async function sendMessage() {
  try {
    await producer.connect();
    await producer.send({
      topic: 'my-topic',
      messages: [
        { value: JSON.stringify({ invalid: '<object>' }) }
      ]
    });
  } catch (error) {
    if (error.name === 'SerializationError') {
      console.error('Failed to serialize message:', error);
      // Log and discard
    } else {
      throw error;
    }
  }
}

Note: Non-retriable errors should not be retried, as they typically indicate configuration or data issues that require manual intervention.

7. Implementing Exponential Backoff

Property	Value	Description
initialDelayMs	`1000`	Initial delay before first retry attempt in milliseconds
maxDelayMs	`30000`	Maximum delay between retries in milliseconds
multiplier	`2`	Multiplier for exponential growth of delay
maxRetries	`5`	Maximum number of retry attempts before giving up

Example: Retry Configuration with Exponential Backoff

const retryConfig = {
  initialDelayMs: 1000,
  maxDelayMs: 30000,
  multiplier: 2,
  maxRetries: 5
};

function calculateDelay(retryCount, config) {
  const delay = Math.min(config.initialDelayMs * Math.pow(config.multiplier, retryCount), config.maxDelayMs);
  return delay;
}

// Example usage
for (let i = 0; i <= retryConfig.maxRetries; i++) {
  const delay = calculateDelay(i, retryConfig);
  console.log(`Retry ${i + 1} after ${delay}ms`);
}

Note: Exponential backoff helps prevent overwhelming the system with rapid retries and allows time for transient issues to resolve.

8. Using Circuit Breaker Pattern

The circuit breaker pattern is a design pattern used to detect failures and encapsulate the logic of preventing a failure from constantly recurring during a time window.

Aspect	Description	Implementation
State	Tracks the health of a service or system	Open, Half-Open, Closed
Failure Threshold	Number of failures before opening the circuit	Configurable (e.g., 5)
Timeout	Duration before transitioning to Half-Open	Configurable (e.g., 60 seconds)

Example: Circuit Breaker in Kafka Producer

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;
import java.util.concurrent.CompletableFuture;

public class CircuitBreakerKafkaProducer {
    private final KafkaProducer<String, String> producer;
    private volatile boolean isOpen = false;
    private int failureCount = 0;
    private long lastFailureTime = 0;

    public CircuitBreakerKafkaProducer() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        this.producer = new KafkaProducer<>(props);
    }

    public CompletableFuture<org.apache.kafka.clients.producer.RecordMetadata> sendWithCircuitBreaker(String topic, String key, String value) {
        if (isOpen && System.currentTimeMillis() - lastFailureTime < 60_000) {
            return CompletableFuture.failedFuture(new RuntimeException("Circuit is open"));
        }

        return producer.send(new ProducerRecord<>(topic, key, value))
                .thenApply(metadata -> {
                    failureCount = 0;
                    return metadata;
                })
                .exceptionally(throwable -> {
                    failureCount++;
                    lastFailureTime = System.currentTimeMillis();
                    if (failureCount >= 5) {
                        isOpen = true;
                    }
                    throw new RuntimeException("Failed to send message", throwable);
                });
    }

    public void resetCircuit() {
        isOpen = false;
        failureCount = 0;
    }
}

Note: The circuit breaker pattern helps in avoiding cascading failures and provides resilience to Kafka producers when downstream services are unavailable.

Summary of Circuit Breaker States

Closed: Normal operation, requests are allowed to proceed.
Open: Failures exceed threshold, all requests fail fast.
Half-Open: After timeout, a few requests are allowed to test if the service is recovered.

9. Configuring Request Timeout

Property	Default Value	Description
request.timeout.ms	`30000`	Timeout for request to broker (in milliseconds). If the request does not complete within this time, it is considered failed.
max.in.flight.requests.per.connection	`5`	Maximum number of unacknowledged requests the client will send on a single connection before blocking.

Example: Setting Request Timeout

{
  "bootstrap.servers": "localhost:9092",
  "request.timeout.ms": 60000,
  "max.in.flight.requests.per.connection": 1
}

10. Implementing Consumer Retry Logic

Property	Description	Default Value	Notes
retry.backoff.ms	Backoff time between retries in milliseconds	`1000`	Used in exponential backoff strategy
retry.max.attempts	Maximum number of retry attempts	`3`	After this, the message is sent to a dead-letter topic
retry.topic.name	Name of the dead-letter topic	`dlq-topic`	Used when retries are exhausted

Example: Consumer Retry Configuration

const consumer = new KafkaConsumer({
  brokers: ['localhost:9092'],
  groupId: 'my-group',
  retry: {
    maxAttempts: 3,
    backoffMs: 1000
  },
  deadLetterTopic: 'dlq-topic'
});

Note: The consumer must be configured to handle retries via maxAttempts and backoffMs. When retries are exhausted, the message is sent to a dead-letter topic.

Example: Handling Retry Logic in Consumer

consumer.on('message', async (message) => {
  try {
    await processMessage(message);
  } catch (error) {
    console.error('Processing failed:', error.message);
    // Retry logic handled by Kafka consumer config
  }
});