GH-1863 - Configure listener error handler and type converter for @RetryableTopics

Author: tomazfernandes

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

@@ -465,7 +465,7 @@ private boolean processMainAndRetryListeners(KafkaListener kafkaListener, Object
 		}
 
 		RetryTopicConfigurer.EndpointProcessor endpointProcessor = endpointToProcess ->
-				this.doProcessKafkaListenerAnnotation(endpointToProcess, kafkaListener, bean);
+				this.processKafkaListenerAnnotationForRetryTopic(endpointToProcess, kafkaListener, bean);
 
 		String beanRef = kafkaListener.beanRef();
 		this.listenerScope.addListener(beanRef, bean);
@@ -546,28 +546,26 @@ protected void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, Kafka
 			this.listenerScope.addListener(beanRef, bean);
 		}
 
-		doProcessKafkaListenerAnnotation(endpoint, kafkaListener, bean);
+		processKafkaListenerAnnotationBeforeRegistration(endpoint, kafkaListener, bean);
 
 		String containerFactory = resolve(kafkaListener.containerFactory());
 		KafkaListenerContainerFactory<?> listenerContainerFactory = resolveContainerFactory(kafkaListener, containerFactory, beanName);
 
 		this.registrar.registerEndpoint(endpoint, listenerContainerFactory);
 
-		endpoint.setBeanFactory(this.beanFactory);
-		String errorHandlerBeanName = resolveExpressionAsString(kafkaListener.errorHandler(), "errorHandler");
-		if (StringUtils.hasText(errorHandlerBeanName)) {
-			resolveErrorHandler(endpoint, kafkaListener);
-		}
-		String converterBeanName = resolveExpressionAsString(kafkaListener.contentTypeConverter(), "contentTypeConverter");
-		if (StringUtils.hasText(converterBeanName)) {
-			resolveContentTypeConverter(endpoint, kafkaListener);
-		}
+		processKafkaListenerEndpointAfterRegistration(endpoint, kafkaListener);
+
 		if (StringUtils.hasText(beanRef)) {
 			this.listenerScope.removeListener(beanRef);
 		}
 	}
 
-	private void doProcessKafkaListenerAnnotation(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener, Object bean) {
+	private void processKafkaListenerAnnotationForRetryTopic(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener, Object bean) {
+		processKafkaListenerAnnotationBeforeRegistration(endpoint, kafkaListener, bean);
+		processKafkaListenerEndpointAfterRegistration(endpoint, kafkaListener);
+	}
+
+	private void processKafkaListenerAnnotationBeforeRegistration(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener, Object bean) {
 		endpoint.setBean(bean);
 		endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
 		endpoint.setId(getEndpointId(kafkaListener));
@@ -595,6 +593,18 @@ private void doProcessKafkaListenerAnnotation(MethodKafkaListenerEndpoint<?, ?>
 		endpoint.setSplitIterables(kafkaListener.splitIterables());
 	}
 
+	private void processKafkaListenerEndpointAfterRegistration(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener) {
+		endpoint.setBeanFactory(this.beanFactory);
+		String errorHandlerBeanName = resolveExpressionAsString(kafkaListener.errorHandler(), "errorHandler");
+		if (StringUtils.hasText(errorHandlerBeanName)) {
+			resolveErrorHandler(endpoint, kafkaListener);
+		}
+		String converterBeanName = resolveExpressionAsString(kafkaListener.contentTypeConverter(), "contentTypeConverter");
+		if (StringUtils.hasText(converterBeanName)) {
+			resolveContentTypeConverter(endpoint, kafkaListener);
+		}
+	}
+
 	private void resolveErrorHandler(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener) {
 		Object errorHandler = resolveExpression(kafkaListener.errorHandler());
 		if (errorHandler instanceof KafkaListenerErrorHandler) {

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

@@ -22,6 +22,7 @@
 import java.time.Clock;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -53,9 +54,14 @@
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.listener.ContainerProperties;
+import org.springframework.kafka.listener.KafkaListenerErrorHandler;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.converter.CompositeMessageConverter;
+import org.springframework.messaging.converter.GenericMessageConverter;
+import org.springframework.messaging.converter.SmartMessageConverter;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.retry.annotation.Backoff;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
@@ -105,6 +111,8 @@ void shouldRetryFirstTopic() {
 		kafkaTemplate.send(FIRST_TOPIC, "Testing topic 1");
 		assertThat(awaitLatch(latchContainer.countDownLatch1)).isTrue();
 		assertThat(awaitLatch(latchContainer.customDltCountdownLatch)).isTrue();
+		assertThat(awaitLatch(latchContainer.customErrorHandlerCountdownLatch)).isTrue();
+		assertThat(awaitLatch(latchContainer.customMessageConverterCountdownLatch)).isTrue();
 	}
 
 	@Test
@@ -154,7 +162,8 @@ static class FirstTopicListener {
 		@Autowired
 		CountDownLatchContainer container;
 
-		@KafkaListener(id = "firstTopicId", topics = FIRST_TOPIC, containerFactory = MAIN_TOPIC_CONTAINER_FACTORY)
+		@KafkaListener(id = "firstTopicId", topics = FIRST_TOPIC, containerFactory = MAIN_TOPIC_CONTAINER_FACTORY,
+				errorHandler = "myCustomErrorHandler", contentTypeConverter = "myCustomMessageConverter")
 		public void listen(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) {
 			logger.debug("Message {} received in topic {}", message, receivedTopic);
 				container.countDownLatch1.countDown();
@@ -256,6 +265,8 @@ static class CountDownLatchContainer {
 		CountDownLatch countDownLatchDltOne = new CountDownLatch(1);
 		CountDownLatch countDownLatchDltTwo = new CountDownLatch(1);
 		CountDownLatch customDltCountdownLatch = new CountDownLatch(1);
+		CountDownLatch customErrorHandlerCountdownLatch = new CountDownLatch(6);
+		CountDownLatch customMessageConverterCountdownLatch = new CountDownLatch(6);
 
 		List<String> knownTopics = new ArrayList<>();
 
@@ -334,6 +345,26 @@ public FirstTopicListener firstTopicListener() {
 			return new FirstTopicListener();
 		}
 
+		@Bean
+		public KafkaListenerErrorHandler myCustomErrorHandler(CountDownLatchContainer container) {
+			return (message, exception) -> {
+				container.customErrorHandlerCountdownLatch.countDown();
+				throw exception;
+			};
+		}
+
+		@Bean
+		public SmartMessageConverter myCustomMessageConverter(CountDownLatchContainer container) {
+			return new CompositeMessageConverter(Collections.singletonList(new GenericMessageConverter())) {
+
+				@Override
+				public Object fromMessage(Message<?> message, Class<?> targetClass, Object conversionHint) {
+					container.customMessageConverterCountdownLatch.countDown();
+					return super.fromMessage(message, targetClass, conversionHint);
+				}
+			};
+		}
+
 		@Bean
 		public SecondTopicListener secondTopicListener() {
 			return new SecondTopicListener();