Warn for non-String consumer property overrides

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java

@@ -22,11 +22,13 @@
 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 @@
  */
 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 @@ private KafkaConsumer<K, V> createConsumerWithAdjustedProperties(String groupId,
 							: 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 @@ private KafkaConsumer<K, V> createConsumerWithAdjustedProperties(String groupId,
 		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());