GH-2000: Doc RetryTopic Publish with Deser. Errors

Resolves #2000

Backport `DelegatingByTypeSerializer`.

This is for 2.7.x; cherry pick forward to main.

* Doc polishing.

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -4444,6 +4444,27 @@ consumerProps.put(ErrorHandlingDeserializer.VALUE_FUNCTION, FailedFooProvider.cl
 ----
 ====
 
+IMPORTANT: If the consumer is configured with an `ErrorHandlingDeserializer` it is important to configure the `KafkaTemplate` and its producer with a serializer that can handle normal objects as well as raw `byte[]` values, which result from deserialization exceptions.
+The generic value type of the template should be `Object`.
+One technique is to use the `DelegatingByTypeSerializer`; an example follows:
+
+====
+[source, java]
+----
+@Bean
+public ProducerFactory<String, Object> producerFactory() {
+  return new DefaultKafkaProducerFactory<>(producerConfiguration(), new StringSerializer(),
+    new DelegatingByTypeSerializer(Map.of(byte[].class, new ByteArraySerializer(),
+          MyNormalObject.class, new JsonSerializer<Object>())));
+}
+
+@Bean
+public KafkaTemplate<String, Object> kafkaTemplate() {
+  return new KafkaTemplate<>(producerFactory());
+}
+----
+====
+
 [[payload-conversion-with-batch]]
 ===== Payload Conversion with Batch Listeners
 

spring-kafka-docs/src/main/asciidoc/retrytopic.adoc

@@ -127,6 +127,27 @@ public RetryTopicConfiguration myOtherRetryTopic(KafkaTemplate<String, MyOtherPo
 
 NOTE: The retry topics' and dlt's consumers will be assigned to a consumer group with a group id that is the combination of the one with you provide in the `groupId` parameter of the `@KafkaListener` annotation with the topic's suffix. If you don't provide any they'll all belong to the same group, and rebalance on a retry topic will cause an unnecessary rebalance on the main topic.
 
+IMPORTANT: If the consumer is configured with an <<error-handling-deserializer,`ErrorHandlingDeserializer`>>, to handle deserilialization exceptions, it is important to configure the `KafkaTemplate` and its producer with a serializer that can handle normal objects as well as raw `byte[]` values, which result from deserialization exceptions.
+The generic value type of the template should be `Object`.
+One technique is to use the `DelegatingByTypeSerializer`; an example follows:
+
+====
+[source, java]
+----
+@Bean
+public ProducerFactory<String, Object> producerFactory() {
+  return new DefaultKafkaProducerFactory<>(producerConfiguration(), new StringSerializer(),
+    new DelegatingByTypeSerializer(Map.of(byte[].class, new ByteArraySerializer(),
+          MyNormalObject.class, new JsonSerializer<Object>())));
+}
+
+@Bean
+public KafkaTemplate<String, Object> kafkaTemplate() {
+  return new KafkaTemplate<>(producerFactory());
+}
+----
+====
+
 ==== Features
 
 Most of the features are available both for the `@RetryableTopic` annotation and the `RetryTopicConfiguration` beans.

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTypeSerializer.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.support.serializer;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.Serializer;
+
+import org.springframework.util.Assert;
+
+/**
+ * Delegates to a serializer based on type.
+ *
+ * @author Gary Russell
+ * @since 2.7.9
+ *
+ */
+public class DelegatingByTypeSerializer implements Serializer<Object> {
+
+	private static final String RAWTYPES = "rawtypes";
+
+	@SuppressWarnings(RAWTYPES)
+	private final Map<Class<?>, Serializer> delegates = new HashMap<>();
+
+	/**
+	 * Construct an instance with the map of delegates.
+	 * @param delegates the delegates.
+	 */
+	@SuppressWarnings(RAWTYPES)
+	public DelegatingByTypeSerializer(Map<Class<?>, Serializer> delegates) {
+		Assert.notNull(delegates, "'delegates' cannot be null");
+		Assert.noNullElements(delegates.values(), "Serializers in delegates map cannot be null");
+		this.delegates.putAll(delegates);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public void configure(Map<String, ?> configs, boolean isKey) {
+		this.delegates.values().forEach(del -> del.configure(configs, isKey));
+	}
+
+	@SuppressWarnings({ RAWTYPES, "unchecked" })
+	@Override
+	public byte[] serialize(String topic, Object data) {
+		Serializer delegate = findDelegate(data);
+		return delegate.serialize(topic, data);
+	}
+
+	@SuppressWarnings({ "unchecked", RAWTYPES })
+	@Override
+	public byte[] serialize(String topic, Headers headers, Object data) {
+		Serializer delegate = findDelegate(data);
+		return delegate.serialize(topic, headers, data);
+	}
+
+	@SuppressWarnings(RAWTYPES)
+	private Serializer findDelegate(Object data) {
+		Serializer delegate = this.delegates.get(data.getClass());
+		if (delegate == null) {
+			throw new SerializationException("No matching delegate for type: " + data.getClass().getName()
+					+ "; supported types: " + this.delegates.keySet().stream()
+							.map(clazz -> clazz.getName())
+							.collect(Collectors.toList()));
+		}
+		return delegate;
+	}
+
+
+}