GH-1964: Add DelegatingByTopic(De)Serializer

Resolves #1964

Select the `Serializer`/`Deserializer` based on the topic name.

* Extract common code; fix removeDelegate; add default; CASE_SENSITIVE can be Boolean.

Author: garyrussell

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

@@ -4136,6 +4136,8 @@ Alternatively, as long as you don't use the fluent API to configure properties,
 [[delegating-serialization]]
 ===== Delegating Serializer and Deserializer
 
+===== Using Headers
+
 Version 2.3 introduced the `DelegatingSerializer` and `DelegatingDeserializer`, which allow producing and consuming records with different key and/or value types.
 Producers must set a header `DelegatingSerializer.VALUE_SERIALIZATION_SELECTOR` to a selector value that is used to select which serializer to use for the value and `DelegatingSerializer.KEY_SERIALIZATION_SELECTOR` for the key; if a match is not found, an `IllegalStateException` is thrown.
 
@@ -4178,15 +4180,55 @@ For another technique to send different types to different topics, see <<routing
 ----
 @Bean
 public ProducerFactory<Integer, Object> producerFactory(Map<String, Object> config) {
-	return new DefaultKafkaProducerFactory<>(config,
-			null, new DelegatingByTypeSerializer(Map.of(
-					byte[].class, new ByteArraySerializer(),
-					Bytes.class, new BytesSerializer(),
-					String.class, new StringSerializer())));
+    return new DefaultKafkaProducerFactory<>(config,
+            null, new DelegatingByTypeSerializer(Map.of(
+                    byte[].class, new ByteArraySerializer(),
+                    Bytes.class, new BytesSerializer(),
+                    String.class, new StringSerializer())));
+}
+----
+====
+
+===== By Topic
+
+Starting with version 2.8, the `DelegatingByTopicSerializer` and `DelegatingByTopicDeserializer` allow selection of a serializer/deserializer based on the topic name.
+Regex `Pattern` s are used to lookup the instance to use.
+The map can be configured using a constructor, or via properties (a comma delimited list of `pattern:serializer`).
+
+====
+[source, java]
+----
+producerConfigs.put(DelegatingByTopicSerializer.VALUE_SERIALIZATION_TOPIC_CONFIG,
+            "topic[0-4]:" + ByteArraySerializer.class.getName()
+        + ", topic[5-9]:" + StringSerializer.class.getName());
+...
+ConsumerConfigs.put(DelegatingByTopicDeserializer.VALUE_SERIALIZATION_TOPIC_CONFIG,
+            "topic[0-4]:" + ByteArrayDeserializer.class.getName()
+        + ", topic[5-9]:" + StringDeserializer.class.getName());
+----
+====
+
+Use `KEY_SERIALIZATION_TOPIC_CONFIG` when using this for keys.
+
+====
+[source, java]
+----
+@Bean
+public ProducerFactory<Integer, Object> producerFactory(Map<String, Object> config) {
+    return new DefaultKafkaProducerFactory<>(config,
+            null,
+            new DelegatingByTopicSerializer(Map.of(
+                    Pattern.compile("topic[0-4]"), new ByteArraySerializer(),
+                    Pattern.compile("topic[5-9]"), new StringSerializer())),
+                    new JsonSerializer<Object>());  // default
 }
 ----
 ====
 
+You can specify a default serializer/deserializer to use when there is no pattern match using `DelegatingByTopicSerialization.KEY_SERIALIZATION_TOPIC_DEFAULT` and `DelegatingByTopicSerialization.VALUE_SERIALIZATION_TOPIC_DEFAULT`.
+
+An additional property `DelegatingByTopicSerialization.CASE_SENSITIVE` (default `true`), when set to `false` makes the topic lookup case insensitive.
+
 [[retrying-deserialization]]
 ===== Retrying Deserializer
 
