Address PR Comments.

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/retrytopic.adoc

@@ -41,10 +41,6 @@ That being said, for consumers handling a single partition the message's process
 
 IMPORTANT: It is guaranteed that a message will never be processed before its due time.
 
-===== Tuning the Delay Precision
-
-Starting with version 2.9, it is no longer necessary to tune the precision because a task scheduler is used to resume the partition and wake the consumer, if necessary.
-
 [[retry-config]]
 ==== Configuration
 

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

@@ -186,6 +186,8 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 
 	private AnnotationEnhancer enhancer;
 
+	private RetryTopicConfigurer retryTopicConfigurer;
+
 	@Override
 	public int getOrder() {
 		return LOWEST_PRECEDENCE;
@@ -508,8 +510,16 @@ private boolean processMainAndRetryListeners(KafkaListener kafkaListener, Object
 	}
 
 	private RetryTopicConfigurer getRetryTopicConfigurer() {
-		return this.beanFactory.getBean(RetryTopicBeanNames.RETRY_TOPIC_CONFIGURER_BEAN_NAME,
-				RetryTopicConfigurer.class);
+		try {
+			this.retryTopicConfigurer = this.beanFactory.getBean(RetryTopicBeanNames.RETRY_TOPIC_CONFIGURER_BEAN_NAME,
+					RetryTopicConfigurer.class);
+		}
+		catch (NoSuchBeanDefinitionException ex) {
+			this.logger.error("A 'RetryTopicConfigurer' with name "
+					+ RetryTopicBeanNames.RETRY_TOPIC_CONFIGURER_BEAN_NAME + "is required.");
+			throw ex;
+		}
+		return this.retryTopicConfigurer;
 	}
 
 	private Method checkProxy(Method methodArg, Object bean) {

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

@@ -76,8 +76,6 @@ public class RetryableTopicAnnotationProcessor {
 
 	private final BeanExpressionContext expressionContext;
 
-	private static final String DEFAULT_SPRING_BOOT_KAFKA_TEMPLATE_NAME = "kafkaTemplate";
-
 	/**
 	 * Construct an instance using the provided parameters and default resolver,
 	 * expression context.
@@ -218,14 +216,11 @@ private EndpointHandlerMethod getDltProcessor(Method listenerMethod, Object bean
 					KafkaOperations.class);
 		}
 		catch (NoSuchBeanDefinitionException ex2) {
-			try {
-				return this.beanFactory.getBean(DEFAULT_SPRING_BOOT_KAFKA_TEMPLATE_NAME, KafkaOperations.class);
-			}
-			catch (NoSuchBeanDefinitionException exc) {
-				exc.addSuppressed(ex2);
-				throw new BeanInitializationException("Could not find a KafkaTemplate to configure the retry topics.", // NOSONAR (lost stack trace)
-						exc);
-			}
+			KafkaOperations<?, ?> kafkaOps = this.beanFactory.getBeanProvider(KafkaOperations.class).getIfUnique();
+			Assert.state(kafkaOps != null, () -> "A single KafkaTemplate bean could not be found in the context; "
+					+ " a single instance must exist, or one specifically named "
+					+ RetryTopicBeanNames.DEFAULT_KAFKA_TEMPLATE_BEAN_NAME);
+			return kafkaOps;
 		}
 	}
 

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

@@ -90,4 +90,5 @@ protected <T> T getBean(String beanName, Class<T> beanClass) {
 	public void setApplicationContext(ApplicationContext applicationContext) {
 		this.applicationContext = applicationContext;
 	}
+
 }

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

@@ -51,6 +51,8 @@ public class ContainerPartitionPausingBackOffManager implements KafkaConsumerBac
 	public ContainerPartitionPausingBackOffManager(ListenerContainerRegistry listenerContainerRegistry,
 			BackOffHandler backOffHandler) {
 
+		Assert.notNull(listenerContainerRegistry, "'listenerContainerRegistry' cannot be null");
+		Assert.notNull(backOffHandler, "'backOffHandler' cannot be null");
 		this.listenerContainerRegistry = listenerContainerRegistry;
 		this.backOffHandler = backOffHandler;
 	}
@@ -71,11 +73,12 @@ public void backOffIfNecessary(Context context) {
 
 	private void pauseConsumptionAndThrow(Context context, Long backOffTime) throws KafkaBackoffException {
 		TopicPartition topicPartition = context.getTopicPartition();
-		getListenerContainerFromContext(context).pausePartition(topicPartition);
-		this.backOffHandler.onNextBackOff(getListenerContainerFromContext(context), topicPartition, backOffTime);
+		MessageListenerContainer container = getListenerContainerFromContext(context);
+		container.pausePartition(topicPartition);
+		this.backOffHandler.onNextBackOff(container, topicPartition, backOffTime);
 		throw new KafkaBackoffException(String.format("Partition %s from topic %s is not ready for consumption, " +
-				"backing off for approx. %s millis.", context.getTopicPartition().partition(),
-				context.getTopicPartition().topic(), backOffTime),
+				"backing off for approx. %s millis.", topicPartition.partition(),
+				topicPartition.topic(), backOffTime),
 				topicPartition, context.getListenerId(), context.getDueTimestamp());
 	}
 

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

@@ -71,7 +71,7 @@ public class ListenerContainerFactoryConfigurer {
 	private Consumer<ConcurrentMessageListenerContainer<?, ?>> containerCustomizer = container -> {
 	};
 
-	private Consumer<CommonErrorHandler> errorHandlerCustomizer = errorHandler -> {
+	private Consumer<DefaultErrorHandler> errorHandlerCustomizer = errorHandler -> {
 	};
 
 	private final DeadLetterPublishingRecovererFactory deadLetterPublishingRecovererFactory;
@@ -152,7 +152,7 @@ public void setContainerCustomizer(Consumer<ConcurrentMessageListenerContainer<?
 		this.containerCustomizer = containerCustomizer;
 	}
 
-	public void setErrorHandlerCustomizer(Consumer<CommonErrorHandler> errorHandlerCustomizer) {
+	public void setErrorHandlerCustomizer(Consumer<DefaultErrorHandler> errorHandlerCustomizer) {
 		this.errorHandlerCustomizer = errorHandlerCustomizer;
 	}
 

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

@@ -349,7 +349,7 @@ public BlockingRetriesConfigurer backOff(BackOff backoff) {
 	 */
 	public static class CustomizersConfigurer {
 
-		private Consumer<CommonErrorHandler> errorHandlerCustomizer;
+		private Consumer<DefaultErrorHandler> errorHandlerCustomizer;
 
 		private Consumer<ConcurrentMessageListenerContainer<?, ?>> listenerContainerCustomizer;
 
@@ -362,7 +362,7 @@ public static class CustomizersConfigurer {
 		 * @return the configurer.
 		 * @see DefaultErrorHandler
 		 */
-		public CustomizersConfigurer customizeErrorHandler(Consumer<CommonErrorHandler> errorHandlerCustomizer) {
+		public CustomizersConfigurer customizeErrorHandler(Consumer<DefaultErrorHandler> errorHandlerCustomizer) {
 			this.errorHandlerCustomizer = errorHandlerCustomizer;
 			return this;
 		}

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

@@ -28,6 +28,8 @@
  * framework requires a scheduler bean and looks for one in this order: 1. A single
  * instance of this class, 2. a single {@link TaskScheduler} bean, 3. when multiple
  * {@link TaskScheduler}s are present, a bean with the name {@code taskScheduler}.
+ * If you use this class, you should provide a {@link TaskScheduler} that is not defined
+ * as a bean; this class will maintain the scheduler's lifecycle.
  *
  * @author Gary Russell
  * @since 2.9

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

@@ -87,7 +87,7 @@ class ListenerContainerFactoryConfigurerTests {
 	private ContainerProperties containerProperties;
 
 	@Captor
-	private ArgumentCaptor<CommonErrorHandler> errorHandlerCaptor;
+	private ArgumentCaptor<DefaultErrorHandler> errorHandlerCaptor;
 
 	private final ConsumerRecord<?, ?> record =
 			new ConsumerRecord<>("test-topic", 1, 1234L, new Object(), new Object());
@@ -104,7 +104,7 @@ class ListenerContainerFactoryConfigurerTests {
 	private OffsetCommitCallback offsetCommitCallback;
 
 	@Mock
-	private java.util.function.Consumer<CommonErrorHandler> errorHandlerCustomizer;
+	private java.util.function.Consumer<DefaultErrorHandler> errorHandlerCustomizer;
 
 	@SuppressWarnings("rawtypes")
 	@Captor
@@ -173,12 +173,10 @@ void shouldSetupErrorHandling() {
 
 		// then
 		then(container).should(times(1)).setCommonErrorHandler(errorHandlerCaptor.capture());
-		CommonErrorHandler errorHandler = errorHandlerCaptor.getValue();
-		assertThat(DefaultErrorHandler.class.isAssignableFrom(errorHandler.getClass())).isTrue();
-		DefaultErrorHandler seekToCurrent = (DefaultErrorHandler) errorHandler;
+		DefaultErrorHandler errorHandler = errorHandlerCaptor.getValue();
 
 		RuntimeException ex = new RuntimeException();
-		seekToCurrent.handleRemaining(ex, records, consumer, container);
+		errorHandler.handleRemaining(ex, records, consumer, container);
 
 		then(recoverer).should(times(1)).accept(record, consumer, ex);
 		then(consumer).should(times(1)).commitAsync(any(Map.class), eq(offsetCommitCallback));

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

@@ -35,10 +35,10 @@
 import org.mockito.ArgumentCaptor;
 
 import org.springframework.beans.factory.BeanFactory;
-import org.springframework.kafka.listener.CommonErrorHandler;
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.ContainerPartitionPausingBackOffManagerFactory;
 import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
+import org.springframework.kafka.listener.DefaultErrorHandler;
 import org.springframework.kafka.listener.KafkaConsumerBackoffManager;
 import org.springframework.kafka.listener.ListenerContainerRegistry;
 import org.springframework.kafka.support.converter.ConversionException;
@@ -87,7 +87,7 @@ void testCreateConfigurer() {
 		Consumer<DeadLetterPublishingRecovererFactory> dlprfCustomizer = mock(Consumer.class);
 		Consumer<RetryTopicConfigurer> rtconfigurer = mock(Consumer.class);
 		Consumer<ListenerContainerFactoryConfigurer> lcfcConsumer = mock(Consumer.class);
-		Consumer<CommonErrorHandler> errorHandlerCustomizer = mock(Consumer.class);
+		Consumer<DefaultErrorHandler> errorHandlerCustomizer = mock(Consumer.class);
 		BackOff backoff = mock(BackOff.class);
 
 		RetryTopicConfigurationSupport support = new RetryTopicConfigurationSupport() {

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

@@ -18,6 +18,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
@@ -33,6 +34,7 @@
 
 import org.springframework.beans.factory.BeanInitializationException;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.kafka.annotation.DltHandler;
@@ -166,13 +168,16 @@ void shouldThrowIfNoKafkaTemplateFound() {
 		given(this.beanFactory.getBean(RetryTopicBeanNames.DEFAULT_KAFKA_TEMPLATE_BEAN_NAME, KafkaOperations.class))
 				.willThrow(NoSuchBeanDefinitionException.class);
 
-		given(this.beanFactory.getBean("kafkaTemplate", KafkaOperations.class))
-				.willThrow(NoSuchBeanDefinitionException.class);
+		@SuppressWarnings({ "unchecked", "rawtypes" })
+		ObjectProvider<KafkaOperations> templateProvider = mock(ObjectProvider.class);
+		given(templateProvider.getIfUnique()).willReturn(null);
+		given(this.beanFactory.getBeanProvider(KafkaOperations.class))
+				.willReturn(templateProvider);
 
 		RetryableTopicAnnotationProcessor processor = new RetryableTopicAnnotationProcessor(beanFactory);
 
 		// given - then
-		assertThatExceptionOfType(BeanInitializationException.class).isThrownBy(() ->
+		assertThatIllegalStateException().isThrownBy(() ->
 				processor.processAnnotation(topics, listenWithRetryAndDlt, annotationWithDlt, beanWithDlt));
 	}
 
@@ -182,8 +187,11 @@ void shouldTrySpringBootDefaultKafkaTemplate() {
 		// setup
 		given(this.beanFactory.getBean(RetryTopicBeanNames.DEFAULT_KAFKA_TEMPLATE_BEAN_NAME, KafkaOperations.class))
 				.willThrow(NoSuchBeanDefinitionException.class);
-		given(this.beanFactory.getBean("kafkaTemplate", KafkaOperations.class))
-				.willReturn(kafkaOperationsFromDefaultName);
+		@SuppressWarnings({ "unchecked", "rawtypes" })
+		ObjectProvider<KafkaOperations> templateProvider = mock(ObjectProvider.class);
+		given(templateProvider.getIfUnique()).willReturn(kafkaOperationsFromDefaultName);
+		given(this.beanFactory.getBeanProvider(KafkaOperations.class))
+				.willReturn(templateProvider);
 		RetryableTopicAnnotationProcessor processor = new RetryableTopicAnnotationProcessor(beanFactory);
 
 		// given - then