Add property to set changeConsumerThreadName for Kafka

Closes gh-36343

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) {

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 {

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);
+	}
+
 }