GH-1552: Support more consumer properties override

Resolves #1552

Previously, non-String-valued consumer property overrides were ignored (and logged).

This is due to the unfortunate design of `Properties` being a subclass of `HashTable`
in that only String-valued properties can be accessed in the nested `default`
properties `HashTable`.

- when creating the merged `Properties` to create the consumer, copy any non-String-valued
properties from the default hash table to the new properties object
- in the DKCF, iterate over the hash table to find any other properties not already found
- log an error if we find non-String-valued default properties

* Code Polishing.

Author: garyrussell

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

@@ -18,11 +18,13 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Enumeration;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.Set;
 import java.util.function.Supplier;
 
 import org.aopalliance.aop.Advice;
@@ -283,23 +285,39 @@ private Consumer<K, V> createConsumerWithAdjustedProperties(String groupId, Stri
 							: modifiedConfigs.get(ConsumerConfig.CLIENT_ID_CONFIG)) + clientIdSuffix);
 		}
 		if (properties != null) {
-			checkForUnsupportedProps(properties);
-			properties.stringPropertyNames()
+			Set<String> stringPropertyNames = properties.stringPropertyNames();  // to get any nested default Properties
+			stringPropertyNames
 					.stream()
 					.filter(name -> !name.equals(ConsumerConfig.CLIENT_ID_CONFIG)
 							&& !name.equals(ConsumerConfig.GROUP_ID_CONFIG))
 					.forEach(name -> modifiedConfigs.put(name, properties.getProperty(name)));
+			properties.entrySet().stream()
+					.filter(entry -> !entry.getKey().equals(ConsumerConfig.CLIENT_ID_CONFIG)
+							&& !entry.getKey().equals(ConsumerConfig.GROUP_ID_CONFIG)
+							&& !stringPropertyNames.contains(entry.getKey())
+							&& entry.getKey() instanceof String)
+					.forEach(entry -> modifiedConfigs.put((String) entry.getKey(), entry.getValue()));
+			checkInaccessible(properties, modifiedConfigs);
 		}
 		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());
+	private void checkInaccessible(Properties properties, Map<String, Object> modifiedConfigs) {
+		List<Object> inaccessible = null;
+		for (Enumeration<?> propertyNames = properties.propertyNames(); propertyNames.hasMoreElements(); ) {
+			Object nextElement = propertyNames.nextElement();
+			if (!modifiedConfigs.containsKey(nextElement)) {
+				if (inaccessible == null) {
+					inaccessible = new ArrayList<>();
+				}
+				inaccessible.add(nextElement);
 			}
-		});
+		}
+		if (inaccessible != null) {
+			LOGGER.error("Non-String-valued default properties are inaccessible; use String values or "
+					+ "make them explicit properties instead of defaults: "
+					+ inaccessible);
+		}
 	}
 
 	@SuppressWarnings("resource")

spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerProperties.java

@@ -308,7 +308,8 @@ public void setCommitLogLevel(LogIfLevelEnabled.Level commitLogLevel) {
 	/**
 	 * Get the consumer properties that will be merged with the consumer properties
 	 * provided by the consumer factory; properties here will supersede any with the same
-	 * name(s) in the consumer factory.
+	 * name(s) in the consumer factory. You can add non-String-valued properties, but the
+	 * property name (hashtable key) must be String; all others will be ignored.
 	 * {@code group.id} and {@code client.id} are ignored.
 	 * @return the properties.
 	 * @see org.apache.kafka.clients.consumer.ConsumerConfig
@@ -324,8 +325,7 @@ public Properties getKafkaConsumerProperties() {
 	 * provided by the consumer factory; properties here will supersede any with the same
 	 * name(s) in the consumer factory.
 	 * {@code group.id} and {@code client.id} are ignored.
-	 * Property values must be {@link String}s; only properties returned by
-	 * {@link Properties#stringPropertyNames()} will be applied.
+	 * Property keys must be {@link String}s.
 	 * @param kafkaConsumerProperties the properties.
 	 * @see org.apache.kafka.clients.consumer.ConsumerConfig
 	 * @see #setGroupId(String)

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -592,7 +592,7 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		@SuppressWarnings(UNCHECKED)
 		ListenerConsumer(GenericMessageListener<?> listener, ListenerType listenerType) {
-			Properties consumerProperties = new Properties(this.containerProperties.getKafkaConsumerProperties());
+			Properties consumerProperties = propertiesFromProperties();
 			checkGroupInstance(consumerProperties, KafkaMessageListenerContainer.this.consumerFactory);
 			this.autoCommit = determineAutoCommit(consumerProperties);
 			this.consumer =
@@ -676,6 +676,19 @@ else if (listener instanceof MessageListener) {
 			this.subBatchPerPartition = setupSubBatchPerPartition();
 		}
 
+		private Properties propertiesFromProperties() {
+			Properties propertyOverrides = this.containerProperties.getKafkaConsumerProperties();
+			Properties props = new Properties(propertyOverrides);
+			Set<String> stringPropertyNames = props.stringPropertyNames();
+			// Copy non-string-valued properties from the default hash table to the new properties object
+			propertyOverrides.forEach((key, value) -> {
+				if (key instanceof String && !stringPropertyNames.contains(key)) {
+					props.put(key, value);
+				}
+			});
+			return props;
+		}
+
 		String getClientId() {
 			return this.clientId;
 		}

spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java

@@ -227,7 +227,10 @@ protected Consumer<Integer, String> createKafkaConsumer(String groupId, String c
 					String clientIdSuffixArg, Properties properties) {
 
 				overrides.set(properties);
-				return super.createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffixArg, properties);
+				Consumer<Integer, String> created = super.createKafkaConsumer(groupId, clientIdPrefix,
+						clientIdSuffixArg, properties);
+				assertThat(KafkaTestUtils.getPropertyValue(created, "requestTimeoutMs", Long.class)).isEqualTo(23000L);
+				return created;
 			}
 
 		};
@@ -240,8 +243,11 @@ protected Consumer<Integer, String> createKafkaConsumer(String groupId, String c
 			listenerThreadNames.add(Thread.currentThread().getName());
 			latch.countDown();
 		});
-		Properties consumerProperties = new Properties();
+		Properties nestedProps = new Properties();
+		nestedProps.put("weCantAccessThisOne", 42);
+		Properties consumerProperties = new Properties(nestedProps);
 		consumerProperties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
+		consumerProperties.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 23000);
 		containerProps.setKafkaConsumerProperties(consumerProperties);
 		final CountDownLatch rebalancePartitionsAssignedLatch = new CountDownLatch(2);
 		containerProps.setConsumerRebalanceListener(new ConsumerRebalanceListener() {