Fix SmartMessageConverter support in batch listeners

Fixes GH-4097

Problem:
When using `@KafkaListener(contentTypeConverter = "...")` with a batch listener
(`batch = "true"`), the SmartMessageConverter was not being applied during
message conversion, resulting in ClassCastException when trying to process
messages that required conversion (e.g., byte[] to String).

Root Cause:
The issue stemmed from two problems:
1. `BatchMessagingMessageConverter` default constructor created an instance
   without a `RecordMessageConverter` (`this(null)`), preventing proper
   per-record conversion within batch processing.
2. `BatchMessagingMessageListenerAdapter` did not override
   `setMessagingConverter()` to propagate the SmartMessageConverter from the
   annotation configuration to the batch converter's record converter.

Additionally, the initial approach of bypassing the parent's validation in
`setMessagingConverter()` conflicted with the framework's design to prevent
the "paradox of choice" - users should configure converters either via factory
setter OR annotation attribute, not both.

Solution:
1. Changed `BatchMessagingMessageConverter` default constructor to always
   create a `MessagingMessageConverter` by default (`this(new MessagingMessageConverter())`).
   This ensures batch converters always have a record converter for per-record
   conversion within batches, and enables annotation-only configuration without
   requiring users to set converters on the factory.

2. Added `setMessagingConverter()` method to `BatchMessagingMessageConverter`
   that propagates the SmartMessageConverter to its internal record converter
   when it's an instance of `MessagingMessageConverter`.

3. Overrode `setMessagingConverter()` in `BatchMessagingMessageListenerAdapter`
   to:
   - Call `super.setMessagingConverter(messageConverter)` first, which applies
     the same validation as the parent class (`Assert.isTrue(!this.converterSet)`)
     to prevent configuration conflicts between factory setter and annotation.
   - Propagate the SmartMessageConverter to the batch converter's record converter
     for proper message conversion in batch processing.

This approach:
- Respects the framework's validation to prevent the "paradox of choice"
- Ensures both batch and per-record conversion paths work correctly
- Allows users to configure converters via annotation without needing to set
  converters on the factory (annotation-only configuration)
- Avoids the complexity and potential issues of cloning converters
- Works for multiple listeners with different contentTypeConverter values

Testing:
Added comprehensive integration test `BatchSmartMessageConverterTests` that:
- Verifies SmartMessageConverter works with batch listeners using
  `@KafkaListener(contentTypeConverter = "...")`
- Tests multiple listeners with different converters to ensure isolation
- Uses `ConcurrentKafkaListenerContainerFactory` with batch mode to verify
  annotation attributes propagate correctly to the record converter

Signed-off-by: Jujuwryy <[email protected]>

Author: Jujuwryy

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java

