CODEWITHSUNDEEP
javaapache-kafkakafka-producer-api
James Bond
James Bond

Reputation: 25

Creating a dead letter queue

How can I create a dead letter queue and test it out? I have a producer and consumer. I also have producer configs like acks, retries.

Producer.java

package org.timothy.producer;

import org.apache.kafka.clients.producer.*;
import org.timothy.producer.common.AppConfigs;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.timothy.producer.common.PropConfigs;

import java.util.concurrent.ExecutionException;

public class Producer{

    private static final Logger logger = LogManager.getLogger(Producer.class);
    public static void main(String[] args) {
        logger.info("Creating Kafka Producer...");
        KafkaProducer<Integer, String> producer = PropConfigs.prodProps();

        logger.info("Start sending messages...");

        for (int i = 1; i <= AppConfigs.numEvents; i++) {
            ProducerRecord<Integer, String> record = new ProducerRecord<>(AppConfigs.topicName, "This is Message: " + i);
            try {
                RecordMetadata metadata = producer.send(record).get();
                System.out.println("Record sent with key " + i + " to partition " + metadata.partition()
                        + " with offset " + metadata.offset());
            }
            catch (ExecutionException e) {
                System.out.println("Error in sending record");
                //System.out.println(e);
                e.printStackTrace();
            }
            catch (InterruptedException e) {
                System.out.println("Error in sending record");
                //System.out.println(e);
            }
        }
        producer.flush();
        producer.close();
        logger.info("Finished - Closing Kafka Producer.");
    }
}

Consumer.java

package org.timothy.producer;

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.timothy.producer.common.AppConfigs;
import org.timothy.producer.common.PropConfigs;
import java.time.Duration;
import java.util.Collections;


public class Consumer{

    private static final Logger logger = LogManager.getLogger(Consumer.class);
    public static void main(String[] args) {
        KafkaConsumer<Integer, String> consumer = PropConfigs.consProps();

        consumer.subscribe(Collections.singleton(AppConfigs.topicName));
        int noMessageFound = 0;

        while(true){
            ConsumerRecords<Integer,String> records = consumer.poll(Duration.ofMillis(1000));
            
            if (records.count() == 0) {
                noMessageFound++;
                if (noMessageFound > AppConfigs.MAX_NO_MESSAGE_FOUND_COUNT)
                    break;
                else
                    continue;
            }

            records.forEach(record -> logger.info("Received new record: " +
                    " Key: " + record.key() +
                    ", Value: " + record.value() +
                    ", Topic: " + record.topic() +
                    ", Partition: " + record.partition() +
                    ", Offset: " + record.offset() + "\n"
            ));
            consumer.commitAsync();

        }

    }
}

How do I implement DLQ in my code? How to test the retries, when I run my program I don't encounter errors that's why I don't know if the program retries or not.

Upvotes: 1

Views: 754

Answers (1)

OneCricketeer
OneCricketeer

Reputation: 191681

when I run my program I don't encounter errors

Then introduce them. Create a counter, when it hits a certain value, throw a RuntimeException. In the catch block, use your producer instance to send an event to a new topic.

If you want to test retries, shutdown the broker or introduce some other network exception

Upvotes: 1

Related Questions