Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Warn for non-String consumer property overrides

Browse Source
This commit is contained in:
Gary Russell
2019-12-13 15:07:10 -05:00
committed by Artem Bilan
parent 10127ec7aa
commit 8c8e56be48
1 changed files with 14 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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());