GH-1745 - Let users provide a RetryTopicConfigurer bean

Author: tomazfernandes

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

@@ -51,13 +51,14 @@
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.beans.factory.ObjectFactory;
 import org.springframework.beans.factory.SmartInitializingSingleton;
-import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
 import org.springframework.beans.factory.config.BeanExpressionContext;
 import org.springframework.beans.factory.config.BeanExpressionResolver;
 import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.beans.factory.config.Scope;
+import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ConfigurableApplicationContext;
@@ -84,6 +85,7 @@
 import org.springframework.kafka.retrytopic.RetryTopicBootstrapper;
 import org.springframework.kafka.retrytopic.RetryTopicConfiguration;
 import org.springframework.kafka.retrytopic.RetryTopicConfigurer;
+import org.springframework.kafka.retrytopic.RetryTopicInternalBeanNames;
 import org.springframework.kafka.support.KafkaNull;
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.lang.Nullable;
@@ -481,19 +483,25 @@ private boolean processMainAndRetryListeners(KafkaListener kafkaListener, Object
 	}
 
 	private RetryTopicConfigurer getRetryTopicConfigurer() {
-		try {
-			return this.beanFactory.getBean(RetryTopicConfigurer.class);
-		}
-		catch (NoSuchBeanDefinitionException e) {
-			if (!(this.beanFactory instanceof AutowireCapableBeanFactory)) {
-				throw new IllegalStateException("BeanFactory must be an instance of "
-						+ AutowireCapableBeanFactory.class.getSimpleName()
-						+ " Provided beanFactory: " + this.beanFactory.getClass().getSimpleName(), e);
-			}
-			((AutowireCapableBeanFactory) this.beanFactory)
-					.createBean(RetryTopicBootstrapper.class)
-					.bootstrapRetryTopic();
-			return this.beanFactory.getBean(RetryTopicConfigurer.class);
+		bootstrapRetryTopicIfNecessary();
+		return this.beanFactory.getBean(RetryTopicInternalBeanNames.RETRY_TOPIC_CONFIGURER, RetryTopicConfigurer.class);
+	}
+
+	private void bootstrapRetryTopicIfNecessary() {
+		if (!(this.beanFactory instanceof BeanDefinitionRegistry)) {
+			throw new IllegalStateException("BeanFactory must be an instance of "
+					+ BeanDefinitionRegistry.class.getSimpleName()
+					+ " to bootstrap the RetryTopic functionality. Provided beanFactory: "
+					+ this.beanFactory.getClass().getSimpleName());
+		}
+		BeanDefinitionRegistry registry = (BeanDefinitionRegistry) this.beanFactory;
+		if (!registry.containsBeanDefinition(RetryTopicInternalBeanNames
+				.RETRY_TOPIC_BOOTSTRAPPER)) {
+			registry.registerBeanDefinition(RetryTopicInternalBeanNames
+							.RETRY_TOPIC_BOOTSTRAPPER,
+					new RootBeanDefinition(RetryTopicBootstrapper.class));
+			this.beanFactory.getBean(RetryTopicInternalBeanNames
+					.RETRY_TOPIC_BOOTSTRAPPER, RetryTopicBootstrapper.class).bootstrapRetryTopic();
 		}
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -33,8 +33,6 @@ public abstract class RetryTopicInternalBeanNames {
 
 	static final String KAFKA_CONSUMER_BACKOFF_MANAGER = "internalKafkaConsumerBackoffManager";
 
-	static final String RETRY_TOPIC_CONFIGURER = "internalRetryTopicConfigurer";
-
 	static final String LISTENER_CONTAINER_FACTORY_RESOLVER_NAME = "internalListenerContainerFactoryResolver";
 
 	static final String LISTENER_CONTAINER_FACTORY_CONFIGURER_NAME = "internalListenerContainerFactoryConfigurer";
@@ -65,4 +63,14 @@ public abstract class RetryTopicInternalBeanNames {
 	 */
 	public static final String DEFAULT_KAFKA_TEMPLATE_BEAN_NAME = "retryTopicDefaultKafkaTemplate";
 
+	/**
+	 * RetryTopicBootstrapper bean name.
+	 */
+	public static final String RETRY_TOPIC_BOOTSTRAPPER = "internalRetryTopicBootstrapper";
+
+	/**
+	 * RetryTopicConfigurer bean name.
+	 */
+	public static final String RETRY_TOPIC_CONFIGURER = "internalRetryTopicConfigurer";
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.