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

Fix Consumer Property Overrides

Browse Source 
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**
This commit is contained in:
Gary Russell
2019-02-23 11:36:34 -05:00
committed by Artem Bilan
parent 45ddac7fea
commit 46aea174a3
4 changed files with 19 additions and 11 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

@@ -126,7 +126,9 @@ public class DefaultKafkaConsumerFactory<K, V> implements ConsumerFactory<K, V>
}
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 @@ public class DefaultKafkaConsumerFactory<K, V> implements ConsumerFactory<K, V>
: 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);
}

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

@@ -282,7 +282,8 @@ public abstract class AbstractMessageListenerContainer<K, V>
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) {

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

@@ -617,7 +617,7 @@ public class ContainerProperties {
* 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 class ContainerProperties {
* 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)

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

@@ -1213,8 +1213,8 @@ public class KafkaMessageListenerContainerTests {
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 class KafkaMessageListenerContainerTests {
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 class KafkaMessageListenerContainerTests {
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);