Start building against Spring Kafka 3.0.3 snapshots

See gh-34153

Author: wilkinsona

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2022 the original author or authors.
+ * Copyright 2012-2023 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.
@@ -29,7 +29,8 @@
 import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.listener.RecordInterceptor;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
-import org.springframework.kafka.support.converter.MessageConverter;
+import org.springframework.kafka.support.converter.BatchMessageConverter;
+import org.springframework.kafka.support.converter.RecordMessageConverter;
 import org.springframework.kafka.transaction.KafkaAwareTransactionManager;
 
 /**
@@ -43,7 +44,9 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer {
 
 	private KafkaProperties properties;
 
-	private MessageConverter messageConverter;
+	private BatchMessageConverter batchMessageConverter;
+
+	private RecordMessageConverter recordMessageConverter;
 
 	private RecordFilterStrategy<Object, Object> recordFilterStrategy;
 
@@ -68,11 +71,19 @@ void setKafkaProperties(KafkaProperties properties) {
 	}
 
 	/**
-	 * Set the {@link MessageConverter} to use.
-	 * @param messageConverter the message converter
+	 * Set the {@link BatchMessageConverter} to use.
+	 * @param batchMessageConverter the message converter
+	 */
+	void setBatchMessageConverter(BatchMessageConverter batchMessageConverter) {
+		this.batchMessageConverter = batchMessageConverter;
+	}
+
+	/**
+	 * Set the {@link RecordMessageConverter} to use.
+	 * @param recordMessageConverter the message converter
 	 */
