Fix Consumer Property Overrides

If the consumer properties object contains `default` properties,
they were ignored. `Properties.forEach()` does not return default
properties.

Use `stringPropertyNames()` instead.

**cherry-pick to 2.2.x**

Author: garyrussell

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

@@ -126,7 +126,9 @@ protected KafkaConsumer<K, V> createKafkaConsumer(@Nullable String groupId, @Nul
 		}
 		boolean shouldModifyClientId = (this.configs.containsKey(ConsumerConfig.CLIENT_ID_CONFIG)
 				&& StringUtils.hasText(clientIdSuffix)) || overrideClientIdPrefix;
-		if (groupId == null && properties == null && !shouldModifyClientId) {
+		if (groupId == null
+				&& (properties == null || properties.stringPropertyNames().size() == 0)
+				&& !shouldModifyClientId) {
 			return createKafkaConsumer(this.configs);
 		}
 		else {
@@ -149,11 +151,11 @@ private KafkaConsumer<K, V> createConsumerWithAdjustedProperties(String groupId,
 							: modifiedConfigs.get(ConsumerConfig.CLIENT_ID_CONFIG)) + clientIdSuffix);
 		}
 		if (properties != null) {
-			properties.forEach((k, v) -> {
-				if (!k.equals(ConsumerConfig.CLIENT_ID_CONFIG) && !k.equals(ConsumerConfig.GROUP_ID_CONFIG)) {
-					modifiedConfigs.put((String) k, v);
-				}
-			});
+			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)));
 		}
 		return createKafkaConsumer(modifiedConfigs);
 	}

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

@@ -282,7 +282,8 @@ protected void checkTopics() {
 		if (this.containerProperties.isMissingTopicsFatal() && this.containerProperties.getTopicPattern() == null) {
 			try (Consumer<K, V> consumer =
 					this.consumerFactory.createConsumer(this.containerProperties.getGroupId(),
-							this.containerProperties.getClientId(), null)) {
+							this.containerProperties.getClientId(), null,
+							this.containerProperties.getConsumerProperties())) {
 				if (consumer != null) {
 					String[] topics = this.containerProperties.getTopics();
 					if (topics == null) {

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

@@ -617,7 +617,7 @@ public void setMissingTopicsFatal(boolean missingTopicsFatal) {
 	 * name(s) in the consumer factory.
 	 * {@code group.id} and {@code client.id} are ignored.
 	 * @return the properties.
-	 * @since 2.1.4
+	 * @since 2.2.4
 	 * @see org.apache.kafka.clients.consumer.ConsumerConfig
 	 * @see #setGroupId(String)
 	 * @see #setClientId(String)
@@ -633,7 +633,7 @@ public Properties getConsumerProperties() {
 	 * name(s) in the consumer factory.
 	 * {@code group.id} and {@code client.id} are ignored.
 	 * @param consumerProperties the properties.
-	 * @since 2.1.4
+	 * @since 2.2.4
 	 * @see org.apache.kafka.clients.consumer.ConsumerConfig
 	 * @see #setGroupId(String)
 	 * @see #setClientId(String)

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

@@ -1213,8 +1213,8 @@ public void testDefinedPartitions() throws Exception {
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props) {
 
 			@Override
-			public Consumer<Integer, String> createConsumer(String groupId, String clientIdPrefix,
-					String clientIdSuffix, Properties properties) {
+			protected KafkaConsumer<Integer, String> createKafkaConsumer(Map<String, Object> configs) {
+				assertThat(configs).containsKey(ConsumerConfig.MAX_POLL_RECORDS_CONFIG);
 				return new KafkaConsumer<Integer, String>(props) {
 
 					@Override
@@ -1238,6 +1238,10 @@ public ConsumerRecords<Integer, String> poll(Duration timeout) {
 			logger.info("defined part: " + message);
 			latch1.countDown();
 		});
+		Properties defaultProperties = new Properties();
+		defaultProperties.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "42");
+		Properties consumerProperties = new Properties(defaultProperties);
+		container1Props.setConsumerProperties(consumerProperties);
 		CountDownLatch stubbingComplete1 = new CountDownLatch(1);
 		KafkaMessageListenerContainer<Integer, String> container1 = spyOnContainer(
 				new KafkaMessageListenerContainer<>(cf, container1Props), stubbingComplete1);
@@ -1266,6 +1270,7 @@ public ConsumerRecords<Integer, String> poll(Duration timeout) {
 			logger.info("defined part: " + message);
 			latch2.countDown();
 		});
+		container2Props.setConsumerProperties(consumerProperties);
 		CountDownLatch stubbingComplete2 = new CountDownLatch(1);
 		KafkaMessageListenerContainer<Integer, String> container2 = spyOnContainer(
 				new KafkaMessageListenerContainer<>(cf, container2Props), stubbingComplete2);