@@ -114,19 +114,21 @@ public void setBatchToRecordAdapter(BatchToRecordAdapter<K, V> batchToRecordAdap
 	 * <p>
 	 * When a {@code SmartMessageConverter} is configured via
 	 * {@code @KafkaListener(contentTypeConverter = "...")}, this method ensures it is
-	 * properly propagated to the batch converter, which will then propagate it to the
-	 * record converter for message conversion in batch listeners.
+	 * properly propagated to the batch converter's record converter for message conversion
+	 * in batch listeners.
 	 * <p>
-	 * This override does not call the parent implementation because the parent's validation
-	 * (checking {@code converterSet}) blocks setting the SmartMessageConverter after
-	 * {@code setBatchMessageConverter} has been called, which is the normal workflow for
-	 * batch listeners.
+	 * Uses the same validation as the parent class to prevent the paradox of choice:
+	 * not allowed when a custom {@link #setBatchMessageConverter(BatchMessageConverter)
+	 * batchMessageConverter} is provided. Since {@link BatchMessagingMessageConverter} now
+	 * always has a default {@link org.springframework.kafka.support.converter.MessagingMessageConverter},
+	 * users can configure the converter via the annotation without needing to set it on the factory.
 	 * @param messageConverter the converter to set
 	 */
 	@Override
 	public void setMessagingConverter(SmartMessageConverter messageConverter) {
-		if (this.batchMessageConverter instanceof BatchMessagingMessageConverter messagingConverter) {
-			messagingConverter.setMessagingConverter(messageConverter);
+		super.setMessagingConverter(messageConverter);
+		if (this.batchMessageConverter instanceof BatchMessagingMessageConverter batchConverter) {
+			batchConverter.setMessagingConverter(messageConverter);
 		}
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java

@@ -89,10 +89,11 @@ public class BatchMessagingMessageConverter implements BatchMessageConverter {
 	private boolean rawRecordHeader;
 
 	/**
-	 * Create an instance that does not convert the record values.
+	 * Create an instance with a default {@link MessagingMessageConverter} for record conversion.
+	 * @since 3.3.11
 	 */
 	public BatchMessagingMessageConverter() {
-		this(null);
+		this(new MessagingMessageConverter());
 	}
 
 	/**

spring-kafka/src/test/java/org/springframework/kafka/listener/BatchSmartMessageConverterTests.java

@@ -41,8 +41,6 @@
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
-import org.springframework.kafka.support.converter.BatchMessagingMessageConverter;
-import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -64,7 +62,7 @@
  */
 @SpringJUnitConfig
 @DirtiesContext
-@EmbeddedKafka(partitions = 1, topics = { "smartBatchTopic" })
+@EmbeddedKafka(partitions = 1, topics = { "smartBatchTopic", "smartBatchTopic2" })
 class BatchSmartMessageConverterTests {
 
 	@Autowired
@@ -75,18 +73,32 @@ class BatchSmartMessageConverterTests {
 
 	@Test
 	void testContentTypeConverterWithBatchListener() throws Exception {
-		// Given: A batch listener with contentTypeConverter configured
 		BatchListener listener = this.config.batchListener();
+		listener.reset(2);
 
-		// When: Send byte[] messages that should be converted to String
 		this.template.send("smartBatchTopic", "hello".getBytes());
 		this.template.send("smartBatchTopic", "world".getBytes());
 
-		// Then: SmartMessageConverter should convert byte[] to String for batch listener
 		assertThat(listener.latch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(listener.received).hasSize(2).containsExactly("hello", "world");
 	}
 
+	@Test
+	void testMultipleListenersWithDifferentConverters() throws Exception {
+		BatchListener listener1 = this.config.batchListener();
+		BatchListener2 listener2 = this.config.batchListener2();
+		listener1.reset(1);
+		listener2.reset(1);
+
+		this.template.send("smartBatchTopic", "foo".getBytes());
+		this.template.send("smartBatchTopic2", "bar".getBytes());
+
+		assertThat(listener1.latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(listener2.latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(listener1.received).hasSize(1).containsExactly("foo");
+		assertThat(listener2.received).hasSize(1).containsExactly("BAR");
+	}
+
 	@Configuration
 	@EnableKafka
 	public static class Config {
@@ -97,8 +109,6 @@ public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory(EmbeddedKa
 					new ConcurrentKafkaListenerContainerFactory<>();
 			factory.setConsumerFactory(consumerFactory(embeddedKafka));
 			factory.setBatchListener(true);
-			// Set up batch converter with record converter - framework will propagate SmartMessageConverter
-			factory.setBatchMessageConverter(new BatchMessagingMessageConverter(new MessagingMessageConverter()));
 			return factory;
 		}
 
@@ -136,19 +146,29 @@ public Map<String, Object> producerConfigs(EmbeddedKafkaBroker embeddedKafka) {
 
 		@Bean
 		public SmartMessageConverter byteArrayToStringConverter() {
-			return new ByteArrayToStringConverter();
+			return new ByteArrayConverter(bytes -> new String(bytes));
+		}
+
+		@Bean
+		public SmartMessageConverter byteArrayToUpperCaseConverter() {
+			return new ByteArrayConverter(bytes -> new String(bytes).toUpperCase());
 		}
 
 		@Bean
 		public BatchListener batchListener() {
 			return new BatchListener();
 		}
 
+		@Bean
+		public BatchListener2 batchListener2() {
+			return new BatchListener2();
+		}
+
 	}
 
 	public static class BatchListener {
 
-		private final CountDownLatch latch = new CountDownLatch(2);
+		private CountDownLatch latch = new CountDownLatch(2);
 
 		private final List<String> received = new ArrayList<>();
 
@@ -166,17 +186,55 @@ public void listen(List<String> messages) {
 			});
 		}
 
+		void reset(int expectedCount) {
+			this.received.clear();
+			this.latch = new CountDownLatch(expectedCount);
+		}
+
+	}
+
+	public static class BatchListener2 {
+
+		private CountDownLatch latch = new CountDownLatch(1);
+
+		private final List<String> received = new ArrayList<>();
+
+		@KafkaListener(
+				id = "batchSmartListener2",
+				topics = "smartBatchTopic2",
+				groupId = "smartBatchGroup2",
+				contentTypeConverter = "byteArrayToUpperCaseConverter",
+				batch = "true"
+		)
+		public void listen(List<String> messages) {
+			messages.forEach(message -> {
+				this.received.add(message);
+				this.latch.countDown();
+			});
+		}
+
+		void reset(int expectedCount) {
+			this.received.clear();
+			this.latch = new CountDownLatch(expectedCount);
+		}
+
 	}
 
 	/**
-	 * Simple SmartMessageConverter for testing that converts byte[] to String.
+	 * Simple SmartMessageConverter for testing that converts byte[] to String using a function.
 	 */
-	static class ByteArrayToStringConverter implements SmartMessageConverter {
+	static class ByteArrayConverter implements SmartMessageConverter {
+
+		private final java.util.function.Function<byte[], String> converter;
+
+		ByteArrayConverter(java.util.function.Function<byte[], String> converter) {
+			this.converter = converter;
+		}
 
 		@Override
 		public Object fromMessage(Message<?> message, Class<?> targetClass) {
 			Object payload = message.getPayload();
-			return (payload instanceof byte[] bytes) ? new String(bytes) : payload;
+			return (payload instanceof byte[] bytes) ? this.converter.apply(bytes) : payload;
 		}
 
 		@Override