GH-987: DeserializationException and dead-letters

Resolves https://github.com/spring-projects/spring-kafka/issues/987

`DeadLetterPublishingRecoverer` now detects `DeserializationException`s
and sets the value of the republished record to the original incoming
`byte[]`.

- STCEH - fix `defaultClassifier` to traverse exception causes

spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,6 +20,9 @@ import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Optional;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 
@@ -34,7 +37,11 @@ import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.springframework.kafka.core.KafkaOperations;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.kafka.support.serializer.DeserializationException;
+import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
 
 /**
  * A {@link BiConsumer} that publishes a failed record to a dead-letter topic.
@@ -47,8 +54,13 @@ public class DeadLetterPublishingRecoverer implements BiConsumer<ConsumerRecord<
 
 	private static final Log logger = LogFactory.getLog(DeadLetterPublishingRecoverer.class); // NOSONAR
 
+	private static final BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition>
+		DEFAULT_DESTINATION_RESOLVER = (cr, e) -> new TopicPartition(cr.topic() + ".DLT", cr.partition());
+
 	private final KafkaTemplate<Object, Object> template;
 
+	private final Map<Class<?>, KafkaTemplate<?, ?>> templates;
+
 	private final boolean transactional;
 
 	private final BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> destinationResolver;
@@ -60,25 +72,65 @@ public class DeadLetterPublishingRecoverer implements BiConsumer<ConsumerRecord<
 	 * dead-letter topic must have at least as many partitions as the original topic.
 	 * @param template the {@link KafkaTemplate} to use for publishing.
 	 */
