From 6484a2f63da2749094f3d328d8a2c4b4514d169a Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Fri, 21 May 2021 08:27:14 -0400
Subject: [PATCH] GH-1808: Fix Container Properties Crosstalk

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**
---
 .../kafka/config/AbstractKafkaListenerContainerFactory.java    | 2 +-
 .../kafka/annotation/ContainerFactoryTests.java                | 3 +++
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java
index 585ebb14..5d4e24d1 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java
@@ -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(),
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/ContainerFactoryTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/ContainerFactoryTests.java
index b3b6b815..0c6dc77f 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/ContainerFactoryTests.java
+++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/ContainerFactoryTests.java
@@ -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")