From 8c8e56be48cfcdf36837093db1a4c1508e2f6d71 Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Fri, 13 Dec 2019 15:07:10 -0500
Subject: [PATCH] Warn for non-String consumer property overrides

---
 .../kafka/core/DefaultKafkaConsumerFactory.java    | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java
index 36312add..6f5f7fd6 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java
@@ -22,11 +22,13 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.function.Supplier;
 
+import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.serialization.Deserializer;
 
+import org.springframework.core.log.LogAccessor;
 import org.springframework.lang.Nullable;
 import org.springframework.util.StringUtils;
 
@@ -57,6 +59,8 @@ import org.springframework.util.StringUtils;
  */
 public class DefaultKafkaConsumerFactory<K, V> implements ConsumerFactory<K, V> {
 
+	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(DefaultKafkaConsumerFactory.class));
+
 	private final Map<String, Object> configs;
 
 	private Supplier<Deserializer<K>> keyDeserializerSupplier;
@@ -179,6 +183,7 @@ public class DefaultKafkaConsumerFactory<K, V> implements ConsumerFactory<K, V>
 							: modifiedConfigs.get(ConsumerConfig.CLIENT_ID_CONFIG)) + clientIdSuffix);
 		}
 		if (properties != null) {
+			checkForUnsupportedProps(properties);
 			properties.stringPropertyNames()
 					.stream()
 					.filter(name -> !name.equals(ConsumerConfig.CLIENT_ID_CONFIG)
@@ -188,6 +193,15 @@ public class DefaultKafkaConsumerFactory<K, V> implements ConsumerFactory<K, V>
 		return createKafkaConsumer(modifiedConfigs);
 	}
 
+	private void checkForUnsupportedProps(Properties properties) {
+		properties.forEach((key, value) -> {
+			if (!(key instanceof String) || !(value instanceof String)) {
+				LOGGER.warn(() -> "Property override for '" + key.toString()
+					+ "' ignored, only <String, String> properties are supported; value is a(n) " + value.getClass());
+			}
+		});
+	}
+
 	protected KafkaConsumer<K, V> createKafkaConsumer(Map<String, Object> configProps) {
 		return new KafkaConsumer<>(configProps, this.keyDeserializerSupplier.get(),
 				this.valueDeserializerSupplier.get());