GH-4384: Ensure non-string override properties are consistent

Fixes: #4384

* Fixes gh-4384: Ensure non-string override properties are consistently applied in DefaultKafkaConsumerFactory
* Fixes gh-4384: Ensure consistent propagation of both string and non-string override properties
* Fix inconsistent handling of non-string properties in DefaultKafkaConsumerFactory
* Fix tests to validate final consumer config propagation and align with Kafka behavior
* Fix failing tests by using valid Kafka configs and proper base configs

Signed-off-by: Aman Gautam <amangautam2128@gmail.com>

(cherry picked from commit f62f0bc)

Author: Gautam-aman

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

@@ -26,7 +26,6 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Properties;
-import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Supplier;
 
@@ -368,13 +367,14 @@ protected Consumer<K, V> createKafkaConsumer(@Nullable String groupId, @Nullable
 		boolean shouldModifyClientId = (this.configs.containsKey(ConsumerConfig.CLIENT_ID_CONFIG)
 				&& StringUtils.hasText(clientIdSuffix)) || overrideClientIdPrefix;
 		if (groupId == null
-				&& (properties == null || properties.stringPropertyNames().isEmpty())
+				&& (properties == null || properties.isEmpty())
 				&& !shouldModifyClientId) {
+
 			return createKafkaConsumer(new HashMap<>(this.configs));
 		}
 		else {
-			return createConsumerWithAdjustedProperties(groupId, clientIdPrefix, properties, overrideClientIdPrefix,
-					clientIdSuffix, shouldModifyClientId);
+			return createConsumerWithAdjustedProperties(groupId, clientIdPrefix, properties,
+					overrideClientIdPrefix, clientIdSuffix, shouldModifyClientId);
 		}
 	}
 
@@ -392,18 +392,23 @@ private Consumer<K, V> createConsumerWithAdjustedProperties(@Nullable String gro
 							: modifiedConfigs.get(ConsumerConfig.CLIENT_ID_CONFIG)) + clientIdSuffix);
 		}
 		if (properties != null) {
-			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()));
+			for (Map.Entry<Object, Object> entry : properties.entrySet()) {
+
+				if (entry.getKey() instanceof String key
+						&& !ConsumerConfig.CLIENT_ID_CONFIG.equals(key)
+						&& !ConsumerConfig.GROUP_ID_CONFIG.equals(key)) {
+
+					Object value = entry.getValue();
+
+					if (value == null) {
+						value = properties.getProperty(key);
+					}
+
+					if (value != null) {
+						modifiedConfigs.put(key, value);
+					}
+				}
+			}
 			checkInaccessible(properties, modifiedConfigs);
 		}
 		return createKafkaConsumer(modifiedConfigs);

spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java

@@ -99,6 +99,62 @@ public void testSupplierProvidedDeserializersAreNotShared() {
 		assertThat(target.getKeyDeserializer()).isNotSameAs(target.getKeyDeserializer());
 	}
 
+	@Test
+	public void testMixedTypeOverridesApplied() {
+		Map<String, Object> configs = KafkaTestUtils.consumerProps(this.embeddedKafka, "testGroup", false);
+
+		Properties overrides = new Properties();
+		overrides.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2);
+		overrides.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+		AtomicReference<ConsumerConfig> capturedConfig = new AtomicReference<>();
+
+		DefaultKafkaConsumerFactory<String, String> factory =
+				new DefaultKafkaConsumerFactory<String, String>(configs) {
+
+					@Override
+					protected KafkaConsumer<String, String> createKafkaConsumer(Map<String, Object> configProps) {
+						capturedConfig.set(new ConsumerConfig(configProps));
+						return mock(KafkaConsumer.class);
+					}
+				};
+
+		factory.createConsumer(null, null, null, overrides);
+
+		ConsumerConfig consumerConfig = capturedConfig.get();
+
+		assertThat(consumerConfig.getInt(ConsumerConfig.MAX_POLL_RECORDS_CONFIG)).isEqualTo(2);
+		assertThat(consumerConfig.getString(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)).isEqualTo("earliest");
+	}
+
+	@Test
+	public void testIntegerAndStringOverridesTogether() {
+		Map<String, Object> configs = KafkaTestUtils.consumerProps(this.embeddedKafka, "testGroup", false);
+
+		Properties overrides = new Properties();
+		overrides.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "5");
+		overrides.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+		AtomicReference<ConsumerConfig> capturedConfig = new AtomicReference<>();
+
+		DefaultKafkaConsumerFactory<String, String> factory =
+				new DefaultKafkaConsumerFactory<String, String>(configs) {
+
+					@Override
+					protected Consumer<String, String> createRawConsumer(Map<String, Object> configProps) {
+						capturedConfig.set(new ConsumerConfig(configProps));
+						return mock(Consumer.class);
+					}
+				};
+
+		factory.createConsumer(null, null, null, overrides);
+
+		ConsumerConfig consumerConfig = capturedConfig.get();
+
+		assertThat(consumerConfig.getInt(ConsumerConfig.MAX_POLL_RECORDS_CONFIG)).isEqualTo(5);
+		assertThat(consumerConfig.getString(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)).isEqualTo("earliest");
+	}
+
 	@Test
 	public void testNoOverridesWhenCreatingConsumer() {
 		Map<String, Object> originalConfig = Collections.emptyMap();