Remove KafkaBackOffManagerConfigurationSupport

Author: tomazfernandes

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

@@ -24,7 +24,6 @@
 
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
-import org.springframework.kafka.config.KafkaBackOffManagerConfigurationSupport;
 import org.springframework.kafka.config.RetryTopicConfigurationSupport;
 
 /**
@@ -82,6 +81,6 @@
 @Retention(RetentionPolicy.RUNTIME)
 @Target(ElementType.TYPE)
 @Documented
-@Import({RetryTopicConfigurationSupport.class, KafkaBackOffManagerConfigurationSupport.class})
+@Import(RetryTopicConfigurationSupport.class)
 public @interface EnableRetryTopic {
 }

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

@@ -21,6 +21,7 @@
 import java.util.stream.Stream;
 
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
@@ -30,6 +31,9 @@
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
 import org.springframework.kafka.listener.KafkaConsumerBackoffManager;
+import org.springframework.kafka.listener.ListenerContainerRegistry;
+import org.springframework.kafka.listener.MessageListenerContainer;
+import org.springframework.kafka.listener.PartitionPausingBackOffManagerFactory;
 import org.springframework.kafka.retrytopic.DeadLetterPublishingRecovererFactory;
 import org.springframework.kafka.retrytopic.DefaultDestinationTopicResolver;
 import org.springframework.kafka.retrytopic.DestinationTopicProcessor;
@@ -57,10 +61,12 @@
  * @author Tomaz Fernandes
  * @since 2.9
 */
-public class RetryTopicConfigurationSupport {
+public class RetryTopicConfigurationSupport implements DisposableBean {
 
 	private final RetryTopicComponentFactory componentFactory = createComponentFactory();
 
+	private DisposableBean disposableBackOffManagerFactory;
+
 	/**
 	 * Return a global {@link RetryTopicConfigurer} for configuring retry topics
 	 * for {@link KafkaListenerEndpoint} instances with a corresponding
@@ -252,6 +258,24 @@ protected Consumer<DestinationTopicResolver> customizeDestinationTopicResolver()
 		};
 	}
 
+	/**
+	 * Provides the {@link KafkaConsumerBackoffManager} instance.
+	 * Override this method to provide a customized implementation.
+	 * A {@link PartitionPausingBackOffManagerFactory} can be used for that purpose,
+	 * or a different implementation can be provided.
+	 * @param registry the {@link ListenerContainerRegistry} to be used to fetch the
+	 * {@link MessageListenerContainer} to be backed off.
+	 * @return the instance.
+	 */
+	@Bean(name = KafkaListenerConfigUtils.KAFKA_CONSUMER_BACK_OFF_MANAGER_BEAN_NAME)
+	public KafkaConsumerBackoffManager kafkaConsumerBackoffManager(
+			@Qualifier(KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)
+					ListenerContainerRegistry registry) {
+		PartitionPausingBackOffManagerFactory factory = new PartitionPausingBackOffManagerFactory(registry);
+		this.disposableBackOffManagerFactory = factory;
+		return factory.create();
+	}
+
 	/**
 	 * Override this method if you want to provide a subclass
 	 * of {@link RetryTopicComponentFactory} with different
@@ -268,6 +292,13 @@ RetryTopicBootstrapper retryTopicBootstrapper(ApplicationContext context) {
 		return new RetryTopicBootstrapper(context, context.getAutowireCapableBeanFactory());
 	}
 
+	@Override
+	public void destroy() throws Exception {
+		if (this.disposableBackOffManagerFactory != null) {
+			this.disposableBackOffManagerFactory.destroy();
+		}
+	}
+
 	public static class BlockingRetriesConfigurer {
 
 		private BackOff backOff;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2022 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.
@@ -18,8 +18,6 @@
 
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
-import org.springframework.context.ApplicationListener;
-import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.kafka.config.KafkaListenerConfigUtils;
 import org.springframework.util.Assert;
 
@@ -88,10 +86,6 @@ protected <T> T getBean(String beanName, Class<T> beanClass) {
 		return this.applicationContext.getBean(beanName, beanClass);
 	}
 
-	protected void addApplicationListener(ApplicationListener<?> applicationListener) {
-		((ConfigurableApplicationContext) this.applicationContext).addApplicationListener(applicationListener);
-	}
-
 	@Override
 	public void setApplicationContext(ApplicationContext applicationContext) {
 		this.applicationContext = applicationContext;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2022 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.
@@ -18,8 +18,7 @@
 
 import java.time.Clock;
 
-import org.springframework.context.ApplicationListener;
-import org.springframework.context.event.ContextClosedEvent;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.util.Assert;
@@ -32,7 +31,8 @@
  * @author Tomaz Fernandes
  * @since 2.7
  */
-public class PartitionPausingBackOffManagerFactory extends AbstractKafkaBackOffManagerFactory {
+public class PartitionPausingBackOffManagerFactory extends AbstractKafkaBackOffManagerFactory
+		implements DisposableBean {
 
 	private boolean timingAdjustmentEnabled = true;
 
@@ -133,9 +133,7 @@ public final void setTaskExecutor(TaskExecutor taskExecutor) {
 
 	@Override
 	protected KafkaConsumerBackoffManager doCreateManager(ListenerContainerRegistry registry) {
-		PartitionPausingBackoffManager kafkaConsumerBackoffManager = getKafkaConsumerBackoffManager(registry);
-		super.addApplicationListener(kafkaConsumerBackoffManager);
-		return kafkaConsumerBackoffManager;
+		return getKafkaConsumerBackoffManager(registry);
 	}
 
 	protected final Clock getDefaultClock() {
@@ -161,7 +159,15 @@ private TaskExecutor getOrCreateTimingAdjustmentThreadExecutor() {
 		}
 		ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
 		executor.initialize();
-		super.addApplicationListener((ApplicationListener<ContextClosedEvent>) event -> executor.shutdown());
+		this.taskExecutor = executor;
 		return executor;
 	}
+
+	@Override
+	public void destroy() throws Exception {
+		if (this.taskExecutor != null
+				&& ThreadPoolTaskExecutor.class.isAssignableFrom(this.taskExecutor.getClass())) {
+			((ThreadPoolTaskExecutor) this.taskExecutor).shutdown();
+		}
+	}
 }

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

@@ -32,6 +32,7 @@
 import org.springframework.kafka.listener.KafkaConsumerBackoffManager;
 import org.springframework.kafka.listener.KafkaConsumerTimingAdjuster;
 import org.springframework.kafka.listener.PartitionPausingBackOffManagerFactory;
+import org.springframework.kafka.listener.PartitionPausingBackoffManager;
 import org.springframework.retry.backoff.ThreadWaitSleeper;
 
 /**
@@ -110,9 +111,11 @@ private void registerSingletons() {
 	}
 
 	private void addApplicationListeners() {
-		((ConfigurableApplicationContext) this.applicationContext)
-				.addApplicationListener(this.applicationContext.getBean(
-						RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME, DefaultDestinationTopicResolver.class));
+		ConfigurableApplicationContext context = (ConfigurableApplicationContext) this.applicationContext;
+		context.addApplicationListener(this.applicationContext.getBean(
+				RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME, DefaultDestinationTopicResolver.class));
+		context.addApplicationListener(this.applicationContext.getBean(
+				RetryTopicInternalBeanNames.KAFKA_CONSUMER_BACKOFF_MANAGER, PartitionPausingBackoffManager.class));
 	}
 
 	private KafkaConsumerBackoffManager createKafkaConsumerBackoffManager() {