Remove legacy code.

Also fix unrelated race in EKIT.

Only allow one `RetryTemplateConfigurationSupport` bean.

Author: garyrussell

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

@@ -58,8 +58,6 @@
 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;
@@ -510,29 +508,11 @@ private boolean processMainAndRetryListeners(KafkaListener kafkaListener, Object
 	}
 
 	private RetryTopicConfigurer getRetryTopicConfigurer() {
-		bootstrapRetryTopicIfNecessary();
 		return this.beanFactory.containsBean("internalRetryTopicConfigurer")
 				? this.beanFactory.getBean("internalRetryTopicConfigurer", RetryTopicConfigurer.class)
 				: this.beanFactory.getBean(RetryTopicBeanNames.RETRY_TOPIC_CONFIGURER_BEAN_NAME, RetryTopicConfigurer.class);
 	}
 
-	@SuppressWarnings("deprecation")
-	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("internalRetryTopicBootstrapper")) {
-			registry.registerBeanDefinition("internalRetryTopicBootstrapper",
-					new RootBeanDefinition(org.springframework.kafka.retrytopic.RetryTopicBootstrapper.class));
-			this.beanFactory.getBean("internalRetryTopicBootstrapper",
-					org.springframework.kafka.retrytopic.RetryTopicBootstrapper.class).bootstrapRetryTopic();
-		}
-	}
-
 	private Method checkProxy(Method methodArg, Object bean) {
 		Method method = methodArg;
 		if (AopUtils.isJdkDynamicProxy(bean)) {

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

@@ -25,6 +25,7 @@
  * Base class for {@link KafkaBackOffManagerFactory} implementations.
  *
  * @author Tomaz Fernandes
+ * @author Gary Russell
  * @since 2.7
  * @see KafkaConsumerBackoffManager
  */
@@ -35,24 +36,23 @@ public abstract class AbstractKafkaBackOffManagerFactory
 
 	private ListenerContainerRegistry listenerContainerRegistry;
 
+	/**
+	 * Creates an instance that will retrieve the {@link ListenerContainerRegistry} from
+	 * the {@link ApplicationContext}.
+	 */
+	public AbstractKafkaBackOffManagerFactory() {
+		this.listenerContainerRegistry = null;
+	}
+
 	/**
 	 * Creates an instance with the provided {@link ListenerContainerRegistry},
 	 * which will be used to fetch the {@link MessageListenerContainer} to back off.
-
 	 * @param listenerContainerRegistry the listenerContainerRegistry to use.
 	 */
 	public AbstractKafkaBackOffManagerFactory(ListenerContainerRegistry listenerContainerRegistry) {
 		this.listenerContainerRegistry = listenerContainerRegistry;
 	}
 
-	/**
-	 * Creates an instance that will retrieve the {@link ListenerContainerRegistry} from
-	 * the {@link ApplicationContext}.
-	 */
-	public AbstractKafkaBackOffManagerFactory() {
-		this.listenerContainerRegistry = null;
-	}
-
 	/**
 	 * Sets the {@link ListenerContainerRegistry}, that will be used to fetch the
 	 * {@link MessageListenerContainer} to back off.

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

@@ -45,8 +45,8 @@ public class ContainerPartitionPausingBackOffManager implements KafkaConsumerBac
 
 	/**
 	 * Construct an instance with the provided registry and back off handler.
-	 * @param listenerContainerRegistry
-	 * @param backOffHandler
+	 * @param listenerContainerRegistry the registry.
+	 * @param backOffHandler the handler.
 	 */
 	public ContainerPartitionPausingBackOffManager(ListenerContainerRegistry listenerContainerRegistry,
 			BackOffHandler backOffHandler) {

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

@@ -16,6 +16,8 @@
 
 package org.springframework.kafka.listener;
 
+import org.springframework.util.Assert;
+
 /**
  * A factory for {@link ContainerPartitionPausingBackoffManager}.
  *
@@ -30,18 +32,21 @@ public class ContainerPartitionPausingBackOffManagerFactory extends AbstractKafk
 	/**
 	 * Construct an instance with the provided properties.
 	 * @param listenerContainerRegistry the registry.
-	 * @param backOffHandler the back off handler.
 	 */
 	public ContainerPartitionPausingBackOffManagerFactory(ListenerContainerRegistry listenerContainerRegistry) {
-
 		super(listenerContainerRegistry);
 	}
 
 	@Override
 	protected KafkaConsumerBackoffManager doCreateManager(ListenerContainerRegistry registry) {
+		Assert.notNull(this.backOffHandler, "a BackOffHandler is required");
 		return new ContainerPartitionPausingBackOffManager(getListenerContainerRegistry(), this.backOffHandler);
 	}
 
+	/**
+	 * Set the back off handler to use in the created handlers.
+	 * @param backOffHandler the handler.
+	 */
 	public void setBackOffHandler(BackOffHandler backOffHandler) {
 		this.backOffHandler = backOffHandler;
 	}