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

GH-1808: Fix Container Properties Crosstalk

Browse Source 
Resolves https://github.com/spring-projects/spring-kafka/issues/1808

Each container got a reference to the same `Properties` object in
`kafkaConsumerProperties`, unless it was created via a `@KafkaListener`
with property overrides.

Do not copy the property from the factory's properties.

**Cherry-pick to all supported branches**
This commit is contained in:
Gary Russell
2021-05-21 08:27:14 -04:00
committed by Artem Bilan
parent dc881d93e7
commit 6484a2f63d
2 changed files with 4 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java
View File

@@ -401,7 +401,7 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
protected void initializeContainer(C instance, KafkaListenerEndpoint endpoint) {
ContainerProperties properties = instance.getContainerProperties();
BeanUtils.copyProperties(this.containerProperties, properties, "topics", "topicPartitions", "topicPattern",
"messageListener", "ackCount", "ackTime", "subBatchPerPartition");
"messageListener", "ackCount", "ackTime", "subBatchPerPartition", "kafkaConsumerProperties");
JavaUtils.INSTANCE
.acceptIfNotNull(this.afterRollbackProcessor, instance::setAfterRollbackProcessor)
.acceptIfCondition(this.containerProperties.getAckCount() > 0, this.containerProperties.getAckCount(),

3
spring-kafka/src/test/java/org/springframework/kafka/annotation/ContainerFactoryTests.java
View File

@@ -59,6 +59,9 @@ public class ContainerFactoryTests {
assertThat(container.getContainerProperties().getAckCount()).isEqualTo(123);
assertThat(KafkaTestUtils.getPropertyValue(container, "concurrency", Integer.class)).isEqualTo(22);
assertThat(customized).isTrue();
ConcurrentMessageListenerContainer<String, String> container2 = factory.createContainer("foo");
assertThat(container.getContainerProperties().getKafkaConsumerProperties())
.isNotSameAs(container2.getContainerProperties().getKafkaConsumerProperties());
}
@SuppressWarnings("unchecked")