In previous tutorial, Exception Handling in Spring for Apache Kafka Streams – Part 1 and Exception Handling in Spring for Apache Kafka Streams – Part 2, we explored different types of exception handlers in Spring for Apache Kafka and Apache Kafka Streams. For implementations like *LogAndContinue…* and *LogAndFail…*, by default, when an error occurs, our application will log the error and either continue or stop. You can also publish this error message to a dead letter queue (DLQ) topic to review the error message and reprocess it later. How exactly does this work? Let’s find out in this tutorial!

I will reuse the example from the previous tutorials.

First, let’s talk a little about DLQ topics. A DLQ topic is a topic that will store all messages published to the Apache Kafka server that our application cannot process after several retry attempts. We will need to manually review them to find a suitable solution for these faulty messages.

For implementations of the Apache Kafka Streams library for the DeserializationExceptionHandler, ProcessingExceptionHandler, and ProductionExceptionHandler interfaces, to publish faulty messages to a DLQ topic, you just need to configure the StreamsConfig.ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG property with the name of the DLQ topic, for example:

Here, I’m configuring the DLQ topic name to “default-dlq-topic”!

Now, let’s run the example again with the stream topology:

Then, publish it to the users topic with key 001 and value Khanh. You will see that our application will log errors, and the error message will be published to the DLQ topic as follows:

For Spring for Apache Kafka library implementations such as RecoveringDeserializationExceptionHandler, RecoveringProcessingExceptionHandler, and RecoveringProductionExceptionHandler, by default, these implementations always use a DLQ topic to publish error messages and recover stream topology. If the DLQ topic information is not declared, our application will log errors and stop.

There are three ways we can define the DLQ topic information for these implementations, in order of priority:

• Using the implementation of the KafkaStreamsDeadLetterDestinationResolver functional interface.
• Using the StreamsConfig.ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG configuration of Apache Kafka Streams as mentioned above.
• Use the ConsumerRecordRecoverer interface implementation with the default implementation DeadLetterPublishingRecoverer.

With the KafkaStreamsDeadLetterDestinationResolver functional interface, you can resolve the DLQ topic name by using information from the ErrorHandlerContext, ConsumerRecord, and Exception objects in the parameters of the resolve() method, with a bean definition, for example:

After defining the resolver bean, you need to configure it for Spring for Apache Kafka’s Recovering…ExceptionHandler as follows:

With the second method using the StreamsConfig.ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG configuration, you can define it as above or define a bean of the StreamsBuilderFactoryBeanConfigurer interface to set the value of the DLQ topic name as follows:

With the third method, you can define a bean for the ConsumerRecordRecoverer interface using the DeadLetterPublishingRecoverer implementation as follows:

With this method, you need to define a bean for the KafkaTemplate class:

The data type of KafkaTemplate must be <byte[], byte[]> and the serializer class must be ByteArraySerializer! This is because the raw source record will be sent to the DLQ topic.

Similarly to the resolver, for this bean recoverer, you also need to configure it for the Spring for Apache Kafka’s Recovering…ExceptionHandler as follows:

Running the example again with the 3 configurations above, you will see that our application will consume the message that was just corrupted. Because it couldn’t process it, it will publish it to the DLQ topic in the order “default1-dlq-topic”.