Implementing Dead Letter Queues

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("group.id", "my-group");
props.put("enable.auto.commit", "false");
props.put("max.poll.records", "10");
props.put("session.timeout.ms", "30000");
props.put("heartbeat.interval.ms", "10000");
props.put("max.poll.interval.ms", "300000");
props.put("retry.backoff.ms", "1000");
props.put("max.in.flight.requests.per.connection", "1");
props.put("acks", "all");
props.put("retries", "3");
props.put("max.poll.interval.ms", "300000");
props.put("deadletterqueue.topic.name", "my-dlq-topic");

kafka-topics.sh --create --topic my-topic-dlq --partitions 1 --replication-factor 1 --bootstrap-server localhost:9092

{
  "deadletter.topic.name": "dlq-topic",
  "deadletter.topic.replication.factor": 3,
  "deadletter.topic.retention.ms": 604800000
}

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "consumer-group");
props.put("enable.auto.commit", "true");
props.put("auto.offset.reset", "earliest");
props.put("deadletter.topic.name", "dlq-topic");

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "dlq-example");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.DEFAULT_DESERIALIZER_CLASS_CONFIG, Serdes.String().deserializer().getClass());
props.put(StreamsConfig.DEFAULT_SERIALIZER_CLASS_CONFIG, Serdes.String().serializer().getClass());

StreamsConfig config = new StreamsConfig(props);

{
  "name": "my-connector",
  "config": {
    "connector.class": "MyConnectorClass",
    "tasks.max": "1",
    "errors.log.enable": "true",
    "errors.log.include.messages": "true",
    "errors.deadletterqueue.topic.name": "my-connector-dlq-topic",
    "errors.deadletterqueue.topic.replication.factor": "3"
  }
}

const errorMetadata = {
  "error.timestamp": new Date().toISOString(),
  "error.type": "SerializationError",
  "error.message": "Failed to deserialize message from topic user-events",
  "original.topic": "user-events",
  "original.partition": 3,
  "original.offset": 12345
};

const dlqMessage = {
  ...errorMetadata,
  payload: rawMessagePayload
};

producer.send({
  topic: "user-events-dlq",
  messages: [{ value: JSON.stringify(dlqMessage) }]
});

# Prometheus configuration (prometheus.yml)
scrape_configs:
  - job_name: 'kafka-dlq-metrics'
    static_configs:
      - targets: ['localhost:8080']

{
  "panels": [
    {
      "title": "DLQ Message Count",
      "targets": [
        {
          "expr": "kafka_dlq_message_count"
        }
      ]
    }
  ]
}

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

const kafka = new Kafka({
  clientId: "dlq-reprocessor",
  brokers: ["localhost:9092"]
});

const consumer = kafka.consumer({ groupId: "dlq-group" });
const producer = kafka.producer();

async function reprocessDLQMessages() {
  await consumer.connect();
  await producer.connect();

  await consumer.subscribe({ topic: "my-topic-dlq" });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      const originalMessage = message.value.toString();
      console.log("Reprocessing message:", originalMessage);

      // Send to the original topic for reprocessing
      await producer.send({
        topic: "my-topic",
        messages: [
          {
            value: originalMessage
          }
        ]
      });

      console.log("Message reprocessed successfully");
    }
  });
}

reprocessDLQMessages().catch(console.error);

{"retention.ms": 1209600000, "retention.bytes": 1073741824}

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;

public class CircuitBreakerProducer {
    private KafkaProducer<String, String> producer;
    private boolean isOpen = false;
    private int failureCount = 0;
    private static final int MAX_FAILURES = 3;

    public CircuitBreakerProducer() {
        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 void sendWithCircuitBreaker(String topic, String key, String value) {
        if (isOpen) {
            System.out.println("Circuit is OPEN, rejecting message");
            return;
        }

        try {
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
            producer.send(record).get(); // Synchronous send for simplicity
            failureCount = 0; // Reset on success
        } catch (Exception e) {
            failureCount++;
            System.out.println("Failure count: " + failureCount);
            if (failureCount >= MAX_FAILURES) {
                isOpen = true;
                System.out.println("Circuit OPEN due to too many failures");
            }
        }
    }

    public void resetCircuit() {
        isOpen = false;
        failureCount = 0;
        System.out.println("Circuit RESET");
    }
}