spring-projectsGH-2226: Add RetryTopicConfigurationSupport

Resolves spring-projects#2226

Add RetryTopicConfigurationSupport
Add @EnableRetryTopic
Add KafkaBackOffManagerConfigurationSupport

Author: tomazfernandes

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

@@ -0,0 +1,87 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.annotation;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.kafka.config.KafkaBackOffManagerConfigurationSupport;
+import org.springframework.kafka.config.RetryTopicConfigurationSupport;
+
+/**
+ * Enables the non-blocking topic-based delayed retries feature. To be used in
+ * {@link Configuration Configuration} classes as follows:
+ * <pre class="code">
+ *
+ * &#064;EnableRetryTopic
+ * &#064;Configuration
+ * public class AppConfig {
+ *
+ * 	&#064;Bean
+ * 	public RetryTopicConfiguration myRetryTopicConfiguration(KafkaTemplate kafkaTemplate) {
+ * 		return RetryTopicConfigurationBuilder
+ * 					.newInstance()
+ * 					.maxAttempts(4)
+ * 					.create(kafkaTemplate);
+ * 	}
+ * 	// other &#064;Bean definitions
+ * }
+ * </pre>
+ *
+ * To configure the feature's components, extend the {@link RetryTopicConfigurationSupport}
+ * class and override the appropriate methods. Then import the subclass using the
+ * {@link Import @Import} annotation on a {@link Configuration @Configuration} class,
+ * such as:
+ *
+ * <pre class="code">
+ *
+ * &#064;Configuration
+ * &#064;EnableKafka
+ * &#064;Import(MyRetryTopicConfigurationSupport.class)
+ * public class AppConfig {
+ * }
+ *
+ * public static class MyRetryTopicConfigurationSupport extends RetryTopicConfigurationSupport {
+ *
+ * 		&#064;Override
+ * 		protected void configureBlockingRetries(BlockingRetriesConfigurer blockingRetries) {
+ * 			blockingRetries
+ * 				.retryOn(ShouldRetryOnlyBlockingException.class, ShouldRetryViaBothException.class)
+ * 				.backOff(new FixedBackOff(50, 3));
+ * 		}
+ *
+ * 		&#064;Override
+ * 		protected void configureNonBlockingRetries(NonBlockingRetriesConfigurer nonBlockingRetries) {
+ * 			nonBlockingRetries
+ * 				.addToFatalExceptions(ShouldSkipBothRetriesException.class);
+ * }
+ * </pre>
+ *
+ * @author Tomaz Fernandes
+ * @since 2.9
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+@Documented
+@Import({RetryTopicConfigurationSupport.class, KafkaBackOffManagerConfigurationSupport.class})
+public @interface EnableRetryTopic {
+}

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

@@ -86,6 +86,7 @@
 import org.springframework.kafka.listener.ContainerGroupSequencer;
 import org.springframework.kafka.listener.KafkaListenerErrorHandler;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
+import org.springframework.kafka.retrytopic.RetryTopicBeanNames;
 import org.springframework.kafka.retrytopic.RetryTopicBootstrapper;
 import org.springframework.kafka.retrytopic.RetryTopicConfiguration;
 import org.springframework.kafka.retrytopic.RetryTopicConfigurer;
@@ -509,11 +510,15 @@ private boolean processMainAndRetryListeners(KafkaListener kafkaListener, Object
 		return true;
 	}
 
+	@SuppressWarnings("deprecation")
 	private RetryTopicConfigurer getRetryTopicConfigurer() {
 		bootstrapRetryTopicIfNecessary();
-		return this.beanFactory.getBean(RetryTopicInternalBeanNames.RETRY_TOPIC_CONFIGURER, RetryTopicConfigurer.class);
+		return this.beanFactory.containsBean(RetryTopicInternalBeanNames.RETRY_TOPIC_CONFIGURER)
+				? this.beanFactory.getBean(RetryTopicInternalBeanNames.RETRY_TOPIC_CONFIGURER, 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 "

spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopicAnnotationProcessor.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.
@@ -32,6 +32,7 @@
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
 import org.springframework.kafka.core.KafkaOperations;
+import org.springframework.kafka.retrytopic.RetryTopicBeanNames;
 import org.springframework.kafka.retrytopic.RetryTopicConfiguration;
 import org.springframework.kafka.retrytopic.RetryTopicConfigurationBuilder;
 import org.springframework.kafka.retrytopic.RetryTopicConfigurer;
@@ -200,6 +201,7 @@ private EndpointHandlerMethod getDltProcessor(Method listenerMethod, Object bean
 				.orElse(RetryTopicConfigurer.DEFAULT_DLT_HANDLER);
 	}
 
+	@SuppressWarnings("deprecation")
 	private KafkaOperations<?, ?> getKafkaTemplate(String kafkaTemplateName, String[] topics) {
 		if (StringUtils.hasText(kafkaTemplateName)) {
 			Assert.state(this.beanFactory != null, "BeanFactory must be set to obtain kafka template by bean name");
@@ -218,12 +220,19 @@ private EndpointHandlerMethod getDltProcessor(Method listenerMethod, Object bean
 		}
 		catch (NoSuchBeanDefinitionException ex) {
 			try {
-				return this.beanFactory.getBean(DEFAULT_SPRING_BOOT_KAFKA_TEMPLATE_NAME, KafkaOperations.class);
+				return this.beanFactory.getBean(RetryTopicBeanNames.DEFAULT_KAFKA_TEMPLATE_BEAN_NAME,
+						KafkaOperations.class);
 			}
-			catch (NoSuchBeanDefinitionException exc) {
-				exc.addSuppressed(ex);
-				throw new BeanInitializationException("Could not find a KafkaTemplate to configure the retry topics.", // NOSONAR (lost stack trace)
-						exc);
+			catch (NoSuchBeanDefinitionException ex2) {
+				try {
+					return this.beanFactory.getBean(DEFAULT_SPRING_BOOT_KAFKA_TEMPLATE_NAME, KafkaOperations.class);
+				}
+				catch (NoSuchBeanDefinitionException exc) {
+					exc.addSuppressed(ex);
+					exc.addSuppressed(ex2);
+					throw new BeanInitializationException("Could not find a KafkaTemplate to configure the retry topics.", // NOSONAR (lost stack trace)
+							exc);
+				}
 			}
 		}
 	}

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

@@ -0,0 +1,108 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.config;
+
+import java.time.Clock;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.event.ContextClosedEvent;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.kafka.listener.KafkaConsumerBackoffManager;
+import org.springframework.kafka.listener.KafkaConsumerTimingAdjuster;
+import org.springframework.kafka.listener.ListenerContainerRegistry;
+import org.springframework.kafka.listener.PartitionPausingBackoffManager;
+import org.springframework.kafka.listener.WakingKafkaConsumerTimingAdjuster;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.util.Assert;
+
+/**
+ * This class provides configuration support for a global
+ * {@link KafkaConsumerBackoffManager} instance. Consider overriding any of the
+ * protected methods for providing different components or configuration.
+ * This class is automatically imported by the
+ * {@link org.springframework.kafka.annotation.EnableRetryTopic @EnableRetryTopic}
+ * annotation.
+ *
+ * @author Tomaz Fernandes
+ * @since 2.9
+ */
+public class KafkaBackOffManagerConfigurationSupport implements ApplicationContextAware {
+
+	private ConfigurableApplicationContext applicationContext;
+
+	/**
+	 * Provides the {@link KafkaConsumerBackoffManager} instance.
+	 * To customize it or any of the components, consider overriding
+	 * one of the more fine graned methods:
+	 * <ul>
+	 *     <li>{@link #backOffManagerClock}</li>
+	 *     <li>{@link #timingAdjuster}</li>
+	 *     <li>{@link #timingAdjusterTaskExecutor}</li>
+	 * </ul>
+	 * @param registry the global {@link ListenerContainerRegistry} instance.
+	 * @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) {
+		return new PartitionPausingBackoffManager(registry, timingAdjuster(timingAdjusterTaskExecutor()), backOffManagerClock());
+	}
+
+	/**
+	 * Override this method to provide a different {@link Clock}
+	 * instance to be used with the {@link KafkaConsumerBackoffManager}.
+	 * @return the instance.
+	 */
+	protected Clock backOffManagerClock() {
+		return Clock.systemUTC();
+	}
+
+	/**
+	 * Override this method to provide a different {@link KafkaConsumerTimingAdjuster}
+	 * to be used with the {@link KafkaConsumerBackoffManager}.
+	 * @param taskExecutor the task executor.
+	 * @return the instance.
+	 */
+	protected KafkaConsumerTimingAdjuster timingAdjuster(TaskExecutor taskExecutor) {
+		return new WakingKafkaConsumerTimingAdjuster(taskExecutor);
+	}
+
+	/**
+	 * Override this method to provide a different {@link TaskExecutor}
+	 * to be used with the {@link KafkaConsumerTimingAdjuster}.
+	 * @return the instance.
+	 */
+	protected TaskExecutor timingAdjusterTaskExecutor() {
+		ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+		executor.initialize();
+		this.applicationContext
+				.addApplicationListener((ApplicationListener<ContextClosedEvent>) event -> executor.shutdown());
+		return executor;
+	}
+
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+		Assert.isInstanceOf(ConfigurableApplicationContext.class, applicationContext);
+		this.applicationContext = (ConfigurableApplicationContext) applicationContext;
+	}
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-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.
@@ -21,6 +21,7 @@
  *
  * @author Juergen Hoeller
  * @author Gary Russell
+ * @author Tomaz Fernandes
  */
 public abstract class KafkaListenerConfigUtils {
 
@@ -36,4 +37,10 @@ public abstract class KafkaListenerConfigUtils {
 	public static final String KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME =
 			"org.springframework.kafka.config.internalKafkaListenerEndpointRegistry";
 
+	/**
+	 * The bean name of the internally managed Kafka consumer back off manager.
+	 */
+	public static final String KAFKA_CONSUMER_BACK_OFF_MANAGER_BEAN_NAME =
+			"org.springframework.kafka.config.internalKafkaConsumerBackOffManager";
+
 }