Merge pull request #10380 from eddumelendez:gh-10376

* pr/10380:
  Polish "Auto-configure Kafka MessageConverter"
  Auto-configure Kafka MessageConverter

spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java

@@ -20,17 +20,21 @@ import org.springframework.boot.autoconfigure.kafka.KafkaProperties.Listener;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.listener.config.ContainerProperties;
+import org.springframework.kafka.support.converter.RecordMessageConverter;
 
 /**
  * Configure {@link ConcurrentKafkaListenerContainerFactory} with sensible defaults.
  *
  * @author Gary Russell
+ * @author Eddú Meléndez
  * @since 1.5.0
  */
 public class ConcurrentKafkaListenerContainerFactoryConfigurer {
 
 	private KafkaProperties properties;
 
+	private RecordMessageConverter messageConverter;
+
 	/**
 	 * Set the {@link KafkaProperties} to use.
 	 * @param properties the properties
@@ -39,6 +43,14 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer {
 		this.properties = properties;
 	}
 
+	/**
+	 * Set the {@link RecordMessageConverter} to use.
+	 * @param messageConverter the message converter
+	 */
+	void setMessageConverter(RecordMessageConverter messageConverter) {
+		this.messageConverter = messageConverter;
+	}
+
 	/**
 	 * Configure the specified Kafka listener container factory. The factory can be
 	 * further tuned and default settings can be overridden.
@@ -50,6 +62,9 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer {
 			ConcurrentKafkaListenerContainerFactory<Object, Object> listenerContainerFactory,
 			ConsumerFactory<Object, Object> consumerFactory) {
 		listenerContainerFactory.setConsumerFactory(consumerFactory);
+		if (this.messageConverter != null) {
+			listenerContainerFactory.setMessageConverter(this.messageConverter);
+		}
 		Listener container = this.properties.getListener();
 		ContainerProperties containerProperties = listenerContainerFactory
 				.getContainerProperties();

spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java

@@ -16,6 +16,7 @@
 
 package org.springframework.boot.autoconfigure.kafka;
 
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
@@ -24,11 +25,13 @@ import org.springframework.kafka.annotation.EnableKafka;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerConfigUtils;
 import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.support.converter.RecordMessageConverter;
 
 /**
  * Configuration for Kafka annotation-driven support.
  *
  * @author Gary Russell
+ * @author Eddú Meléndez
  * @since 1.5.0
  */
 @Configuration
@@ -37,8 +40,12 @@ class KafkaAnnotationDrivenConfiguration {
 
 	private final KafkaProperties properties;
 
-	KafkaAnnotationDrivenConfiguration(KafkaProperties properties) {
+	private final RecordMessageConverter messageConverter;
+
+	KafkaAnnotationDrivenConfiguration(KafkaProperties properties,
+			ObjectProvider<RecordMessageConverter> messageConverter) {
 		this.properties = properties;
+		this.messageConverter = messageConverter.getIfUnique();
 	}
 
 	@Bean
@@ -46,6 +53,7 @@ class KafkaAnnotationDrivenConfiguration {
 	public ConcurrentKafkaListenerContainerFactoryConfigurer kafkaListenerContainerFactoryConfigurer() {
 		ConcurrentKafkaListenerContainerFactoryConfigurer configurer = new ConcurrentKafkaListenerContainerFactoryConfigurer();
 		configurer.setKafkaProperties(this.properties);
+		configurer.setMessageConverter(this.messageConverter);
 		return configurer;
 	}
 

spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfiguration.java

@@ -18,6 +18,7 @@ package org.springframework.boot.autoconfigure.kafka;
 
 import java.io.IOException;
 
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@@ -36,12 +37,14 @@ import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer;
 import org.springframework.kafka.support.LoggingProducerListener;
 import org.springframework.kafka.support.ProducerListener;
+import org.springframework.kafka.support.converter.RecordMessageConverter;
 
 /**
  * {@link EnableAutoConfiguration Auto-configuration} for Apache Kafka.
  *
  * @author Gary Russell
  * @author Stephane Nicoll
+ * @author Eddú Meléndez
  * @since 1.5.0
  */
 @Configuration
@@ -52,8 +55,12 @@ public class KafkaAutoConfiguration {
 
 	private final KafkaProperties properties;
 
-	public KafkaAutoConfiguration(KafkaProperties properties) {
+	private final RecordMessageConverter messageConverter;
+
+	public KafkaAutoConfiguration(KafkaProperties properties,
+			ObjectProvider<RecordMessageConverter> messageConverter) {
 		this.properties = properties;
+		this.messageConverter = messageConverter.getIfUnique();
 	}
 
 	@Bean
@@ -63,6 +70,9 @@ public class KafkaAutoConfiguration {
 			ProducerListener<Object, Object> kafkaProducerListener) {
 		KafkaTemplate<Object, Object> kafkaTemplate = new KafkaTemplate<>(
 				kafkaProducerFactory);
+		if (this.messageConverter != null) {
+			kafkaTemplate.setMessageConverter(this.messageConverter);
+		}
 		kafkaTemplate.setProducerListener(kafkaProducerListener);
 		kafkaTemplate.setDefaultTopic(this.properties.getTemplate().getDefaultTopic());
 		return kafkaTemplate;

spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java

@@ -30,12 +30,14 @@ import org.apache.kafka.common.serialization.IntegerDeserializer;
 import org.apache.kafka.common.serialization.IntegerSerializer;
 import org.apache.kafka.common.serialization.LongDeserializer;
 import org.apache.kafka.common.serialization.LongSerializer;
-import org.junit.After;
 import org.junit.Test;
 
 import org.springframework.beans.DirectFieldAccessor;
-import org.springframework.boot.test.util.TestPropertyValues;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
-import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
@@ -43,31 +45,33 @@ import org.springframework.kafka.core.KafkaAdmin;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
 import org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer;
+import org.springframework.kafka.support.converter.MessagingMessageConverter;
+import org.springframework.kafka.support.converter.RecordMessageConverter;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.entry;
+import static org.mockito.Mockito.mock;
 
 /**
  * Tests for {@link KafkaAutoConfiguration}.
  *
  * @author Gary Russell
  * @author Stephane Nicoll
+ * @author Eddú Meléndez
  */
 public class KafkaAutoConfigurationTests {
 
-	private AnnotationConfigApplicationContext context;
+	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
-
+			.withConfiguration(AutoConfigurations.of(KafkaAutoConfiguration.class));
-	@After
-	public void closeContext() {
-		if (this.context != null) {
-			this.context.close();
-		}
-	}
 
 	@Test
 	public void consumerProperties() {
-		load("spring.kafka.bootstrap-servers=foo:1234", "spring.kafka.properties.foo=bar",
+		this.contextRunner
+				.withUserConfiguration(TestConfiguration.class)
+				.withPropertyValues(
+						"spring.kafka.bootstrap-servers=foo:1234",
+						"spring.kafka.properties.foo=bar",
 						"spring.kafka.properties.baz=qux",
 						"spring.kafka.properties.foo.bar.baz=qux.fiz.buz",
 						"spring.kafka.ssl.key-password=p1",
@@ -86,8 +90,9 @@ public class KafkaAutoConfigurationTests {
 						"spring.kafka.consumer.group-id=bar",
 						"spring.kafka.consumer.heartbeat-interval=234",
 						"spring.kafka.consumer.key-deserializer = org.apache.kafka.common.serialization.LongDeserializer",
-				"spring.kafka.consumer.value-deserializer = org.apache.kafka.common.serialization.IntegerDeserializer");
+						"spring.kafka.consumer.value-deserializer = org.apache.kafka.common.serialization.IntegerDeserializer"
-		DefaultKafkaConsumerFactory<?, ?> consumerFactory = this.context
+				).run((context) -> {
+			DefaultKafkaConsumerFactory<?, ?> consumerFactory = context
 					.getBean(DefaultKafkaConsumerFactory.class);
 			Map<String, Object> configs = consumerFactory.getConfigurationProperties();
 			// common
@@ -123,14 +128,20 @@ public class KafkaAutoConfigurationTests {
 			assertThat(configs.get("baz")).isEqualTo("qux");
 			assertThat(configs.get("foo.bar.baz")).isEqualTo("qux.fiz.buz");
 			assertThat(configs.get("fiz.buz")).isEqualTo("fix.fox");
+		});
 	}
 
 	@Test
 	public void producerProperties() {
-		load("spring.kafka.clientId=cid",
+		this.contextRunner
+				.withUserConfiguration(TestConfiguration.class)
+				.withPropertyValues(
+						"spring.kafka.clientId=cid",
 						"spring.kafka.properties.foo.bar.baz=qux.fiz.buz",
-				"spring.kafka.producer.acks=all", "spring.kafka.producer.batch-size=20",
+						"spring.kafka.producer.acks=all",
-				"spring.kafka.producer.bootstrap-servers=bar:1234", // test override
+						"spring.kafka.producer.batch-size=20",
+						"spring.kafka.producer.bootstrap-servers=bar:1234", // test
+						// override
 						"spring.kafka.producer.buffer-memory=12345",
 						"spring.kafka.producer.compression-type=gzip",
 						"spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.LongSerializer",
@@ -141,8 +152,9 @@ public class KafkaAutoConfigurationTests {
 						"spring.kafka.producer.ssl.keystore-password=p5",
 						"spring.kafka.producer.ssl.truststore-location=classpath:tsLocP",
 						"spring.kafka.producer.ssl.truststore-password=p6",
-				"spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.IntegerSerializer");
+						"spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.IntegerSerializer"
-		DefaultKafkaProducerFactory<?, ?> producerFactory = this.context
+				).run((context) -> {
+			DefaultKafkaProducerFactory<?, ?> producerFactory = context
 					.getBean(DefaultKafkaProducerFactory.class);
 			Map<String, Object> configs = producerFactory.getConfigurationProperties();
 			// common
@@ -167,15 +179,16 @@ public class KafkaAutoConfigurationTests {
 			assertThat(configs.get(ProducerConfig.RETRIES_CONFIG)).isEqualTo(2);
 			assertThat(configs.get(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG))
 					.isEqualTo(IntegerSerializer.class);
-		assertThat(this.context.getBeansOfType(KafkaJaasLoginModuleInitializer.class))
+			assertThat(context.getBeansOfType(KafkaJaasLoginModuleInitializer.class))
 					.isEmpty();
 			assertThat(configs.get("foo.bar.baz")).isEqualTo("qux.fiz.buz");
 			assertThat(configs.get("fiz.buz")).isEqualTo("fix.fox");
+		});
 	}
 
 	@Test
 	public void adminProperties() {
-		load("spring.kafka.clientId=cid",
+		this.contextRunner.withPropertyValues("spring.kafka.clientId=cid",
 				"spring.kafka.properties.foo.bar.baz=qux.fiz.buz",
 				"spring.kafka.admin.fail-fast=true",
 				"spring.kafka.admin.properties.fiz.buz=fix.fox",
@@ -183,8 +196,8 @@ public class KafkaAutoConfigurationTests {
 				"spring.kafka.admin.ssl.keystore-location=classpath:ksLocP",
 				"spring.kafka.admin.ssl.keystore-password=p5",
 				"spring.kafka.admin.ssl.truststore-location=classpath:tsLocP",
-				"spring.kafka.admin.ssl.truststore-password=p6");
+				"spring.kafka.admin.ssl.truststore-password=p6").run((context) -> {
-		KafkaAdmin admin = this.context.getBean(KafkaAdmin.class);
+			KafkaAdmin admin = context.getBean(KafkaAdmin.class);
 			Map<String, Object> configs = admin.getConfig();
 			// common
 			assertThat(configs.get(AdminClientConfig.CLIENT_ID_CONFIG)).isEqualTo("cid");
@@ -197,34 +210,40 @@ public class KafkaAutoConfigurationTests {
 					.endsWith(File.separator + "tsLocP");
 			assertThat(configs.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG))
 					.isEqualTo("p6");
-		assertThat(this.context.getBeansOfType(KafkaJaasLoginModuleInitializer.class))
+			assertThat(context.getBeansOfType(KafkaJaasLoginModuleInitializer.class))
 					.isEmpty();
 			assertThat(configs.get("foo.bar.baz")).isEqualTo("qux.fiz.buz");
 			assertThat(configs.get("fiz.buz")).isEqualTo("fix.fox");
 			assertThat(KafkaTestUtils.getPropertyValue(admin, "fatalIfBrokerNotAvailable",
 					Boolean.class)).isTrue();
+		});
 	}
 
 	@SuppressWarnings("unchecked")
 	@Test
 	public void listenerProperties() {
-		load("spring.kafka.template.default-topic=testTopic",
+		this.contextRunner.withUserConfiguration(TestConfiguration.class)
+				.withPropertyValues("spring.kafka.template.default-topic=testTopic",
 						"spring.kafka.listener.ack-mode=MANUAL",
 						"spring.kafka.listener.ack-count=123",
 						"spring.kafka.listener.ack-time=456",
 						"spring.kafka.listener.concurrency=3",
 						"spring.kafka.listener.poll-timeout=2000",
-				"spring.kafka.listener.type=batch", "spring.kafka.jaas.enabled=true",
+						"spring.kafka.listener.type=batch",
+						"spring.kafka.jaas.enabled=true",
 						"spring.kafka.jaas.login-module=foo",
 						"spring.kafka.jaas.control-flag=REQUISITE",
-				"spring.kafka.jaas.options.useKeyTab=true");
+						"spring.kafka.jaas.options.useKeyTab=true"
-		DefaultKafkaProducerFactory<?, ?> producerFactory = this.context
+				).run((context) -> {
+			DefaultKafkaProducerFactory<?, ?> producerFactory = context
 					.getBean(DefaultKafkaProducerFactory.class);
-		DefaultKafkaConsumerFactory<?, ?> consumerFactory = this.context
+			DefaultKafkaConsumerFactory<?, ?> consumerFactory = context
 					.getBean(DefaultKafkaConsumerFactory.class);
-		KafkaTemplate<?, ?> kafkaTemplate = this.context.getBean(KafkaTemplate.class);
+			KafkaTemplate<?, ?> kafkaTemplate = context.getBean(KafkaTemplate.class);
-		KafkaListenerContainerFactory<?> kafkaListenerContainerFactory = this.context
+			KafkaListenerContainerFactory<?> kafkaListenerContainerFactory = context
 					.getBean(KafkaListenerContainerFactory.class);
+			assertThat(kafkaTemplate.getMessageConverter()).isInstanceOf(
+					MessagingMessageConverter.class);
 			assertThat(new DirectFieldAccessor(kafkaTemplate)
 					.getPropertyValue("producerFactory")).isEqualTo(producerFactory);
 			assertThat(kafkaTemplate.getDefaultTopic()).isEqualTo("testTopic");
@@ -238,9 +257,9 @@ public class KafkaAutoConfigurationTests {
 			assertThat(dfa.getPropertyValue("containerProperties.pollTimeout"))
 					.isEqualTo(2000L);
 			assertThat(dfa.getPropertyValue("batchListener")).isEqualTo(true);
-		assertThat(this.context.getBeansOfType(KafkaJaasLoginModuleInitializer.class))
+			assertThat(context.getBeansOfType(KafkaJaasLoginModuleInitializer.class))
 					.hasSize(1);
-		KafkaJaasLoginModuleInitializer jaas = this.context
+			KafkaJaasLoginModuleInitializer jaas = context
 					.getBean(KafkaJaasLoginModuleInitializer.class);
 			dfa = new DirectFieldAccessor(jaas);
 			assertThat(dfa.getPropertyValue("loginModule")).isEqualTo("foo");
@@ -248,14 +267,45 @@ public class KafkaAutoConfigurationTests {
 					.isEqualTo(AppConfigurationEntry.LoginModuleControlFlag.REQUISITE);
 			assertThat(((Map<String, String>) dfa.getPropertyValue("options")))
 					.containsExactly(entry("useKeyTab", "true"));
+		});
+	}
+
+	@Test
+	public void testKafkaTemplateRecordMessageConverters() {
+		this.contextRunner.withUserConfiguration(MessageConverterConfiguration.class)
+				.run((context) -> {
+					KafkaTemplate kafkaTemplate = context.getBean(KafkaTemplate.class);
+					assertThat(kafkaTemplate.getMessageConverter())
+							.isSameAs(context.getBean("myMessageConverter"));
+				});
+	}
+
+	@Test
+	public void testConcurrentKafkaListenerContainerFactoryWithCustomMessageConverters() {
+		this.contextRunner.withUserConfiguration(MessageConverterConfiguration.class)
+				.run((context) -> {
+					ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory = context
+							.getBean(ConcurrentKafkaListenerContainerFactory.class);
+					DirectFieldAccessor dfa = new DirectFieldAccessor(
+							kafkaListenerContainerFactory);
+					assertThat(dfa.getPropertyValue("messageConverter"))
+							.isSameAs(context.getBean("myMessageConverter"));
+				});
+	}
+
+	@Configuration
+	protected static class TestConfiguration {
+
+	}
+
+	@Configuration
+	protected static class MessageConverterConfiguration {
+
+		@Bean
+		public RecordMessageConverter myMessageConverter() {
+			return mock(RecordMessageConverter.class);
 		}
 
-	private void load(String... environment) {
-		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext();
-		ctx.register(KafkaAutoConfiguration.class);
-		TestPropertyValues.of(environment).applyTo(ctx);
-		ctx.refresh();
-		this.context = ctx;
 	}
 
 }

spring-boot-docs/src/main/asciidoc/spring-boot-features.adoc

@@ -4846,6 +4846,8 @@ public class MyBean {
 }
 ----
 
+NOTE: If a `RecordMessageConverter` bean is defined, it is associated automatically to the
+auto-configured `KafkaTemplate`.
 
 
 [[boot-features-kafka-receiving-a-message]]
@@ -4853,7 +4855,8 @@ public class MyBean {
 When the Apache Kafka infrastructure is present, any bean can be annotated with
 `@KafkaListener` to create a listener endpoint. If no `KafkaListenerContainerFactory`
 has been defined, a default one is configured automatically with keys defined in
-`spring.kafka.listener.*`.
+`spring.kafka.listener.*`. Also, if a `RecordMessageConverter` bean is defined, it is
+associated automatically to the default factory.
 
 The following component creates a listener endpoint on the `someTopic` topic: