GH-4328: Expose native Kafka Streams DLQ configuration

Fixes: #4328

Extract the DLT record header-building logic into a new DeadLetterRecordManager
class so it can be used outside of DeadLetterPublishingRecoverer.

Add two new handlers, RecoveringProcessingExceptionHandler and
RecoveringProductionExceptionHandler, to handle processing and production errors, respectively.

Add a common AbstractRecoveringExceptionHandler that centralizes
the shared error-handling logic for all provided recovering exception handlers.

Add a KafkaStreamsDeadLetterDestinationResolver to allow users to define
dead-letter routing logic used by the Kafka Streams native DLQ in the provided
exception handler implementations.

Update StreamsBuilderFactoryBean to expose a dead-letter queue topic name
as a possible destination for all provided recovering exception handler implementations.

Signed-off-by: Loïc Greffier <loic.greffier@michelin.com>

Author: loicgreffier

spring-kafka-docs/src/main/antora/modules/ROOT/pages/streams.adoc

@@ -297,26 +297,103 @@ Spring Integration automatically provides an implementation using its `GatewayPr
 It also requires a `MessagingMessageConverter` to convert the key, value and metadata (including headers) to/from a Spring Messaging `Message<?>`.
 See {spring-integration-url}/kafka.html#streams-integration[Calling a Spring Integration Flow from a `KStream`] for more information.
 
+[[recovery-strategies]]
+== Recovery Strategies
+
+The framework provides the following exception handlers, which follow the same recovery strategies:
+
+- xref:streams.adoc#streams-deser-recovery[`RecoveringDeserializationExceptionHandler`]
+- xref:streams.adoc#streams-processing-recovery[`RecoveringProcessingExceptionHandler`]
+- xref:streams.adoc#streams-production-recovery[`RecoveringProductionExceptionHandler`]
+
+The recovery strategies are as follows, by priority order:
+
+- If a xref:streams.adoc#dead-letter-destination-resolver[`KafkaStreamsDeadLetterDestinationResolver`] is defined, resume the stream and forward the failed record to the resolved topic-partition using the native Kafka Streams DLQ.
+- If xref:streams.adoc#dead-letter-queue-topic-name-property[`errors.dead.letter.queue.topic.name`] is defined and set to a topic name, resume the stream and forward the failed record to that topic using the native Kafka Streams DLQ.
+- If a `ConsumerRecordRecoverer` implementation is defined, invoke it and resume the stream without producing dead-letter records, as handling is delegated to the `ConsumerRecordRecoverer`. For example, the provided xref:streams.adoc#dead-letter-publishing-recoverer[`DeadLetterPublishingRecoverer`] implementation can be used.
+- Fail the stream without producing dead-letter records.
+
+When a dead-letter record is published to a dead-letter topic, whether through the native Kafka Streams DLQ or via a `ConsumerRecordRecoverer` implementation, the record is enriched with the xref:kafka/annotation-error-handling.adoc#dead-letters[Spring for Apache Kafka DLT headers].
+
 [[streams-deser-recovery]]
 == Recovery from Deserialization Exceptions
 
-Version 2.3 introduced the `RecoveringDeserializationExceptionHandler` which can take some action when a deserialization exception occurs.
-Refer to the Kafka documentation about `DeserializationExceptionHandler`, of which the `RecoveringDeserializationExceptionHandler` is an implementation.
-The `RecoveringDeserializationExceptionHandler` is configured with a `ConsumerRecordRecoverer` implementation.
-The framework provides the `DeadLetterPublishingRecoverer` which sends the failed record to a dead-letter topic.
-See xref:kafka/annotation-error-handling.adoc#dead-letters[Publishing Dead-letter Records] for more information about this recoverer.
+Version 2.3 introduced the `RecoveringDeserializationExceptionHandler`, which can take some action when a deserialization exception occurs.
+It implements the `DeserializationExceptionHandler` interface (refer to the Kafka documentation for details) and follows the xref:streams.adoc#recovery-strategies[Spring for Apache Kafka recovery strategies].
+
+To configure the `RecoveringDeserializationExceptionHandler`, add the following property to your streams configuration:
+
+[source, java]
+----
+@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
+public KafkaStreamsConfiguration kStreamsConfigs() {
+    Map<String, Object> props = new HashMap<>();
+    ...
+    props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
+            RecoveringDeserializationExceptionHandler.class);
+    ...
+    return new KafkaStreamsConfiguration(props);
+}
+----
+
+[[streams-processing-recovery]]
+== Recovery from Processing Exceptions
+
+Version 4.1 introduces the `RecoveringProcessingExceptionHandler`, which can take some action when an exception occurs during stream processing.
+It implements the `ProcessingExceptionHandler` interface (refer to the Kafka documentation for details), introduced by https://cwiki.apache.org/confluence/display/KAFKA/KIP-1033%3A+Add+Kafka+Streams+exception+handler+for+exceptions+occurring+during+processing[KIP-1033]
+and follows the xref:streams.adoc#recovery-strategies[Spring for Apache Kafka recovery strategies].
+
+To enable the `RecoveringProcessingExceptionHandler`, add the following property to your streams configuration:
+
+[source, java]
+----
+@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
+public KafkaStreamsConfiguration kStreamsConfigs() {
+    Map<String, Object> props = new HashMap<>();
+    ...
+    props.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
+            RecoveringProcessingExceptionHandler.class);
+    ...
+}
+----
+
+[[streams-production-recovery]]
+== Recovery from Production Exceptions
+
+Version 4.1 introduces the `RecoveringProductionExceptionHandler`, which can take some action when an exception occurs during record production or serialization.
+It implements the `ProductionExceptionHandler` interface (refer to the Kafka documentation for details) and follows the xref:streams.adoc#recovery-strategies[Spring for Apache Kafka recovery strategies].
+
+To configure the `RecoveringProductionExceptionHandler`, add the following property to your streams configuration:
+
+[source, java]
+----
+@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
+public KafkaStreamsConfiguration kStreamsConfigs() {
+    Map<String, Object> props = new HashMap<>();
+    ...
+    props.put(StreamsConfig.PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
+            RecoveringProductionExceptionHandler.class);
+    ...
+}
+----
+
+[[dead-letter-publishing-recoverer]]
+== Dead Letter Publishing Recoverer
+
+The framework provides the `DeadLetterPublishingRecoverer`, which sends the failed record to a dead-letter topic.
+See xref:kafka/annotation-error-handling.adoc#dead-letters[Publishing Dead-letter records] for more information about this recoverer.
 
-To configure the recoverer, add the following properties to your streams configuration:
+To configure the recoverer for an exception handler, add the following properties to your streams configuration:
 
 [source, java]
 ----
 @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
 public KafkaStreamsConfiguration kStreamsConfigs() {
     Map<String, Object> props = new HashMap<>();
     ...
-    props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
+    props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
             RecoveringDeserializationExceptionHandler.class);
-    props.put(RecoveringDeserializationExceptionHandler.KSTREAM_DESERIALIZATION_RECOVERER, recoverer());
+    props.put(RecoveringDeserializationExceptionHandler.RECOVERER, recoverer());
     ...
     return new KafkaStreamsConfiguration(props);
 }
@@ -330,6 +407,60 @@ public DeadLetterPublishingRecoverer recoverer() {
 
 Of course, the `recoverer()` bean can be your own implementation of `ConsumerRecordRecoverer`.
 
+[[dead-letter-destination-resolver]]
+== Dead Letter Destination Resolver
+
+A bean of type `KafkaStreamsDeadLetterDestinationResolver` can be defined to activate native DLQ routing in the exception handlers.
+
+It determines the DLQ topic name and partition resolution logic to use, based on the error handler context, the input record of the failed processor, and the exception thrown:
+
+[source, java]
+----
+@Bean
+public KafkaStreamsDeadLetterDestinationResolver resolver() {
+    return (context, record, ex) -> {
+        if (ex instanceof FooException) return new TopicPartition("dlqTopic1", -1);
+        if (record instanceof Foo) return new TopicPartition("dlqTopic2", -1);
+        if (context.processorNodeId().equals("processor-1")) return new TopicPartition("dlqTopic3", -1);
+        return new TopicPartition("defaultDlqTopic", -1);
+    };
+}
+----
+
+A negative partition number indicates that the partition will be determined by the default partitioner.
+
+The `KafkaStreamsDeadLetterDestinationResolver` can then be injected into the exception handlers as follows:
+
+[source, java]
+----
+@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
+public KafkaStreamsConfiguration kStreamsConfigs() {
+    Map<String, Object> props = new HashMap<>();
+    ...
+    props.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
+            RecoveringProcessingExceptionHandler.class);
+    props.put(RecoveringProcessingExceptionHandler.DLQ_DESTINATION_RESOLVER, resolver());
+    ...
+    return new KafkaStreamsConfiguration(props);
+}
+----
+
+[[dead-letter-queue-topic-name-property]]
+== Dead Letter Queue Topic Name Property
+
+The Kafka Streams property `errors.dead.letter.queue.topic.name` can be defined and set to a topic name to activate native DLQ routing in the exception handlers.
+This directly specifies the DLQ topic to which the enabled exception handlers will route all failed records.
+
+Alternatively, this can be set programmatically through the `StreamsBuilderFactoryBean`:
+
+[source, java]
+----
+@Bean
+public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() {
+    return sfb -> sfb.setDeadLetterTopicName("deadLetterQueueTopic");
+}
+----
+
 [[kafka-streams-iq-support]]
 == Interactive Query Support
 

spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc

@@ -36,3 +36,20 @@ See xref:kafka/kafka-queues.adoc#share-acknowledgment-api[ShareAcknowledgment AP
 
 The default value of `sameIntervalTopicReuseStrategy` in `RetryTopicConfigurationBuilder` has been changed from `MULTIPLE_TOPICS` to `SINGLE_TOPIC` to align with the `@RetryableTopic` annotation default.
 See xref:retrytopic/topic-naming.adoc[Topic Naming] for more information.
+
+[[x41-kafka-streams-native-dlq]]
+=== Kafka Streams Native DLQ Support
+
+Spring for Apache Kafka now provides exception handlers that support Kafka Streams DLQ (KIP-1034, Kafka 4.2):
+
+- xref:streams.adoc#streams-deser-recovery[`RecoveringDeserializationExceptionHandler`] (updated).
+- xref:streams.adoc#streams-processing-recovery[`RecoveringProcessingExceptionHandler`] (new).
+- xref:streams.adoc#streams-production-recovery[`RecoveringProductionExceptionHandler`] (new).
+
+The provided exception handlers support multiple Kafka Streams DLQ enabling strategies and dead-letter topic resolution options:
+
+- The Kafka Streams property xref:streams.adoc#dead-letter-queue-topic-name-property[`errors.dead.letter.queue.topic.name`].
+- An implementation of the new xref:streams.adoc#dead-letter-destination-resolver[`KafkaStreamsDeadLetterDestinationResolver`] functional interface for dynamic resolution.
+- The `setDeadLetterTopicName()` method of the xref:streams.adoc#dead-letter-queue-topic-name-property[`StreamsBuilderFactoryBean`].
+
+`RecoveringDeserializationExceptionHandler.KSTREAM_DESERIALIZATION_RECOVERER` has been deprecated in favor of `RecoveringDeserializationExceptionHandler.RECOVERER` to align with the new handlers properties.

spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java

@@ -105,6 +105,8 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 
 	private @Nullable StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler;
 
+	private @Nullable String deadLetterTopicName;
+
 	private boolean autoStartup = true;
 
 	private int phase = Integer.MAX_VALUE - 1000; // NOSONAR magic #
@@ -163,6 +165,24 @@ public synchronized void setBeanName(String name) {
 		this.beanName = name;
 	}
 
+	/**
+	 * Set the dead letter topic name.
+	 * @param deadLetterTopicName the dead letter topic name.
+	 * @since 4.1.0
+	 */
+	public void setDeadLetterTopicName(String deadLetterTopicName) {
+		this.deadLetterTopicName = deadLetterTopicName;
+	}
+
+	/**
+	 * Get the dead letter topic name.
+	 * @return the dead letter topic name.
+	 * @since 4.1.0
+	 */
+	public @Nullable String getDeadLetterTopicName() {
+		return this.deadLetterTopicName;
+	}
+
 	/**
 	 * Set the streams configuration {@link Properties} on this factory.
 	 * @param streamsConfig the streams configuration.
@@ -365,6 +385,9 @@ public void start() {
 				try {
 					Assert.state(this.properties != null,
 							"streams configuration properties must not be null");
+					if (this.deadLetterTopicName != null) {
+						this.properties.put(StreamsConfig.ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG, this.deadLetterTopicName);
+					}
 					this.kafkaStreams = this.kafkaStreamsCustomizer.initKafkaStreams(
 							this.topology, this.properties, this.clientSupplier
 					);