GH-2038: Fix DelegatingByTopicSerialization Config

Resolves #2038

- Don't add duplicate delegates
- Configure pre-loaded delegates

* Fix duplicate check.

Author: garyrussell

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

@@ -21,6 +21,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.regex.Pattern;
 
@@ -74,6 +75,8 @@ public abstract class DelegatingByTopicSerialization<T extends Closeable> implem
 
 	private final Map<Pattern, T> delegates = new ConcurrentHashMap<>();
 
+	private final Set<String> patterns = ConcurrentHashMap.newKeySet();
+
 	private T defaultDelegate;
 
 	private boolean forKeys;
@@ -84,7 +87,11 @@ public DelegatingByTopicSerialization() {
 	}
 
 	public DelegatingByTopicSerialization(Map<Pattern, T> delegates, T defaultDelegate) {
+		Assert.notNull(delegates, "'delegates' cannot be null");
+		Assert.notNull(defaultDelegate, "'defaultDelegate' cannot be null");
 		this.delegates.putAll(delegates);
+		delegates.keySet().forEach(pattern -> Assert.isTrue(this.patterns.add(pattern.pattern()),
+				"Duplicate pattern: " + pattern.pattern()));
 		this.defaultDelegate = defaultDelegate;
 	}
 
@@ -98,6 +105,9 @@ public void setCaseSensitive(boolean caseSensitive) {
 
 	@SuppressWarnings(UNCHECKED)
 	protected void configure(Map<String, ?> configs, boolean isKey) {
+		if (this.delegates.size() > 0) {
+			this.delegates.values().forEach(delegate -> configureDelegate(configs, isKey, delegate));
+		}
 		this.forKeys = isKey;
 		Object insensitive = configs.get(CASE_SENSITIVE);
 		if (insensitive instanceof String) {
@@ -107,7 +117,7 @@ else if (insensitive instanceof Boolean) {
 			this.cased = (Boolean) insensitive;
 		}
 		String configKey = defaultKey();
-		if (configKey != null) {
+		if (configKey != null && configs.containsKey(configKey)) {
 			buildDefault(configs, configKey, isKey, configs.get(configKey));
 		}
 		configKey = configKey();
@@ -146,6 +156,10 @@ private void processMap(Map<String, ?> configs, boolean isKey, String configKey,
 	protected void build(Map<String, ?> configs, boolean isKey, String configKey, Object delegate, Pattern pattern) {
 
 		if (isInstance(delegate)) {
+			if (pattern != null && !this.patterns.add(pattern.pattern())) {
+				LOGGER.debug(() -> "Delegate already configured for " + pattern.pattern());
+				return;
+			}
 			this.delegates.put(pattern, (T) delegate);
 			configureDelegate(configs, isKey, (T) delegate);
 		}
@@ -208,8 +222,9 @@ private Map<Pattern, T> createDelegates(String mappings, Map<String, ?> configs,
 		return delegateMap;
 	}
 
+	@Nullable
 	private T createInstanceAndConfigure(Map<String, ?> configs, boolean isKey,
-			Map<Pattern, T> delegates2, Pattern pattern, String className) {
+			Map<Pattern, T> delegates2, @Nullable Pattern pattern, String className) {
 
 		try {
 			Class<?> clazz = ClassUtils.forName(className.trim(), ClassUtils.getDefaultClassLoader());
@@ -240,6 +255,10 @@ else if (key instanceof String) {
 	protected T instantiateAndConfigure(Map<String, ?> configs, boolean isKey, Map<Pattern, T> delegates2,
 			@Nullable Pattern pattern, Class<?> clazz) {
 
+		if (pattern != null && !this.patterns.add(pattern.pattern())) {
+			LOGGER.debug(() -> "Delegate already configured for " + pattern.pattern());
+			return null;
+		}
 		try {
 			@SuppressWarnings(UNCHECKED)
 			T delegate = (T) clazz.getDeclaredConstructor().newInstance();
@@ -282,7 +301,7 @@ protected T findDelegate(String topic) {
 		}
 		if (delegate == null) {
 			throw new IllegalStateException(
-					"No serializer found for topic '" + topic + "'");
+					"No (de)serializer found for topic '" + topic + "'");
 		}
 		return delegate;
 	}

spring-kafka/src/test/java/org/springframework/kafka/support/serializer/SerializationIntegrationTests.java

@@ -0,0 +1,95 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+import java.util.Map;
+import java.util.regex.Pattern;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.listener.ContainerProperties;
+import org.springframework.kafka.listener.KafkaMessageListenerContainer;
+import org.springframework.kafka.listener.MessageListener;
+import org.springframework.kafka.test.condition.EmbeddedKafkaCondition;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+
+/**
+ * @author Gary Russell
+ * @since 2.8.1
+ *
+ */
+@EmbeddedKafka(topics = SerializationIntegrationTests.DBTD_TOPIC)
+public class SerializationIntegrationTests {
+
+	public static final String DBTD_TOPIC = "dbtd";
+
+	@Test
+	void configurePreLoadedDelegates() {
+		Map<String, Object> consumerProps =
+				KafkaTestUtils.consumerProps(DBTD_TOPIC, "false", EmbeddedKafkaCondition.getBroker());
+		consumerProps.put(DelegatingByTopicDeserializer.VALUE_SERIALIZATION_TOPIC_CONFIG, DBTD_TOPIC + ":"
+				+ TestDeserializer.class.getName());
+		TestDeserializer testDeser = new TestDeserializer();
+		DelegatingByTopicDeserializer delegating = new DelegatingByTopicDeserializer(
+				Map.of(Pattern.compile(DBTD_TOPIC), testDeser), new StringDeserializer());
+		DefaultKafkaConsumerFactory<String, Object> cFact =
+				new DefaultKafkaConsumerFactory<String, Object>(consumerProps,
+					new StringDeserializer(), delegating);
+		ContainerProperties props = new ContainerProperties(DBTD_TOPIC);
+		props.setMessageListener(mock(MessageListener.class));
+		KafkaMessageListenerContainer<String, Object> container = new KafkaMessageListenerContainer<>(cFact, props);
+		container.start();
+		assertThat(KafkaTestUtils.getPropertyValue(container, "listenerConsumer.consumer.valueDeserializer"))
+				.isSameAs(delegating);
+		Map<?, ?> delegates = KafkaTestUtils.getPropertyValue(delegating, "delegates", Map.class);
+		assertThat(delegates).hasSize(1);
+		assertThat(delegates.values().iterator().next()).isSameAs(testDeser);
+		assertThat(testDeser.configured).isTrue();
+		container.stop();
+	}
+
+	static class TestDeserializer implements Deserializer<Object> {
+
+		boolean configured;
+
+		@Override
+		public void configure(Map<String, ?> configs, boolean isKey) {
+			this.configured = true;
+		}
+
+		@Override
+		public Object deserialize(String topic, byte[] data) {
+			return null;
+		}
+
+		@Override
+		public Object deserialize(String topic, Headers headers, byte[] data) {
+			return Deserializer.super.deserialize(topic, headers, data);
+		}
+
+	}
+
+}
+