Address review comments

Make EndpointCustomizerFactory ctor public

Author: tomazfernandes

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

@@ -148,9 +148,8 @@ public KafkaTemplate<String, Object> kafkaTemplate() {
 ----
 ====
 
-IMPORTANT: You can use multiple `@KafkaListener` annotations for the same topic with or without manual partition assignment along with non-blocking retries, but mind that only one configuration will be used for a given topic.
-It's best to use a single `RetryTopicConfiguration` bean for configuration of such topics;
-if you're using multiple `@RetryableTopic` annotations for the same topic, make sure all of them have the same values, otherwise one of them will be applied to all of that topic's listeners and the other annotations' values will be ignored.
+IMPORTANT: Multiple `@KafkaListener` annotations can be used for the same topic with or without manual partition assignment along with non-blocking retries, but only one configuration will be used for a given topic.
+It's best to use a single `RetryTopicConfiguration` bean for configuration of such topics; if multiple `@RetryableTopic` annotations are being used for the same topic, all of them should have the same values, otherwise one of them will be applied to all of that topic's listeners and the other annotations' values will be ignored.
 
 ==== Features
 

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopic.java

@@ -220,9 +220,10 @@ public long delay() {
 		}
 
 		/**
-		 * Returns the number of partitions the
+		 * Return the number of partitions the
 		 * retry topics should be created with.
 		 * @return the number of partitions.
+		 * @since 2.7.13
 		 */
 		public int numPartitions() {
 			return this.numPartitions;

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/EndpointCustomizerFactory.java

@@ -53,7 +53,7 @@ public class EndpointCustomizerFactory {
 
 	private final RetryTopicNamesProviderFactory retryTopicNamesProviderFactory;
 
-	EndpointCustomizerFactory(DestinationTopic.Properties destinationProperties, EndpointHandlerMethod beanMethod,
+	public EndpointCustomizerFactory(DestinationTopic.Properties destinationProperties, EndpointHandlerMethod beanMethod,
 							BeanFactory beanFactory, RetryTopicNamesProviderFactory retryTopicNamesProviderFactory) {
 
 		this.destinationProperties = destinationProperties;
@@ -71,7 +71,7 @@ protected EndpointCustomizer addSuffixesAndMethod(DestinationTopic.Properties pr
 		RetryTopicNamesProviderFactory.RetryTopicNamesProvider namesProvider =
 				this.retryTopicNamesProviderFactory.createRetryTopicNamesProvider(properties);
 		return endpoint -> {
-			Collection<EndpointCustomizer.TopicNamesHolder> topics = getTopicNames(namesProvider, endpoint);
+			Collection<EndpointCustomizer.TopicNamesHolder> topics = customizeAndRegisterTopics(namesProvider, endpoint);
 			endpoint.setId(namesProvider.getEndpointId(endpoint));
 			endpoint.setGroupId(namesProvider.getGroupId(endpoint));
 			if (endpoint.getTopics().isEmpty() && endpoint.getTopicPartitionsToAssign() != null) {
@@ -92,7 +92,7 @@ protected EndpointCustomizer addSuffixesAndMethod(DestinationTopic.Properties pr
 		};
 	}
 
-	private TopicPartitionOffset[] getTopicPartitions(DestinationTopic.Properties properties,
+	private static TopicPartitionOffset[] getTopicPartitions(DestinationTopic.Properties properties,
 													RetryTopicNamesProviderFactory.RetryTopicNamesProvider namesProvider,
 													TopicPartitionOffset[] topicPartitionOffsets) {
 		return Stream.of(topicPartitionOffsets)
@@ -102,7 +102,7 @@ private TopicPartitionOffset[] getTopicPartitions(DestinationTopic.Properties pr
 				.toArray(TopicPartitionOffset[]::new);
 	}
 
-	protected Collection<EndpointCustomizer.TopicNamesHolder> getTopicNames(
+	protected Collection<EndpointCustomizer.TopicNamesHolder> customizeAndRegisterTopics(
 			RetryTopicNamesProviderFactory.RetryTopicNamesProvider namesProvider,
 			MethodKafkaListenerEndpoint<?, ?> endpoint) {
 

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurerTests.java

@@ -46,15 +46,14 @@
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.beans.factory.support.DefaultListableBeanFactory;
 import org.springframework.beans.factory.support.RootBeanDefinition;
-import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerEndpointRegistrar;
 import org.springframework.kafka.config.MethodKafkaListenerEndpoint;
 import org.springframework.kafka.config.MultiMethodKafkaListenerEndpoint;
 import org.springframework.kafka.support.EndpointHandlerMethod;
+import org.springframework.kafka.test.condition.LogLevels;
 import org.springframework.test.util.ReflectionTestUtils;
-import org.springframework.util.Assert;
 
 /**
  * @author Tomaz Fernandes
@@ -358,21 +357,14 @@ void shouldInstantiateIfNotInContainer() {
 
 	}
 
+	@LogLevels(classes = RetryTopicConfigurer.class, level = "info")
 	@Test
 	@SuppressWarnings("deprecation")
 	void shouldLogConsumerRecordMessage() {
 		RetryTopicConfigurer.LoggingDltListenerHandlerMethod method =
 				new RetryTopicConfigurer.LoggingDltListenerHandlerMethod();
 		method.logMessage(consumerRecordMessage);
-		LogAccessor logger = (LogAccessor) ReflectionTestUtils
-				.getField(RetryTopicConfigurer.class, "LOGGER");
-		Assert.notNull(logger, "No LOGGER found in class " + RetryTopicConfigurer.class.getSimpleName());
-		if (logger.isInfoEnabled()) {
-			then(consumerRecordMessage).should().topic();
-		}
-		else {
-			then(consumerRecordMessage).shouldHaveNoInteractions();
-		}
+		then(consumerRecordMessage).should().topic();
 	}
 
 	@Test