GH-2501: Fix Class Tangle

See #2501

New tangle between `container <-> properties`.

Move the new properties and accessors to the abstract container.

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -1749,7 +1749,7 @@ For the `ConcurrentMessageListenerContainer`, the `<beanName>` part of the threa
 So, with a bean name of `container`, threads in this container will be named `container-0-C-1`, `container-1-C-1` etc., after the container is started the first time; `container-0-C-2`, `container-1-C-2` etc., after a stop and subsequent start.
 
 Starting with version `3.0.1`, you can now change the name of the thread, regardless of which executor is used.
-Set the `ContainerProperties.changeConsumerThreadName` property to `true` and the `ContainerProperties.threadNameSupplier` will be invoked to obtain the thread name.
+Set the `AbstractMessageListenerContainer.changeConsumerThreadName` property to `true` and the `AbstractMessageListenerContainer.threadNameSupplier` will be invoked to obtain the thread name.
 This is a `Function<MessageListenerContainer, String>`, with the default implementation returning `container.getListenerId()`.
 
 [[kafka-listener-meta]]

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

@@ -19,6 +19,7 @@
 
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.function.Function;
 import java.util.regex.Pattern;
 
 import org.apache.commons.logging.LogFactory;
@@ -38,6 +39,7 @@
 import org.springframework.kafka.listener.BatchInterceptor;
 import org.springframework.kafka.listener.CommonErrorHandler;
 import org.springframework.kafka.listener.ContainerProperties;
+import org.springframework.kafka.listener.MessageListenerContainer;
 import org.springframework.kafka.listener.RecordInterceptor;
 import org.springframework.kafka.listener.adapter.BatchToRecordAdapter;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
@@ -110,6 +112,10 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 
 	private String correlationHeaderName;
 
+	private Boolean changeConsumerThreadName;
+
+	private Function<MessageListenerContainer, String> threadNameSupplier;
+
 	@Override
 	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
 		this.applicationContext = applicationContext;
@@ -334,6 +340,29 @@ public void setCorrelationHeaderName(String correlationHeaderName) {
 		this.correlationHeaderName = correlationHeaderName;
 	}
 
