GH-1715,GH-1717: GH-920 Polishing

Fixes GH-1715.

Fixes GH-1717.

Remove leading space in javadoRemove leading space in javadocc

Author: tomazfernandes

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -411,7 +411,8 @@ private boolean processMainAndRetryListeners(KafkaListener kafkaListener, Object
 				resolveContainerFactory(kafkaListener, resolve(kafkaListener.containerFactory()), beanName);
 
 		getRetryTopicConfigurer()
-				.processMainAndRetryListeners(endpointProcessor, endpoint, retryTopicConfiguration, this.registrar, factory);
+				.processMainAndRetryListeners(endpointProcessor, endpoint, retryTopicConfiguration,
+						this.registrar, factory, this.defaultContainerFactoryBeanName);
 
 		this.listenerScope.removeListener(beanRef);
 		return true;

spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopicAnnotationProcessor.java

@@ -60,6 +60,8 @@ public class RetryableTopicAnnotationProcessor {
 		PARSER = new SpelExpressionParser();
 	}
 
+	private final static String DEFAULT_SPRING_BOOT_KAFKA_TEMPLATE_NAME = "kafkaTemplate";
+
 	public RetryableTopicAnnotationProcessor(BeanFactory beanFactory) {
 		this.beanFactory = beanFactory;
 	}
@@ -163,7 +165,12 @@ private RetryTopicConfigurer.EndpointHandlerMethod getDltProcessor(Method listen
 			return this.beanFactory.getBean(RetryTopicInternalBeanNames.DEFAULT_KAFKA_TEMPLATE_BEAN_NAME, KafkaOperations.class);
 		}
 		catch (NoSuchBeanDefinitionException ex) {
-			throw new BeanInitializationException("Could not find a KafkaTemplate to configure the retry topics.", ex);
+			try {
+				return this.beanFactory.getBean(DEFAULT_SPRING_BOOT_KAFKA_TEMPLATE_NAME, KafkaOperations.class);
+			}
+			catch (NoSuchBeanDefinitionException exc) {
+				throw new BeanInitializationException("Could not find a KafkaTemplate to configure the retry topics.", exc);
+			}
 		}
 	}
 }

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

@@ -190,6 +190,10 @@ public int hashCode() {
 			return Objects.hash(this.delayMs, this.suffix, this.type, this.maxAttempts, this.numPartitions,
 					this.dltStrategy, this.kafkaOperations);
 		}