-	public DeadLetterPublishingRecoverer(KafkaTemplate<Object, Object> template) {
-		this(template, (cr, e) -> new TopicPartition(cr.topic() + ".DLT", cr.partition()));
+	public DeadLetterPublishingRecoverer(KafkaTemplate<? extends Object, ? extends Object> template) {
+		this(template, DEFAULT_DESTINATION_RESOLVER);
 	}
 
 	/**
 	 * Create an instance with the provided template and destination resolving function,
 	 * that receives the failed consumer record and the exception and returns a
-	 * {@link TopicPartition}. If the partition in the {@link TopicPartition} is less than 0, no
-	 * partition is set when publishing to the topic.
+	 * {@link TopicPartition}. If the partition in the {@link TopicPartition} is less than
+	 * 0, no partition is set when publishing to the topic.
 	 * @param template the {@link KafkaTemplate} to use for publishing.
 	 * @param destinationResolver the resolving function.
 	 */
-	public DeadLetterPublishingRecoverer(KafkaTemplate<Object, Object> template,
+	public DeadLetterPublishingRecoverer(KafkaTemplate<? extends Object, ? extends Object> template,
+			BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> destinationResolver) {
+		this(Collections.singletonMap(Object.class, template), destinationResolver);
+	}
+
+	/**
+	 * Create an instance with the provided templates and a default destination resolving
+	 * function that returns a TopicPartition based on the original topic (appended with
+	 * ".DLT") from the failed record, and the same partition as the failed record.
+	 * Therefore the dead-letter topic must have at least as many partitions as the
+	 * original topic. The templates map keys are classes and the value the corresponding
+	 * template to use for objects (producer record values) of that type. A
+	 * {@link java.util.LinkedHashMap} is recommended when there is more than one
+	 * template, to ensure the map is traversed in order.
+	 * @param templates the {@link KafkaTemplate}s to use for publishing.
+	 */
+	public DeadLetterPublishingRecoverer(Map<Class<?>, KafkaTemplate<? extends Object, ? extends Object>> templates) {
+		this(templates, DEFAULT_DESTINATION_RESOLVER);
+	}
+
+	/**
+	 * Create an instance with the provided templates and destination resolving function,
+	 * that receives the failed consumer record and the exception and returns a
+	 * {@link TopicPartition}. If the partition in the {@link TopicPartition} is less than
+	 * 0, no partition is set when publishing to the topic. The templates map keys are
+	 * classes and the value the corresponding template to use for objects (producer
+	 * record values) of that type. A {@link java.util.LinkedHashMap} is recommended when
+	 * there is more than one template, to ensure the map is traversed in order.
+	 * @param templates the {@link KafkaTemplate}s to use for publishing.
+	 * @param destinationResolver the resolving function.
+	 */
+	@SuppressWarnings("unchecked")
+	public DeadLetterPublishingRecoverer(Map<Class<?>, KafkaTemplate<? extends Object, ? extends Object>> templates,
 			BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> destinationResolver) {
 
-		Assert.notNull(template, "The template cannot be null");
+		Assert.isTrue(!ObjectUtils.isEmpty(templates), "At least one template is required");
 		Assert.notNull(destinationResolver, "The destinationResolver cannot be null");
-		this.template = template;
-		this.transactional = template.isTransactional();
+		this.template = templates.size() == 1 ? (KafkaTemplate<Object, Object>) templates.values().iterator().next() : null;
+		this.templates = templates;
+		this.transactional = templates.values().iterator().next().isTransactional();
+		Boolean tx = this.transactional;
+		Assert.isTrue(!templates.values()
+			.stream()
+			.map(t -> t.isTransactional())
+			.filter(t -> !t.equals(tx))
+			.findFirst()
+			.isPresent(), "All templates must have the same setting for transactional");
 		this.destinationResolver = destinationResolver;
 	}
 
@@ -87,42 +139,72 @@ public class DeadLetterPublishingRecoverer implements BiConsumer<ConsumerRecord<
 		TopicPartition tp = this.destinationResolver.apply(record, exception);
 		RecordHeaders headers = new RecordHeaders(record.headers().toArray());
 		enhanceHeaders(headers, record, exception);
-		ProducerRecord<Object, Object> outRecord = createProducerRecord(record, tp, headers);
-		if (this.transactional && !this.template.inTransaction()) {
-			this.template.executeInTransaction(t -> {
+		DeserializationException deserEx = ListenerUtils.getExceptionFromHeader(record,
+				ErrorHandlingDeserializer2.VALUE_DESERIALIZER_EXCEPTION_HEADER, logger);
+		if (deserEx == null) {
+			deserEx = ListenerUtils.getExceptionFromHeader(record,
+					ErrorHandlingDeserializer2.KEY_DESERIALIZER_EXCEPTION_HEADER, logger);
+		}
+		ProducerRecord<Object, Object> outRecord = createProducerRecord(record, tp, headers,
+				deserEx == null ? null : deserEx.getData());
+		KafkaTemplate<Object, Object> kafkaTemplate = findTemplateForValue(outRecord.value());
+		if (this.transactional && !kafkaTemplate.inTransaction()) {
+			kafkaTemplate.executeInTransaction(t -> {
 				publish(outRecord, t);
 				return null;
 			});
 		}
 		else {
-			publish(outRecord, this.template);
+			publish(outRecord, kafkaTemplate);
 		}
 	}
 
+	@SuppressWarnings("unchecked")
+	private KafkaTemplate<Object, Object> findTemplateForValue(Object value) {
+		if (this.template != null) {
+			return this.template;
+		}
+		Optional<Class<?>> key = this.templates.keySet()
+			.stream()
+			.filter((k) -> k.isAssignableFrom(value.getClass()))
+			.findFirst();
+		if (key.isPresent()) {
+			return (KafkaTemplate<Object, Object>) this.templates.get(key.get());
+		}
+		if (logger.isWarnEnabled()) {
+			logger.warn("Failed to find a template for " + value.getClass() + " attemting to use the last entry");
+		}
+		return (KafkaTemplate<Object, Object>) this.templates.values()
+				.stream()
+				.reduce((first,  second) -> second)
+				.get();
+	}
+
 	/**
-	 * Subclasses can override this method to customize the producer record to send to the DLQ.
-	 * The default implementation simply copies the key and value from the consumer record
-	 * and adds the headers. The timestamp is not set (the original timestamp is in one of
-	 * the headers).
-	 * IMPORTANT: if the partition in the {@link TopicPartition} is less than 0, it must be set to null
-	 * in the {@link ProducerRecord}.
+	 * Subclasses can override this method to customize the producer record to send to the
+	 * DLQ. The default implementation simply copies the key and value from the consumer
+	 * record and adds the headers. The timestamp is not set (the original timestamp is in
+	 * one of the headers). IMPORTANT: if the partition in the {@link TopicPartition} is
+	 * less than 0, it must be set to null in the {@link ProducerRecord}.
 	 * @param record the failed record
-	 * @param topicPartition the {@link TopicPartition} returned by the destination resolver.
+	 * @param topicPartition the {@link TopicPartition} returned by the destination
+	 * resolver.
 	 * @param headers the headers - original record headers plus DLT headers.
+	 * @param value the value to use instead of the consumer record value.
 	 * @return the producer record to send.
 	 * @see KafkaHeaders
 	 */
 	protected ProducerRecord<Object, Object> createProducerRecord(ConsumerRecord<?, ?> record,
-			TopicPartition topicPartition, RecordHeaders headers) {
+			TopicPartition topicPartition, RecordHeaders headers, @Nullable byte[] value) {
 
 		return new ProducerRecord<>(topicPartition.topic(),
 				topicPartition.partition() < 0 ? null : topicPartition.partition(),
-				record.key(), record.value(), headers);
+				record.key(), value == null ? record.value() : value, headers);
 	}
 
-	private void publish(ProducerRecord<Object, Object> outRecord, KafkaOperations<Object, Object> template) {
+	private void publish(ProducerRecord<Object, Object> outRecord, KafkaOperations<Object, Object> kafkaTemplate) {
 		try {
-			template.send(outRecord).addCallback(result -> {
+			kafkaTemplate.send(outRecord).addCallback(result -> {
 				if (logger.isDebugEnabled()) {
 					logger.debug("Successful dead-letter publication: " + result);
 				}

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -16,9 +16,6 @@
 
 package org.springframework.kafka.listener;
 
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.io.ObjectInputStream;
 import java.lang.reflect.Type;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -54,9 +51,6 @@ import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
-import org.apache.kafka.common.header.Header;
-import org.apache.kafka.common.header.Headers;
-import org.apache.kafka.common.header.internals.RecordHeaders;
 
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.kafka.KafkaException;
@@ -1341,21 +1335,9 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 		}
 
 		public void checkDeser(final ConsumerRecord<K, V> record, String headerName) {
-			Header header = record.headers().lastHeader(headerName);
-			if (header != null) {
-				try {
-					DeserializationException ex = (DeserializationException) new ObjectInputStream(
-							new ByteArrayInputStream(header.value())).readObject();
-					Headers headers = new RecordHeaders(Arrays.stream(record.headers().toArray())
-							.filter(h -> !h.key()
-									.startsWith(ErrorHandlingDeserializer2.KEY_DESERIALIZER_EXCEPTION_HEADER_PREFIX))
-							.collect(Collectors.toList()));
-					ex.setHeaders(headers);
-					throw ex;
-				}
-				catch (IOException | ClassNotFoundException | ClassCastException e) {
-					this.logger.error("Failed to deserialize a deserialization exception", e);
-				}
+			DeserializationException exception = ListenerUtils.getExceptionFromHeader(record, headerName, this.logger);
+			if (exception != null) {
+				throw exception;
 			}
 		}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,21 @@
 
 package org.springframework.kafka.listener;
 
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.util.Arrays;
+import java.util.stream.Collectors;
+
+import org.apache.commons.logging.Log;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+
+import org.springframework.kafka.support.serializer.DeserializationException;
+import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
@@ -55,4 +70,36 @@ public final class ListenerUtils {
 		return listenerType;
 	}
 
+	/**
+	 * Extract a {@link DeserializationException} from the supplied header name, if
+	 * present.
+	 * @param record the consumer record.
+	 * @param headerName the header name.
+	 * @param logger the logger for logging errors.
+	 * @return the exception or null.
+	 * @since 2.3
+	 */
+	@Nullable
+	public static DeserializationException getExceptionFromHeader(final ConsumerRecord<?, ?> record,
+			String headerName, Log logger) {
+
+		Header header = record.headers().lastHeader(headerName);
+		if (header != null) {
+			try {
+				DeserializationException ex = (DeserializationException) new ObjectInputStream(
+						new ByteArrayInputStream(header.value())).readObject();
+				Headers headers = new RecordHeaders(Arrays.stream(record.headers().toArray())
+						.filter(h -> !h.key()
+								.startsWith(ErrorHandlingDeserializer2.KEY_DESERIALIZER_EXCEPTION_HEADER_PREFIX))
+						.collect(Collectors.toList()));
+				ex.setHeaders(headers);
+				return ex;
+			}
+			catch (IOException | ClassNotFoundException | ClassCastException e) {
+				logger.error("Failed to deserialize a deserialization exception", e);
+			}
+		}
+		return null;
+	}
+
 }

spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java

@@ -241,7 +241,9 @@ public class SeekToCurrentErrorHandler implements ContainerAwareErrorHandler {
 		}
 	}
 
-	private BiPredicate<ConsumerRecord<?, ?>, Exception> getSkipPredicate(List<ConsumerRecord<?, ?>> records, Exception thrownException) {
+	private BiPredicate<ConsumerRecord<?, ?>, Exception> getSkipPredicate(List<ConsumerRecord<?, ?>> records,
+			Exception thrownException) {
+
 		if (this.classifier.classify(thrownException)) {
 			return this.failureTracker::skip;
 		}
@@ -263,7 +265,9 @@ public class SeekToCurrentErrorHandler implements ContainerAwareErrorHandler {
 		classified.put(MethodArgumentResolutionException.class, false);
 		classified.put(NoSuchMethodException.class, false);
 		classified.put(ClassCastException.class, false);
-		return new ExtendedBinaryExceptionClassifier(classified, true);
+		ExtendedBinaryExceptionClassifier defaultClassifier = new ExtendedBinaryExceptionClassifier(classified, true);
+		defaultClassifier.setTraverseCauses(true);
+		return defaultClassifier;
 	}
 
 	/**

spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentRecovererTests.java

@@ -31,6 +31,7 @@ import static org.mockito.Mockito.verifyNoMoreInteractions;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
@@ -45,6 +46,8 @@ import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.consumer.OffsetCommitCallback;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.Serializer;
 import org.junit.ClassRule;
 import org.junit.Test;
 
@@ -54,6 +57,8 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.event.ConsumerStoppedEvent;
 import org.springframework.kafka.listener.ContainerProperties.AckMode;
+import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -79,7 +84,8 @@ public class SeekToCurrentRecovererTests {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("seekTestMaxFailures", "false", embeddedKafka);
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
-		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
+		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props, null,
+				new ErrorHandlingDeserializer2<>(new JsonDeserializer<>(String.class)));
 		ContainerProperties containerProps = new ContainerProperties(topic1);
 		containerProps.setPollTimeout(10_000);
 
@@ -87,6 +93,10 @@ public class SeekToCurrentRecovererTests {
 		senderProps.put(ProducerConfig.RETRIES_CONFIG, 1);
 		DefaultKafkaProducerFactory<Object, Object> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		final KafkaTemplate<Object, Object> template = new KafkaTemplate<>(pf);
+		Serializer<?> byteArraySerializer = new ByteArraySerializer();
+		@SuppressWarnings("unchecked")
+		DefaultKafkaProducerFactory<Object, Object> dltPf =
+				new DefaultKafkaProducerFactory<Object, Object>(senderProps, null, (Serializer<Object>) byteArraySerializer);
 		final CountDownLatch latch = new CountDownLatch(1);
 		AtomicReference<String> data = new AtomicReference<>();
 		containerProps.setMessageListener((MessageListener<Integer, String>) message -> {
@@ -102,8 +112,12 @@ public class SeekToCurrentRecovererTests {
 		container.setBeanName("testSeekMaxFailures");
 		final CountDownLatch recoverLatch = new CountDownLatch(1);
 		final AtomicReference<String> failedGroupId = new AtomicReference<>();
-		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template,
-				(r, e) -> new TopicPartition(topic1DLT, r.partition())) {
+		Map<Class<?>, KafkaTemplate<?, ?>> templates = new LinkedHashMap<>();
+		templates.put(String.class, template);
+		templates.put(byte[].class, new KafkaTemplate<>(dltPf));
+		DeadLetterPublishingRecoverer recoverer =
+				new DeadLetterPublishingRecoverer(templates,
+						(r, e) -> new TopicPartition(topic1DLT, r.partition())) {
 
 			@Override
 			public void accept(ConsumerRecord<?, ?> record, Exception exception) {
@@ -126,19 +140,28 @@ public class SeekToCurrentRecovererTests {
 		container.start();
 
 		template.setDefaultTopic(topic1);
-		template.sendDefault(0, 0, "foo");
-		template.sendDefault(0, 0, "bar");
+		template.sendDefault(0, 0, "\"foo\"");
+		template.sendDefault(0, 0, "\"bar\"");
 		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
 		assertThat(data.get()).isEqualTo("bar");
 		assertThat(recoverLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(failedGroupId.get()).isEqualTo("seekTestMaxFailures");
-		container.stop();
-		Consumer<Integer, String> consumer = cf.createConsumer();
+
+		props.put(ConsumerConfig.GROUP_ID_CONFIG, "seekTestMaxFailures.dlt");
+		DefaultKafkaConsumerFactory<Integer, String> dltcf = new DefaultKafkaConsumerFactory<>(props);
+		Consumer<Integer, String> consumer = dltcf.createConsumer();
 		embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic1DLT);
 		ConsumerRecord<Integer, String> dltRecord = KafkaTestUtils.getSingleRecord(consumer, topic1DLT);
 		assertThat(dltRecord.value()).isEqualTo("foo");
+		template.sendDefault(0, 0, "junkJson");
+		dltRecord = KafkaTestUtils.getSingleRecord(consumer, topic1DLT);
+		assertThat(dltRecord.value()).isEqualTo("junkJson");
+		container.stop();
 		pf.destroy();
+		dltPf.destroy();
+		consumer.close();
 		assertThat(stopLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		verify(errorHandler, times(4)).handle(any(), any(), any(), any());
 		verify(errorHandler).clearThreadState();
 	}
 
@@ -155,7 +178,7 @@ public class SeekToCurrentRecovererTests {
 			eh.handle(new RuntimeException(), records, consumer, null);
 			fail("Expected exception");
 		}
-		catch (KafkaException e) {
+		catch (@SuppressWarnings("unused") KafkaException e) {
 			// NOSONAR
 		}
 		verify(consumer).seek(new TopicPartition("foo", 0),  0L);
@@ -197,7 +220,7 @@ public class SeekToCurrentRecovererTests {
 			eh.handle(new RuntimeException(), records, consumer, container);
 			fail("Expected exception");
 		}
-		catch (KafkaException e) {
+		catch (@SuppressWarnings("unused") KafkaException e) {
 			// NOSONAR
 		}
 		verify(consumer).seek(new TopicPartition("foo", 0),  0L);
@@ -234,7 +257,7 @@ public class SeekToCurrentRecovererTests {
 				eh.handle(new RuntimeException(), records, consumer, null);
 				fail("Expected exception");
 			}
-			catch (KafkaException e) {
+			catch (@SuppressWarnings("unused") KafkaException e) {
 				// NOSONAR
 			}
 		}

src/reference/asciidoc/kafka.adoc

@@ -3084,6 +3084,28 @@ The record sent to the dead-letter topic is enhanced with the following headers:
 * `KafkaHeaders.DLT_ORIGINAL_TIMESTAMP_TYPE`: The original timestamp type.
 
 
+Starting with version 2.3, when used in conjunction with an `ErrorHandlingDeserializer2`, the publisher will restore the record `value()`, in the dead-letter producer record, to the original value that failed to be deserialized.
+Previously, the `value()` was null and user code had to decode the `DeserializationException` from the message headers.
+In addition, you can provide multiple `KafkaTemplate` s to the publisher; this might be needed, for example, if you want to publish the `byte[]` from a `DeserializationException`, as well as values using a different serializer from records that were deserialized successfully.
+Here is an example of configuring the publisher with `KafkaTemplate` s that use a `String` and `byte[]` serializer:
+
+====
+[source, java]
+----
+@Bean
+public DeadLetterPublishingRecoverer publisher(KafkaTemplate<?, ?> stringTemplate,
+        KafkaTemplate<?, ?> bytesTemplate) {
+
+    Map<Class<?>, KafkaTemplate<?, ?>> templates = new LinkedHashMap<>();
+    templates.put(String.class, stringTemplate);
+    templates.put(byte[].class, bytesTemplate);
+    return new DeadLetterPublishingRecoverer(templates);
+}
+----
+====
+
+The publisher uses the map keys to locate a template that is suitable for the `value()` about to be published.
+A `LinkedHashMap` is recommended so that the keys are examined in order.
 
 [[kerberos]]
 ==== Kerberos

src/reference/asciidoc/whats-new.adoc

@@ -21,6 +21,10 @@ It now sets it to false automatically unless specifically set in the consumer fa
 The `SeekToCurrentErrorHandler` now treats certain exceptions as fatal and disables retry for those, invoking the recoverer on first failure.
 See <<seek-to-current>> for more information.
 
+The `DeadLetterPublishingRecoverer`, when used in conjunction with an `ErrorHandlingDeserializer2`, now sets the payload of the message sent to the dead-letter topic, to the original value that could not be deserialized.
+Previously, it was `null` and user code needed to extract the `DeserializationException` from the message headers.
+See <<dead-letters>> for more information.
+
 ==== TopicBuilder
 
 A new class `TopicBuilder` is provided for more convenient creation of `NewTopic` `@Bean` s for automatic topic provisioning.