Example on handling processing exception in Spring Cloud Streams with Kafka Streams Binder and the functional style processor

Question

I am using Spring Cloud Streams with the Kafka Streams Binder, the functional style processor API and also multiple processors.

It's really cool to configure a processing application with multiple processors and multiple Kafka topics in this way and staying in the Spring Boot universe with /actuator, WebClient and so on. Actually I like it more than using plain Apache Kafka Streams.

BUT: I would like to integrate exception handling for exceptions occurring within the processors and sending these unprocessable messages to a DLQ. I have setup already DLQs for deserialization errors, but I found no good advice on achieving this besides sobychacko's answer on a similar question. But this is only a snippet! Does anybody have a more detailed example? I am asking this because the Spring Cloud Stream documentation on branching looks quite different.

Answer 1

Glad to hear about your usage of Spring Cloud Stream with Kafka Streams.

The reference docs you mentioned are from an older release. Please navigate to the newer docs from this page: https://spring.io/projects/spring-cloud-stream#learn

This question has come up before. See if these could help with your use case:

Error handling in Spring Cloud Kafka Streams

How to stop sending to kafka topic when control goes to catch block Functional kafka spring