+
+		public boolean isMainEndpoint() {
+			return Type.MAIN.equals(this.type);
+		}
 	}
 
 	enum Type {

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

@@ -88,8 +88,12 @@ public List<DestinationTopic.Properties> createProperties() {
 	}
 
 	private List<DestinationTopic.Properties> createPropertiesForFixedDelaySingleTopic() {
-		return Arrays.asList(createMainTopicProperties(), createRetryProperties(1,
-				DestinationTopic.Type.SINGLE_TOPIC_RETRY, getShouldRetryOn()), createDltProperties());
+		return isNoDltStrategy()
+					? Arrays.asList(createMainTopicProperties(),
+							createRetryProperties(1, DestinationTopic.Type.SINGLE_TOPIC_RETRY, getShouldRetryOn()))
+					: Arrays.asList(createMainTopicProperties(),
+							createRetryProperties(1, DestinationTopic.Type.SINGLE_TOPIC_RETRY, getShouldRetryOn()),
+							createDltProperties());
 	}
 
 	private boolean isSingleTopicFixedDelay() {
@@ -101,26 +105,29 @@ private boolean isSingleTopicStrategy() {
 	}
 
 	private List<DestinationTopic.Properties> createPropertiesForDefaultTopicStrategy() {
-		return IntStream.rangeClosed(0, DltStrategy.NO_DLT.equals(this.dltStrategy)
+		return IntStream.rangeClosed(0, isNoDltStrategy()
 							? this.maxAttempts - 1
 							: this.maxAttempts)
-				.mapToObj(index -> createRetryOrDltTopicSuffixes(index))
+				.mapToObj(this::createRetryOrDltTopicSuffixes)
 				.collect(Collectors.toList());
 	}
 
-	private DestinationTopic.Properties createMainTopicProperties() {
-		return new DestinationTopic.Properties(0, MAIN_TOPIC_SUFFIX, DestinationTopic.Type.MAIN, this.maxAttempts,
-				this.numPartitions, this.dltStrategy, this.kafkaOperations, getShouldRetryOn(), this.timeout);
+	private boolean isNoDltStrategy() {
+		return DltStrategy.NO_DLT.equals(this.dltStrategy);
 	}
 
 	private DestinationTopic.Properties createRetryOrDltTopicSuffixes(int index) {
 		BiPredicate<Integer, Throwable> shouldRetryOn = getShouldRetryOn();
 		return index == 0
 				? createMainTopicProperties()
 				: index < this.maxAttempts
-				? createRetryProperties(index,
-				DestinationTopic.Type.RETRY, shouldRetryOn)
-				: createDltProperties();
+					? createRetryProperties(index, DestinationTopic.Type.RETRY, shouldRetryOn)
+					: createDltProperties();
+	}
+
+	private DestinationTopic.Properties createMainTopicProperties() {
+		return new DestinationTopic.Properties(0, MAIN_TOPIC_SUFFIX, DestinationTopic.Type.MAIN, this.maxAttempts,
+				this.numPartitions, this.dltStrategy, this.kafkaOperations, getShouldRetryOn(), this.timeout);
 	}
 
 	private DestinationTopic.Properties createDltProperties() {

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

@@ -59,17 +59,15 @@ public class ListenerContainerFactoryResolver {
 		this.retryEndpointCache = new Cache();
 
 		this.mainEndpointResolvers = Arrays.asList(
-				(fromKafkaListenerAnnotation, configuration) -> this.mainEndpointCache
-						.fromCache(fromKafkaListenerAnnotation, configuration),
+				this.mainEndpointCache::fromCache,
 				(fromKafkaListenerAnnotation, configuration) -> fromKafkaListenerAnnotation,
 				(fromKLAnnotation, configuration) -> configuration.factoryFromRetryTopicConfiguration,
 				(fromKLAnnotation, configuration) -> fromBeanName(configuration.listenerContainerFactoryName),
 				(fromKLAnnotation, configuration) ->
 						fromBeanName(RetryTopicInternalBeanNames.DEFAULT_LISTENER_FACTORY_BEAN_NAME));
 
 		this.retryEndpointResolvers = Arrays.asList(
-				(fromKafkaListenerAnnotation, configuration) ->
-						this.retryEndpointCache.fromCache(fromKafkaListenerAnnotation, configuration),
+				this.retryEndpointCache::fromCache,
 				(fromKLAnnotation, configuration) -> configuration.factoryFromRetryTopicConfiguration,
 				(fromKLAnnotation, configuration) -> fromBeanName(configuration.listenerContainerFactoryName),
 				(fromKLAnnotation, configuration) -> fromKLAnnotation,
@@ -78,19 +76,38 @@ public class ListenerContainerFactoryResolver {
 	}
 
 	ConcurrentKafkaListenerContainerFactory<?, ?> resolveFactoryForMainEndpoint(
-			KafkaListenerContainerFactory<?> factoryFromKafkaListenerAnnotation, Configuration config) {
+			@Nullable KafkaListenerContainerFactory<?> factoryFromKafkaListenerAnnotationInstance,
+			String defaultContainerFactoryBeanName,
+			Configuration config) {
+		KafkaListenerContainerFactory<?> factoryFromKafkaListenerAnnotation =
+				getFactoryFromKLA(factoryFromKafkaListenerAnnotationInstance, defaultContainerFactoryBeanName);
 		ConcurrentKafkaListenerContainerFactory<?, ?> resolvedFactory = resolveFactory(this.mainEndpointResolvers,
 				factoryFromKafkaListenerAnnotation, config);
 		return this.mainEndpointCache.addIfAbsent(factoryFromKafkaListenerAnnotation, config, resolvedFactory);
 	}
 
 	ConcurrentKafkaListenerContainerFactory<?, ?> resolveFactoryForRetryEndpoint(
-			KafkaListenerContainerFactory<?> factoryFromKafkaListenerAnnotation, Configuration config) {
+			@Nullable KafkaListenerContainerFactory<?> factoryFromKafkaListenerAnnotationInstance,
+			String defaultContainerFactoryBeanName,
+			Configuration config) {
+		KafkaListenerContainerFactory<?> factoryFromKafkaListenerAnnotation =
+				getFactoryFromKLA(factoryFromKafkaListenerAnnotationInstance, defaultContainerFactoryBeanName);
 		ConcurrentKafkaListenerContainerFactory<?, ?> resolvedFactory = resolveFactory(this.retryEndpointResolvers,
 				factoryFromKafkaListenerAnnotation, config);
 		return this.retryEndpointCache.addIfAbsent(factoryFromKafkaListenerAnnotation, config, resolvedFactory);
 	}
 
+	@Nullable
+	private KafkaListenerContainerFactory<?> getFactoryFromKLA(KafkaListenerContainerFactory<?> factoryFromKafkaListenerAnnotationInstance,
+															String defaultContainerFactoryBeanName) {
+		KafkaListenerContainerFactory<?> factoryFromKafkaListenerAnnotation =
+				factoryFromKafkaListenerAnnotationInstance;
+		if (factoryFromKafkaListenerAnnotation == null) {
+			factoryFromKafkaListenerAnnotation = fromBeanName(defaultContainerFactoryBeanName);
+		}
+		return factoryFromKafkaListenerAnnotation;
+	}
+
 	private ConcurrentKafkaListenerContainerFactory<?, ?> resolveFactory(List<FactoryResolver> factoryResolvers,
 			KafkaListenerContainerFactory<?> factoryFromKafkaListenerAnnotation,
 			Configuration config) {

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

@@ -239,72 +239,71 @@ public RetryTopicConfigurer(DestinationTopicProcessor destinationTopicProcessor,
 	/**
 	 * Entrypoint for creating and configuring the retry and dlt endpoints, as well as the
 	 * container factory that will create the corresponding listenerContainer.
-	 *
 	 * @param endpointProcessor function that will process the endpoints
 	 * processListener method.
 	 * @param mainEndpoint the endpoint based on which retry and dlt endpoints are also
 	 * created and processed.
 	 * @param configuration the configuration for the topic.
-	 * @param registrar The {@link KafkaListenerEndpointRegistrar} that will register the
-	 * endpoints.
+	 * @param registrar The {@link KafkaListenerEndpointRegistrar} that will register the endpoints.
 	 * @param factory The factory provided in the {@link org.springframework.kafka.annotation.KafkaListener}
-	 * annotation, if any.
+	 * @param defaultContainerFactoryBeanName The default factory bean name for the
+	 * {@link org.springframework.kafka.annotation.KafkaListener}
 	 *
 	 */
 	public void processMainAndRetryListeners(EndpointProcessor endpointProcessor,
 											MethodKafkaListenerEndpoint<?, ?> mainEndpoint,
 											RetryTopicConfiguration configuration,
 											KafkaListenerEndpointRegistrar registrar,
-											@Nullable KafkaListenerContainerFactory<?> factory) {
+											@Nullable KafkaListenerContainerFactory<?> factory,
+											String defaultContainerFactoryBeanName) {
 		throwIfMultiMethodEndpoint(mainEndpoint);
-		configureMainEndpoint(mainEndpoint, endpointProcessor, registrar, factory, configuration);
 		DestinationTopicProcessor.Context context =
 				new DestinationTopicProcessor.Context(configuration.getDestinationTopicProperties());
-		configureRetryAndDltEndpoints(mainEndpoint, endpointProcessor, factory, registrar, configuration, context);
+		configureEndpoints(mainEndpoint, endpointProcessor, factory, registrar, configuration, context,
+				defaultContainerFactoryBeanName);
 		this.destinationTopicProcessor.processRegisteredDestinations(getTopicCreationFunction(configuration), context);
 	}
 
-	private void configureMainEndpoint(MethodKafkaListenerEndpoint<?, ?> mainEndpoint,
+	private void configureEndpoints(MethodKafkaListenerEndpoint<?, ?> mainEndpoint,
 									EndpointProcessor endpointProcessor,
-									KafkaListenerEndpointRegistrar registrar,
 									KafkaListenerContainerFactory<?> factory,
-									RetryTopicConfiguration configuration) {
-		ConcurrentKafkaListenerContainerFactory<?, ?> resolvedFactory =
-				resolveAndConfigureFactoryForMainEndpoint(factory, configuration);
-		endpointProcessor.process(mainEndpoint);
-		registrar.registerEndpoint(mainEndpoint, resolvedFactory);
-		mainEndpoint.setBeanFactory(this.beanFactory);
-	}
-
-	private void configureRetryAndDltEndpoints(MethodKafkaListenerEndpoint<?, ?> mainEndpoint,
-											EndpointProcessor endpointProcessor,
-											KafkaListenerContainerFactory<?> factory,
-											KafkaListenerEndpointRegistrar registrar,
-											RetryTopicConfiguration configuration,
-											DestinationTopicProcessor.Context context) {
-		ConcurrentKafkaListenerContainerFactory<?, ?> resolvedFactory =
-				resolveAndConfigureFactoryForRetryEndpoint(factory, configuration);
-
+									KafkaListenerEndpointRegistrar registrar,
+									RetryTopicConfiguration configuration,
+									DestinationTopicProcessor.Context context,
+									String defaultContainerFactoryBeanName) {
 		this.destinationTopicProcessor
 				.processDestinationTopicProperties(destinationTopicProperties ->
-						processAndRegisterRetryDltDestination(mainEndpoint,
+						processAndRegisterEndpoints(mainEndpoint,
 								endpointProcessor,
-								resolvedFactory,
+								factory,
+								defaultContainerFactoryBeanName,
 								registrar,
 								configuration,
 								context,
 								destinationTopicProperties),
 						context);
 	}
 
-	private void processAndRegisterRetryDltDestination(MethodKafkaListenerEndpoint<?, ?> mainEndpoint, EndpointProcessor endpointProcessor,
-													KafkaListenerContainerFactory<?> resolvedFactory, KafkaListenerEndpointRegistrar registrar,
-													RetryTopicConfiguration configuration, DestinationTopicProcessor.Context context,
-													DestinationTopic.Properties destinationTopicProperties) {
-		MethodKafkaListenerEndpoint<Object, Object> endpoint = new MethodKafkaListenerEndpoint<>();
+	private void processAndRegisterEndpoints(MethodKafkaListenerEndpoint<?, ?> mainEndpoint, EndpointProcessor endpointProcessor,
+											KafkaListenerContainerFactory<?> factory,
+											String defaultFactoryBeanName,
+											KafkaListenerEndpointRegistrar registrar,
+											RetryTopicConfiguration configuration, DestinationTopicProcessor.Context context,
+											DestinationTopic.Properties destinationTopicProperties) {
+
+		ConcurrentKafkaListenerContainerFactory<?, ?> resolvedFactory =
+				destinationTopicProperties.isMainEndpoint()
+						? resolveAndConfigureFactoryForMainEndpoint(factory, defaultFactoryBeanName, configuration)
+						: resolveAndConfigureFactoryForRetryEndpoint(factory, defaultFactoryBeanName, configuration);
+
+		MethodKafkaListenerEndpoint<?, ?> endpoint = destinationTopicProperties.isMainEndpoint()
+				? mainEndpoint
+				: new MethodKafkaListenerEndpoint<>();
+
 		endpointProcessor.accept(endpoint);
 
-		EndpointHandlerMethod endpointBeanMethod = getEndpointHandlerMethod(mainEndpoint, configuration, destinationTopicProperties);
+		EndpointHandlerMethod endpointBeanMethod =
+				getEndpointHandlerMethod(mainEndpoint, configuration, destinationTopicProperties);
 
 		createEndpointCustomizer(endpointBeanMethod, destinationTopicProperties)
 						.customizeEndpointAndCollectTopics(endpoint)
@@ -319,7 +318,8 @@ private void processAndRegisterRetryDltDestination(MethodKafkaListenerEndpoint<?
 	}
 
 	private EndpointHandlerMethod getEndpointHandlerMethod(MethodKafkaListenerEndpoint<?, ?> mainEndpoint,
-														RetryTopicConfiguration configuration, DestinationTopic.Properties props) {
+														RetryTopicConfiguration configuration,
+														DestinationTopic.Properties props) {
 		EndpointHandlerMethod dltHandlerMethod = configuration.getDltHandlerMethod();
 		EndpointHandlerMethod retryBeanMethod = new EndpointHandlerMethod(mainEndpoint.getBean(), mainEndpoint.getMethod());
 		return props.isDltTopic() ? getDltEndpointHandlerMethodOrDefault(dltHandlerMethod) : retryBeanMethod;
@@ -355,17 +355,20 @@ private EndpointHandlerMethod getDltEndpointHandlerMethodOrDefault(EndpointHandl
 
 	private ConcurrentKafkaListenerContainerFactory<?, ?> resolveAndConfigureFactoryForMainEndpoint(
 			KafkaListenerContainerFactory<?> providedFactory,
-			RetryTopicConfiguration configuration) {
+			String defaultFactoryBeanName, RetryTopicConfiguration configuration) {
 
 		return this.listenerContainerFactoryConfigurer
 				.configure(this.containerFactoryResolver.resolveFactoryForMainEndpoint(providedFactory,
+						defaultFactoryBeanName,
 						configuration.forContainerFactoryResolver()));
 	}
 
 	private ConcurrentKafkaListenerContainerFactory<?, ?> resolveAndConfigureFactoryForRetryEndpoint(KafkaListenerContainerFactory<?> providedFactory,
+																									String defaultFactoryBeanName,
 																									RetryTopicConfiguration configuration) {
 		return this.listenerContainerFactoryConfigurer
 				.configure(this.containerFactoryResolver.resolveFactoryForRetryEndpoint(providedFactory,
+						defaultFactoryBeanName,
 						configuration.forContainerFactoryResolver()));
 	}
 
@@ -435,7 +438,8 @@ private EndpointCustomizer addSuffixesAndMethod(String topicSuffix, Object bean,
 			};
 		}
 
-		private Collection<TopicNamesHolder> customizeAndRegisterTopics(Suffixer suffixer, MethodKafkaListenerEndpoint<?, ?> endpoint) {
+		private Collection<TopicNamesHolder> customizeAndRegisterTopics(Suffixer suffixer,
+																		MethodKafkaListenerEndpoint<?, ?> endpoint) {
 			return getTopics(endpoint)
 					.stream()
 					.map(topic -> new TopicNamesHolder(topic, suffixer.maybeAddTo(topic)))
@@ -494,7 +498,8 @@ public EndpointHandlerMethod(Class<?> beanClass, String methodName) {
 			this.method = Arrays.stream(ReflectionUtils.getDeclaredMethods(beanClass))
 					.filter(mthd -> mthd.getName().equals(methodName))
 					.findFirst()
-					.orElseThrow(() -> new IllegalArgumentException(String.format("No method %s in class %s", methodName, beanClass)));
+					.orElseThrow(() -> new IllegalArgumentException(
+							String.format("No method %s in class %s", methodName, beanClass)));
 			this.beanClass = beanClass;
 		}
 
@@ -528,13 +533,12 @@ static class LoggingDltListenerHandlerMethod {
 
 		public void logMessage(Object message) {
 			if (message instanceof ConsumerRecord) {
-				LOGGER.info(() -> "Received message in dlt listener: " + ListenerUtils.recordToString((ConsumerRecord<?, ?>) message));
+				LOGGER.info(() -> "Received message in dlt listener: "
+						+ ListenerUtils.recordToString((ConsumerRecord<?, ?>) message));
 			}
 			else {
 				LOGGER.info(() -> "Received message in dlt listener.");
 			}
 		}
 	}
 }
-
-