+	/**
+	 * Set to true to instruct the container to change the consumer thread name during
+	 * initialization.
+	 * @param changeConsumerThreadName true to change.
+	 * @since 3.0.1
+	 * @see #setThreadNameSupplier(Function)
+	 */
+	public void setChangeConsumerThreadName(boolean changeConsumerThreadName) {
+		this.changeConsumerThreadName = changeConsumerThreadName;
+	}
+
+	/**
+	 * Set a function used to change the consumer thread name. The default returns the
+	 * container {@code listenerId}.
+	 * @param threadNameSupplier the function.
+	 * @since 3.0.1
+	 * @see #setChangeConsumerThreadName(boolean)
+	 */
+	public void setThreadNameSupplier(Function<MessageListenerContainer, String> threadNameSupplier) {
+		Assert.notNull(threadNameSupplier, "'threadNameSupplier' cannot be null");
+		this.threadNameSupplier = threadNameSupplier;
+	}
+
 	@SuppressWarnings("deprecation")
 	@Override
 	public void afterPropertiesSet() {
@@ -414,7 +443,9 @@ protected void initializeContainer(C instance, KafkaListenerEndpoint endpoint) {
 						properties::setSubBatchPerPartition)
 				.acceptIfNotNull(this.errorHandler, instance::setGenericErrorHandler)
 				.acceptIfNotNull(this.commonErrorHandler, instance::setCommonErrorHandler)
-				.acceptIfNotNull(this.missingTopicsFatal, instance.getContainerProperties()::setMissingTopicsFatal);
+				.acceptIfNotNull(this.missingTopicsFatal, instance.getContainerProperties()::setMissingTopicsFatal)
+				.acceptIfNotNull(this.changeConsumerThreadName, instance::setChangeConsumerThreadName)
+				.acceptIfNotNull(this.threadNameSupplier, instance::setThreadNameSupplier);
 		Boolean autoStart = endpoint.getAutoStartup();
 		if (autoStart != null) {
 			instance.setAutoStartup(autoStart);

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

@@ -25,6 +25,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
@@ -126,6 +127,11 @@ public abstract class AbstractMessageListenerContainer<K, V>
 	@Nullable
 	private String mainListenerId;
 
+	private boolean changeConsumerThreadName;
+
+	@NonNull
+	private Function<MessageListenerContainer, String> threadNameSupplier = container -> container.getListenerId();
+
 	/**
 	 * Construct an instance with the provided factory and properties.
 	 * @param consumerFactory the factory.
@@ -423,6 +429,48 @@ public void setTopicCheckTimeout(int topicCheckTimeout) {
 		this.topicCheckTimeout = topicCheckTimeout;
 	}
 
+	/**
+	 * Return true if the container should change the consumer thread name during
+	 * initialization.
+	 * @return true to change.
+	 * @since 3.0.1
+	 */
+	public boolean isChangeConsumerThreadName() {
+		return this.changeConsumerThreadName;
+	}
+
+	/**
+	 * Set to true to instruct the container to change the consumer thread name during
+	 * initialization.
+	 * @param changeConsumerThreadName true to change.
+	 * @since 3.0.1
+	 * @see #setThreadNameSupplier(Function)
+	 */
+	public void setChangeConsumerThreadName(boolean changeConsumerThreadName) {
+		this.changeConsumerThreadName = changeConsumerThreadName;
+	}
+
+	/**
+	 * Return the function used to change the consumer thread name.
+	 * @return the function.
+	 * @since 3.0.1
+	 */
+	public Function<MessageListenerContainer, String> getThreadNameSupplier() {
+		return this.threadNameSupplier;
+	}
+
+	/**
+	 * Set a function used to change the consumer thread name. The default returns the
+	 * container {@code listenerId}.
+	 * @param threadNameSupplier the function.
+	 * @since 3.0.1
+	 * @see #setChangeConsumerThreadName(boolean)
+	 */
+	public void setThreadNameSupplier(Function<MessageListenerContainer, String> threadNameSupplier) {
+		Assert.notNull(threadNameSupplier, "'threadNameSupplier' cannot be null");
+		this.threadNameSupplier = threadNameSupplier;
+	}
+
 	protected RecordInterceptor<K, V> getRecordInterceptor() {
 		return this.recordInterceptor;
 	}

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

@@ -23,7 +23,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.function.Function;
 import java.util.regex.Pattern;
 
 import org.aopalliance.aop.Advice;
@@ -34,7 +33,6 @@
 import org.springframework.core.task.AsyncTaskExecutor;
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.kafka.support.micrometer.KafkaListenerObservationConvention;
-import org.springframework.lang.NonNull;
 import org.springframework.lang.Nullable;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.transaction.PlatformTransactionManager;
@@ -289,11 +287,6 @@ public enum EOSMode {
 
 	private KafkaListenerObservationConvention observationConvention;
 
-	private boolean changeConsumerThreadName;
-
-	@NonNull
-	private Function<MessageListenerContainer, String> threadNameSupplier = container -> container.getListenerId();
-
 	/**
 	 * Create properties for a container that will subscribe to the specified topics.
 	 * @param topics the topics.
@@ -952,48 +945,6 @@ public void setObservationConvention(KafkaListenerObservationConvention observat
 		this.observationConvention = observationConvention;
 	}
 
-	/**
-	 * Return true if the container should change the consumer thread name during
-	 * initialization.
-	 * @return true to change.
-	 * @since 3.0.1
-	 */
-	public boolean isChangeConsumerThreadName() {
-		return this.changeConsumerThreadName;
-	}
-
-	/**
-	 * Set to true to instruct the container to change the consumer thread name during
-	 * initialization.
-	 * @param changeConsumerThreadName true to change.
-	 * @since 3.0.1
-	 * @see #setThreadNameSupplier(Function)
-	 */
-	public void setChangeConsumerThreadName(boolean changeConsumerThreadName) {
-		this.changeConsumerThreadName = changeConsumerThreadName;
-	}
-
-	/**
-	 * Return the function used to change the consumer thread name.
-	 * @return the function.
-	 * @since 3.0.1
-	 */
-	public Function<MessageListenerContainer, String> getThreadNameSupplier() {
-		return this.threadNameSupplier;
-	}
-
-	/**
-	 * Set a function used to change the consumer thread name. The default returns the
-	 * container {@code listenerId}.
-	 * @param threadNameSupplier the function.
-	 * @since 3.0.1
-	 * @see #setChangeConsumerThreadName(boolean)
-	 */
-	public void setThreadNameSupplier(Function<MessageListenerContainer, String> threadNameSupplier) {
-		Assert.notNull(threadNameSupplier, "'threadNameSupplier' cannot be null");
-		this.threadNameSupplier = threadNameSupplier;
-	}
-
 	@Override
 	public String toString() {
 		return "ContainerProperties ["
@@ -1030,7 +981,6 @@ public String toString() {
 				+ (this.observationConvention != null
 						? "\n observationConvention=" + this.observationConvention
 						: "")
-				+ "\n changeConsumerThreadName=" + this.changeConsumerThreadName
 				+ "\n]";
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -1397,9 +1397,10 @@ public void run() {
 		}
 
 		protected void initialize() {
-			if (this.containerProperties.isChangeConsumerThreadName()) {
+			if (KafkaMessageListenerContainer.this.thisOrParentContainer.isChangeConsumerThreadName()) {
 				Thread.currentThread().setName(
-						this.containerProperties.getThreadNameSupplier().apply(KafkaMessageListenerContainer.this));
+						KafkaMessageListenerContainer.this.thisOrParentContainer.getThreadNameSupplier()
+								.apply(KafkaMessageListenerContainer.this));
 			}
 			publishConsumerStartingEvent();
 			this.consumerThread = Thread.currentThread();

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -920,6 +920,7 @@ public void testProjection() throws InterruptedException {
 		assertThat(this.listener.projectionLatch.await(60, TimeUnit.SECONDS)).isTrue();
 		assertThat(this.listener.name).isEqualTo("SomeName");
 		assertThat(this.listener.username).isEqualTo("SomeUsername");
+		assertThat(this.listener.customThreadName).isEqualTo("foo.projection-0");
 	}
 
 	@SuppressWarnings("unchecked")
@@ -1167,6 +1168,8 @@ public KafkaListenerContainerFactory<?> projectionListenerContainerFactory() {
 			typeMapper.addTrustedPackages("*");
 			converter.setTypeMapper(typeMapper);
 			factory.setMessageConverter(new ProjectingMessageConverter(converter));
+			factory.setChangeConsumerThreadName(true);
+			factory.setThreadNameSupplier(container -> "foo." + container.getListenerId());
 			return factory;
 		}
 
@@ -1888,6 +1891,8 @@ static class Listener implements ConsumerSeekAware {
 
 		volatile String batchOverrideStackTrace;
 
+		volatile String customThreadName;
+
 		@KafkaListener(id = "manualStart", topics = "manualStart",
 				containerFactory = "kafkaAutoStartFalseListenerContainerFactory",
 				properties = { ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG + ":301000",
@@ -2206,6 +2211,7 @@ public void projectionListener(ProjectionSample sample) {
 			this.username = sample.getUsername();
 			this.name = sample.getName();
 			this.projectionLatch.countDown();
+			this.customThreadName = Thread.currentThread().getName();
 		}
 
 		@KafkaListener(id = "customMethodArgumentResolver", topics = "annotated39")

spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java

@@ -137,7 +137,6 @@ protected Consumer<Integer, String> createKafkaConsumer(String groupId, String c
 		ContainerProperties containerProps = new ContainerProperties(topic1);
 		containerProps.setLogContainerConfig(true);
 		containerProps.setClientId("client");
-		containerProps.setChangeConsumerThreadName(true);
 
 		final CountDownLatch latch = new CountDownLatch(3);
 		final Set<String> listenerThreadNames = new ConcurrentSkipListSet<>();
@@ -153,6 +152,7 @@ protected Consumer<Integer, String> createKafkaConsumer(String groupId, String c
 				new ConcurrentMessageListenerContainer<>(cf, containerProps);
 		container.setConcurrency(2);
 		container.setBeanName("testAuto");
+		container.setChangeConsumerThreadName(true);
 		BlockingQueue<KafkaEvent> events = new LinkedBlockingQueue<>();
 		CountDownLatch stopLatch = new CountDownLatch(4);
 		container.setApplicationEventPublisher(e -> {