From 9cb57637947b79d9e5fb812eb4e62851c49054de Mon Sep 17 00:00:00 2001
From: Moritz Halbritter <mkammerer@vmware.com>
Date: Wed, 26 Jul 2023 13:45:10 +0200
Subject: [PATCH] Add property to set changeConsumerThreadName for Kafka

Closes gh-36343
---
 ...entKafkaListenerContainerFactoryConfigurer.java |  1 +
 .../boot/autoconfigure/kafka/KafkaProperties.java  | 14 ++++++++++++++
 ...fkaListenerContainerFactoryConfigurerTests.java | 12 +++++++++++-
 3 files changed, 26 insertions(+), 1 deletion(-)

diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java
index e842936da0..edd8f4a37c 100644
--- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java
+++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java
@@ -199,6 +199,7 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer {
 		map.from(this.recordInterceptor).to(factory::setRecordInterceptor);
 		map.from(this.batchInterceptor).to(factory::setBatchInterceptor);
 		map.from(this.threadNameSupplier).to(factory::setThreadNameSupplier);
+		map.from(properties::getChangeConsumerThreadName).to(factory::setChangeConsumerThreadName);
 	}
 
 	private void configureContainer(ContainerProperties container) {
diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java
index 50a2890821..c4b2f55734 100644
--- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java
+++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java
@@ -1043,6 +1043,12 @@ public class KafkaProperties {
 		 */
 		private boolean autoStartup = true;
 
+		/**
+		 * Whether to instruct the container to change the consumer thread name during
+		 * initialization.
+		 */
+		private Boolean changeConsumerThreadName;
+
 		public Type getType() {
 			return this.type;
 		}
@@ -1179,6 +1185,14 @@ public class KafkaProperties {
 			this.autoStartup = autoStartup;
 		}
 
+		public Boolean getChangeConsumerThreadName() {
+			return this.changeConsumerThreadName;
+		}
+
+		public void setChangeConsumerThreadName(Boolean changeConsumerThreadName) {
+			this.changeConsumerThreadName = changeConsumerThreadName;
+		}
+
 	}
 
 	public static class Ssl {
diff --git a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurerTests.java b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurerTests.java
index 2bac853b2e..b82a6efcab 100644
--- a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurerTests.java
+++ b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurerTests.java
@@ -42,11 +42,14 @@ class ConcurrentKafkaListenerContainerFactoryConfigurerTests {
 
 	private ConsumerFactory<Object, Object> consumerFactory;
 
+	private KafkaProperties properties;
+
 	@BeforeEach
 	@SuppressWarnings("unchecked")
 	void setUp() {
 		this.configurer = new ConcurrentKafkaListenerContainerFactoryConfigurer();
-		this.configurer.setKafkaProperties(new KafkaProperties());
+		this.properties = new KafkaProperties();
+		this.configurer.setKafkaProperties(this.properties);
 		this.factory = spy(new ConcurrentKafkaListenerContainerFactory<>());
 		this.consumerFactory = mock(ConsumerFactory.class);
 
@@ -60,4 +63,11 @@ class ConcurrentKafkaListenerContainerFactoryConfigurerTests {
 		then(this.factory).should().setThreadNameSupplier(function);
 	}
 
+	@Test
+	void shouldApplyChangeConsumerThreadName() {
+		this.properties.getListener().setChangeConsumerThreadName(true);
+		this.configurer.configure(this.factory, this.consumerFactory);
+		then(this.factory).should().setChangeConsumerThreadName(true);
+	}
+
 }