GH-527: @KafkaListener: populate converter beans

Fixes #527

* Add test

* Changed listener factory creation order

* Add client id to manualConsumerFactory call

* Refactor to add Converter beans to DefaultFormattingConversionService

* Move addFormatters to afterSingletonsInstantiated

* Add documentation

* Improve documentation

Author: zersetz-end

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2017 the original author or authors.
+ * Copyright 2014-2018 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.
@@ -50,6 +50,10 @@
 import org.springframework.core.MethodIntrospector;
 import org.springframework.core.Ordered;
 import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.core.convert.converter.GenericConverter;
+import org.springframework.format.Formatter;
+import org.springframework.format.FormatterRegistry;
 import org.springframework.format.support.DefaultFormattingConversionService;
 import org.springframework.kafka.config.KafkaListenerConfigUtils;
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
@@ -98,6 +102,7 @@
  * @author Artem Bilan
  * @author Dariusz Szablinski
  * @author Venil Noronha
+ * @author Dimitri Penner
  *
  * @see KafkaListener
  * @see KafkaListenerErrorHandler
@@ -224,6 +229,9 @@ public void afterSingletonsInstantiated() {
 		if (handlerMethodFactory != null) {
 			this.messageHandlerMethodFactory.setMessageHandlerMethodFactory(handlerMethodFactory);
 		}
+		else {
+			addFormatters(this.messageHandlerMethodFactory.defaultFormattingConversionService);
+		}
 
 		// Actually register all listeners
 		this.registrar.afterPropertiesSet();
@@ -630,6 +638,27 @@ private String resolve(String value) {
 		return value;
 	}
 
+	private void addFormatters(FormatterRegistry registry) {
+		for (Converter<?, ?> converter : getBeansOfType(Converter.class)) {
+			registry.addConverter(converter);
+		}
+		for (GenericConverter converter : getBeansOfType(GenericConverter.class)) {
+			registry.addConverter(converter);
+		}
+		for (Formatter<?> formatter : getBeansOfType(Formatter.class)) {
+			registry.addFormatter(formatter);
+		}
+	}
+
+	private <T> Collection<T> getBeansOfType(Class<T> type) {
+		if (KafkaListenerAnnotationBeanPostProcessor.this.beanFactory instanceof ListableBeanFactory) {
+			return ((ListableBeanFactory) KafkaListenerAnnotationBeanPostProcessor.this.beanFactory).getBeansOfType(type).values();
+		}
+		else {
+			return Collections.emptySet();
+		}
+	}
+
 	/**
 	 * An {@link MessageHandlerMethodFactory} adapter that offers a configurable underlying
 	 * instance to use. Useful if the factory to use is determined once the endpoints
@@ -638,6 +667,8 @@ private String resolve(String value) {
 	 */
 	private class KafkaHandlerMethodFactoryAdapter implements MessageHandlerMethodFactory {
 
+		private DefaultFormattingConversionService defaultFormattingConversionService = new DefaultFormattingConversionService();
+
 		private MessageHandlerMethodFactory messageHandlerMethodFactory;
 
 		public void setMessageHandlerMethodFactory(MessageHandlerMethodFactory kafkaHandlerMethodFactory1) {
@@ -664,17 +695,17 @@ private MessageHandlerMethodFactory createDefaultMessageHandlerMethodFactory() {
 					(KafkaListenerAnnotationBeanPostProcessor.this.beanFactory instanceof ConfigurableBeanFactory ?
 							(ConfigurableBeanFactory) KafkaListenerAnnotationBeanPostProcessor.this.beanFactory : null);
 
-			DefaultFormattingConversionService conversionService = new DefaultFormattingConversionService();
-			defaultFactory.setConversionService(conversionService);
+
+			defaultFactory.setConversionService(this.defaultFormattingConversionService);
 
 			List<HandlerMethodArgumentResolver> argumentResolvers = new ArrayList<>();
 
 			// Annotation-based argument resolution
-			argumentResolvers.add(new HeaderMethodArgumentResolver(conversionService, cbf));
+			argumentResolvers.add(new HeaderMethodArgumentResolver(this.defaultFormattingConversionService, cbf));
 			argumentResolvers.add(new HeadersMethodArgumentResolver());
 
 			// Type-based argument resolution
-			final GenericMessageConverter messageConverter = new GenericMessageConverter(conversionService);
+			final GenericMessageConverter messageConverter = new GenericMessageConverter(this.defaultFormattingConversionService);
 			argumentResolvers.add(new MessageMethodArgumentResolver(messageConverter));
 			argumentResolvers.add(new PayloadArgumentResolver(messageConverter) {
 

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -20,6 +20,8 @@
 import static org.mockito.ArgumentMatchers.anyMap;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.BDDMockito.willAnswer;
+import static org.mockito.BDDMockito.willReturn;
+import static org.mockito.BDDMockito.willThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 
@@ -50,6 +52,7 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.event.EventListener;
 import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
+import org.springframework.core.convert.converter.Converter;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
@@ -104,6 +107,7 @@
  * @author Artem Bilan
  * @author Dariusz Szablinski
  * @author Venil Noronha
+ * @author Dimitri Penner
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
@@ -120,7 +124,7 @@ public class EnableKafkaIntegrationTests {
 			"annotated18", "annotated19", "annotated20", "annotated21", "annotated21reply", "annotated22",
 			"annotated22reply", "annotated23", "annotated23reply", "annotated24", "annotated24reply",
 			"annotated25", "annotated25reply1", "annotated25reply2", "annotated26", "annotated27", "annotated28",
-			"annotated29", "annotated30", "annotated30reply", "annotated31");
+			"annotated29", "annotated30", "annotated30reply", "annotated31", "annotated32");
 
 //	@Rule
 //	public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.TRACE,
@@ -159,6 +163,9 @@ public class EnableKafkaIntegrationTests {
 	@Autowired
 	private List<?> quxGroup;
 
+	@Autowired
+	private FooConverter fooConverter;
+
 	@Test
 	public void testAnonymous() {
 		MessageListenerContainer container = this.registry
@@ -598,6 +605,25 @@ public void testBadAckConfig() throws Exception {
 						+ "the listener container must have a MANUAL Ackmode to populate the Acknowledgment.");
 	}
 
+	@Test
+	public void testConverterBean() throws Exception {
+		@SuppressWarnings("unchecked")
+		Converter<String, Foo> converterDelegate = mock(Converter.class);
+		fooConverter.setDelegate(converterDelegate);
+
+		Foo foo = new Foo();
+		willReturn(foo).given(converterDelegate).convert("{'bar':'foo'}");
+		template.send("annotated32", 0, 1, "{'bar':'foo'}");
+		assertThat(this.listener.latch20.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.listener.listen16foo).isEqualTo(foo);
+
+		willThrow(new RuntimeException()).given(converterDelegate).convert("foobar");
+		template.send("annotated32", 0, 1, "foobar");
+		assertThat(this.config.listen16ErrorLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.config.listen16Exception).isNotNull();
+		assertThat(this.config.listen16Message).isEqualTo("foobar");
+	}
+
 	@Configuration
 	@EnableKafka
 	@EnableTransactionManagement(proxyTargetClass = true)
@@ -769,6 +795,19 @@ public KafkaListenerContainerFactory<?> batchManualFactory2() {
 			return factory;
 		}
 
+		@Bean
+		public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
+		recordAckListenerContainerFactory() {
+			ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
+					new ConcurrentKafkaListenerContainerFactory<>();
+			factory.setConsumerFactory(manualConsumerFactory("clientIdViaProps4"));
+			ContainerProperties props = factory.getContainerProperties();
+			props.setAckMode(AckMode.RECORD);
+			props.setAckOnError(true);
+			props.setErrorHandler(listen16ErrorHandler());
+			return factory;
+		}
+
 		@Bean
 		public DefaultKafkaConsumerFactory<Integer, String> consumerFactory() {
 			return new DefaultKafkaConsumerFactory<>(consumerConfigs());
@@ -976,6 +1015,26 @@ public KafkaListenerErrorHandler voidSendToErrorHandler() {
 			};
 		}
 
+		private Throwable listen16Exception;
+
+		private Object listen16Message;
+
+		private CountDownLatch listen16ErrorLatch = new CountDownLatch(1);
+
+		@Bean
+		public ConsumerAwareErrorHandler listen16ErrorHandler() {
+			return (e, r, c) -> {
+				listen16Exception = e;
+				listen16Message = r.value();
+				listen16ErrorLatch.countDown();
+			};
+		}
+
+		@Bean
+		public FooConverter fooConverter() {
+			return new FooConverter();
+		}
+
 	}
 
 	static class Listener implements ConsumerSeekAware {
@@ -1020,6 +1079,8 @@ static class Listener implements ConsumerSeekAware {
 
 		private final CountDownLatch latch19 = new CountDownLatch(1);
 
+		private final CountDownLatch latch20 = new CountDownLatch(1);
+
 		private final CountDownLatch eventLatch = new CountDownLatch(1);
 
 		private volatile Integer partition;
@@ -1036,6 +1097,8 @@ static class Listener implements ConsumerSeekAware {
 
 		private Foo foo;
 
+		private Foo listen16foo;
+
 		private volatile ListenerContainerIdleEvent event;
 
 		private volatile List<Integer> keys;
@@ -1214,6 +1277,13 @@ public void listen15(List<Message<?>> list, Acknowledgment ack) {
 			this.latch15.countDown();
 		}
 
+		@KafkaListener(id = "converter", topics = "annotated32", containerFactory = "recordAckListenerContainerFactory",
+				groupId = "converter.explicitGroupId")
+		public void listen16(Foo foo) {
+			this.listen16foo = foo;
+			this.latch20.countDown();
+		}
+
 		@KafkaListener(id = "errorHandler", topics = "annotated20", errorHandler = "consumeException")
 		public String errorHandler(String data) throws Exception {
 			throw new Exception("return this");
@@ -1405,4 +1475,22 @@ public boolean filter(ConsumerRecord<Integer, String> consumerRecord) {
 
 	}
 
+	public static class FooConverter implements Converter<String, Foo> {
+
+		private Converter<String, Foo> delegate;
+
+		public Converter<String, Foo> getDelegate() {
+			return delegate;
+		}
+
+		public void setDelegate(
+				Converter<String, Foo> delegate) {
+			this.delegate = delegate;
+		}
+
+		@Override
+		public Foo convert(String source) {
+			return delegate.convert(source);
+		}
+	}
 }

src/reference/asciidoc/kafka.adoc

@@ -1271,6 +1271,20 @@ public void listen(List<Foo> foos, @Header(KafkaHeaders.OFFSET) List<Long> offse
 
 Notice that you can still access the batch headers too.
 
+Starting with _versions 2.1.1_, the `org.springframework.core.convert.ConversionService` used by the default
+`org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory` to reslove parameters for the invocation
+of a listener method is supplied with all beans implementing any of the following interfaces:
+
+ - `org.springframework.core.convert.converter.Converter`
+ - `org.springframework.core.convert.converter.GenericConverter`
+ - `org.springframework.format.Formatter`
+
+This allows you to further customize listener deserialization without changing the default configuration for
+ `ConsumerFactory` and `KafkaListenerContainerFactory`.
+
+IMPORTANT: Setting a custom `MessageHandlerMethodFactory` on the `KafkaListenerEndpointRegistrar` through a
+`KafkaListenerConfigurer` bean will disable this feature.
+
 [[headers]]
 ==== Message Headers