-	void setMessageConverter(MessageConverter messageConverter) {
-		this.messageConverter = messageConverter;
+	void setRecordMessageConverter(RecordMessageConverter recordMessageConverter) {
+		this.recordMessageConverter = recordMessageConverter;
 	}
 
 	/**
@@ -151,7 +162,8 @@ private void configureListenerFactory(ConcurrentKafkaListenerContainerFactory<Ob
 		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
 		Listener properties = this.properties.getListener();
 		map.from(properties::getConcurrency).to(factory::setConcurrency);
-		map.from(this.messageConverter).to(factory::setMessageConverter);
+		map.from(this.batchMessageConverter).to(factory::setBatchMessageConverter);
+		map.from(this.recordMessageConverter).to(factory::setRecordMessageConverter);
 		map.from(this.recordFilterStrategy).to(factory::setRecordFilterStrategy);
 		map.from(this.replyTemplate).to(factory::setReplyTemplate);
 		if (properties.getType().equals(Listener.Type.BATCH)) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2022 the original author or authors.
+ * Copyright 2012-2023 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.
@@ -19,7 +19,6 @@
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.autoconfigure.kafka.KafkaProperties.Listener.Type;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.annotation.EnableKafka;
@@ -35,7 +34,6 @@
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
 import org.springframework.kafka.support.converter.BatchMessageConverter;
 import org.springframework.kafka.support.converter.BatchMessagingMessageConverter;
-import org.springframework.kafka.support.converter.MessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
 import org.springframework.kafka.transaction.KafkaAwareTransactionManager;
 
@@ -51,7 +49,7 @@ class KafkaAnnotationDrivenConfiguration {
 
 	private final KafkaProperties properties;
 
-	private final RecordMessageConverter messageConverter;
+	private final RecordMessageConverter recordMessageConverter;
 
 	private final RecordFilterStrategy<Object, Object> recordFilterStrategy;
 
@@ -70,7 +68,7 @@ class KafkaAnnotationDrivenConfiguration {
 	private final RecordInterceptor<Object, Object> recordInterceptor;
 
 	KafkaAnnotationDrivenConfiguration(KafkaProperties properties,
-			ObjectProvider<RecordMessageConverter> messageConverter,
+			ObjectProvider<RecordMessageConverter> recordMessageConverter,
 			ObjectProvider<RecordFilterStrategy<Object, Object>> recordFilterStrategy,
 			ObjectProvider<BatchMessageConverter> batchMessageConverter,
 			ObjectProvider<KafkaTemplate<Object, Object>> kafkaTemplate,
@@ -80,10 +78,10 @@ class KafkaAnnotationDrivenConfiguration {
 			ObjectProvider<AfterRollbackProcessor<Object, Object>> afterRollbackProcessor,
 			ObjectProvider<RecordInterceptor<Object, Object>> recordInterceptor) {
 		this.properties = properties;
-		this.messageConverter = messageConverter.getIfUnique();
+		this.recordMessageConverter = recordMessageConverter.getIfUnique();
 		this.recordFilterStrategy = recordFilterStrategy.getIfUnique();
 		this.batchMessageConverter = batchMessageConverter
-				.getIfUnique(() -> new BatchMessagingMessageConverter(this.messageConverter));
+				.getIfUnique(() -> new BatchMessagingMessageConverter(this.recordMessageConverter));
 		this.kafkaTemplate = kafkaTemplate.getIfUnique();
 		this.transactionManager = kafkaTransactionManager.getIfUnique();
 		this.rebalanceListener = rebalanceListener.getIfUnique();
@@ -97,9 +95,8 @@ class KafkaAnnotationDrivenConfiguration {
 	ConcurrentKafkaListenerContainerFactoryConfigurer kafkaListenerContainerFactoryConfigurer() {
 		ConcurrentKafkaListenerContainerFactoryConfigurer configurer = new ConcurrentKafkaListenerContainerFactoryConfigurer();
 		configurer.setKafkaProperties(this.properties);
-		MessageConverter messageConverterToUse = (this.properties.getListener().getType().equals(Type.BATCH))
-				? this.batchMessageConverter : this.messageConverter;
-		configurer.setMessageConverter(messageConverterToUse);
+		configurer.setBatchMessageConverter(this.batchMessageConverter);
+		configurer.setRecordMessageConverter(this.recordMessageConverter);
 		configurer.setRecordFilterStrategy(this.recordFilterStrategy);
 		configurer.setReplyTemplate(this.kafkaTemplate);
 		configurer.setTransactionManager(this.transactionManager);

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

@@ -519,7 +519,7 @@ void testConcurrentKafkaListenerContainerFactoryWithCustomMessageConverter() {
 		this.contextRunner.withUserConfiguration(MessageConverterConfiguration.class).run((context) -> {
 			ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory = context
 					.getBean(ConcurrentKafkaListenerContainerFactory.class);
-			assertThat(kafkaListenerContainerFactory).hasFieldOrPropertyWithValue("messageConverter",
+			assertThat(kafkaListenerContainerFactory).hasFieldOrPropertyWithValue("recordMessageConverter",
 					context.getBean("myMessageConverter"));
 		});
 	}
@@ -531,7 +531,7 @@ void testConcurrentKafkaListenerContainerFactoryInBatchModeWithCustomMessageConv
 				.withPropertyValues("spring.kafka.listener.type=batch").run((context) -> {
 					ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory = context
 							.getBean(ConcurrentKafkaListenerContainerFactory.class);
-					assertThat(kafkaListenerContainerFactory).hasFieldOrPropertyWithValue("messageConverter",
+					assertThat(kafkaListenerContainerFactory).hasFieldOrPropertyWithValue("batchMessageConverter",
 							context.getBean("myBatchMessageConverter"));
 				});
 	}
@@ -543,7 +543,7 @@ void testConcurrentKafkaListenerContainerFactoryInBatchModeWrapsCustomMessageCon
 					ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory = context
 							.getBean(ConcurrentKafkaListenerContainerFactory.class);
 					Object messageConverter = ReflectionTestUtils.getField(kafkaListenerContainerFactory,
-							"messageConverter");
+							"batchMessageConverter");
 					assertThat(messageConverter).isInstanceOf(BatchMessagingMessageConverter.class);
 					assertThat(((BatchMessageConverter) messageConverter).getRecordMessageConverter())
 							.isSameAs(context.getBean("myMessageConverter"));
@@ -555,7 +555,8 @@ void testConcurrentKafkaListenerContainerFactoryInBatchModeWithNoMessageConverte
 		this.contextRunner.withPropertyValues("spring.kafka.listener.type=batch").run((context) -> {
 			ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory = context
 					.getBean(ConcurrentKafkaListenerContainerFactory.class);
-			Object messageConverter = ReflectionTestUtils.getField(kafkaListenerContainerFactory, "messageConverter");
+			Object messageConverter = ReflectionTestUtils.getField(kafkaListenerContainerFactory,
+					"batchMessageConverter");
 			assertThat(messageConverter).isInstanceOf(BatchMessagingMessageConverter.class);
 			assertThat(((BatchMessageConverter) messageConverter).getRecordMessageConverter()).isNull();
 		});

spring-boot-project/spring-boot-dependencies/build.gradle

@@ -1384,7 +1384,7 @@ bom {
 			]
 		}
 	}
-	library("Spring Kafka", "3.0.2") {
+	library("Spring Kafka", "3.0.3-SNAPSHOT") {
 		group("org.springframework.kafka") {
 			modules = [
 				"spring-kafka",