@@ -4261,7 +4303,7 @@ For convenience, starting with version 2.3, the framework also provides a `Strin
 
 Starting with version 2.7.1, message payload conversion can be delegated to a `spring-messaging` `SmartMessageConverter`; this enables conversion, for example, to be based on the `MessageHeaders.CONTENT_TYPE` header.
 
-IMPORTANT:	The `KafkaMessageConverter.fromMessage()` method is called for outbound conversion to a `ProducerRecord` with the message payload in the `ProducerRecord.value()` property.
+IMPORTANT:    The `KafkaMessageConverter.fromMessage()` method is called for outbound conversion to a `ProducerRecord` with the message payload in the `ProducerRecord.value()` property.
 The `KafkaMessageConverter.toMessage()` method is called for inbound conversion from `ConsumerRecord` with the payload being the `ConsumerRecord.value()` property.
 The `SmartMessageConverter.toMessage()` method is called to create a new outbound `Message<?>` from the `Message` passed to`fromMessage()` (usually by `KafkaTemplate.send(Message<?> msg)`).
 Similarly, in the `KafkaMessageConverter.toMessage()` method, after the converter has created a new `Message<?>` from the `ConsumerRecord`, the `SmartMessageConverter.fromMessage()` method is called and then the final inbound message is created with the newly converted payload.
@@ -5065,13 +5107,13 @@ The error handler can be configured with one or more `RetryListener` s, receivin
 @FunctionalInterface
 public interface RetryListener {
 
-	void failedDelivery(ConsumerRecord<?, ?> record, Exception ex, int deliveryAttempt);
+    void failedDelivery(ConsumerRecord<?, ?> record, Exception ex, int deliveryAttempt);
 
-	default void recovered(ConsumerRecord<?, ?> record, Exception ex) {
-	}
+    default void recovered(ConsumerRecord<?, ?> record, Exception ex) {
+    }
 
-	default void recoveryFailed(ConsumerRecord<?, ?> record, Exception original, Exception failure) {
-	}
+    default void recoveryFailed(ConsumerRecord<?, ?> record, Exception original, Exception failure) {
+    }
 
 }
 ----
@@ -5322,13 +5364,13 @@ Starting with version 2.7, the processor can be configured with one or more `Ret
 @FunctionalInterface
 public interface RetryListener {
 
-	void failedDelivery(ConsumerRecord<?, ?> record, Exception ex, int deliveryAttempt);
+    void failedDelivery(ConsumerRecord<?, ?> record, Exception ex, int deliveryAttempt);
 
-	default void recovered(ConsumerRecord<?, ?> record, Exception ex) {
-	}
+    default void recovered(ConsumerRecord<?, ?> record, Exception ex) {
+    }
 
-	default void recoveryFailed(ConsumerRecord<?, ?> record, Exception original, Exception failure) {
-	}
+    default void recoveryFailed(ConsumerRecord<?, ?> record, Exception original, Exception failure) {
+    }
 
 }
 ----

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

@@ -56,3 +56,9 @@ See <<error-handlers>> for more information.
 ==== Listener Container Changes
 
 The `interceptBeforeTx` container property is now `true` by default.
+
+[[x28-serializers]]
+==== Serializer/Deserializer Changes
+
+The `DelegatingByTopicSerializer` and `DelegatingByTopicDeserializer` are now provided.
+See <<delegating-serialization>> for more information.

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

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2019-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.Map;
+import java.util.regex.Pattern;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.Deserializer;
+
+/**
+ * A {@link Deserializer} that delegates to other deserializers based on the topic name.
+ *
+ * @author Gary Russell
+ * @since 2.8
+ *
+ */
+public class DelegatingByTopicDeserializer extends DelegatingByTopicSerialization<Deserializer<?>>
+		implements Deserializer<Object> {
+
+	/**
+	 * Construct an instance that will be configured in {@link #configure(Map, boolean)}
+	 * with consumer properties.
+	 */
+	public DelegatingByTopicDeserializer() {
+	}
+
+	/**
+	 * Construct an instance with the supplied mapping of topic name patterns to delegate
+	 * deserializers.
+	 * @param delegates the map of delegates.
+	 * @param defaultDelegate the default to use when no topic name match.
+	 */
+	public DelegatingByTopicDeserializer(Map<Pattern, Deserializer<?>> delegates, Deserializer<?> defaultDelegate) {
+		super(delegates, defaultDelegate);
+	}
+
+	@Override
+	public void configure(Map<String, ?> configs, boolean isKey) {
+		super.configure(configs, isKey);
+	}
+
+	@Override
+	protected Deserializer<?> configureDelegate(Map<String, ?> configs, boolean isKey, Deserializer<?> delegate) {
+		delegate.configure(configs, isKey);
+		return delegate;
+	}
+
+	@Override
+	protected boolean isInstance(Object delegate) {
+		return delegate instanceof Deserializer;
+	}
+
+	@Override
+	public Object deserialize(String topic, byte[] data) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public Object deserialize(String topic, Headers headers, byte[] data) {
+		return findDelegate(topic).deserialize(topic, headers, data);
+	}
+
+}