GH-4097: Fix SmartMessageConverter support in batch listeners

Fixes: #4097

`BatchMessagingMessageConverter` was missing `setMessagingConverter()` method that exists in `MessagingMessageConverter`, causing `SmartMessageConverter` configured via `@KafkaListener(contentTypeConverter)` to be ignored in batch listeners.

This inconsistency between regular and batch listeners leads to `ClassCastException` when `byte[]` values aren't converted to the expected `String` type, breaking the contract that `SmartMessageConverter` should work the same way regardless of listener type.

The fix ensures `SmartMessageConverter` propagation works consistently by:
- Adding setMessagingConverter() to `BatchMessagingMessageConverter` that delegates to underlying `MessagingMessageConverter`
- Overriding setMessagingConverter() in `BatchMessagingMessageListenerAdapter` to propagate the converter to batch converter
- Maintaining the same `SmartMessageConverter` behavior between regular and batch listeners

Integration testing revealed the root cause of GH-4097. When Spring processes
a `@KafkaListener` with `contentTypeConverter` and `batch="true"`, the framework:

1. Calls `setBatchMessageConverter()` on the adapter
2. This internally calls `setMessageConverter()` which sets `converterSet=true`
3. Spring then tries to apply `contentTypeConverter` by calling `setMessagingConverter()`
4. The parent's validation `Assert.isTrue(!this.converterSet, ...)` blocks this

The unit test didn't catch this because it bypassed the adapter and Spring
framework integration entirely.

Changes:
- `BatchMessagingMessageListenerAdapter.setMessagingConverter()`: Override now
  directly applies `SmartMessageConverter` to batch converter (which propagates
  to record converter) without calling super, bypassing the validation that
  doesn't apply to the batch listener workflow

- `BatchSmartMessageConverterTests`: Replaced unit test with full integration
  test using `@SpringJUnitConfig`, `@EmbeddedKafka`, `ConcurrentKafkaListenerContainerFactory`,
  and `@KafkaListener` to verify the complete framework flow

- Added minimal `ByteArrayToStringConverter` (24 lines) for testing as no
  existing Spring Framework converter provides simple byte[] to String
  conversion needed for this test scenario

All tests pass and checkstyle validation successful.

- Use pattern matching for instanceof in setMessagingConverter() to avoid explicit casting
- Fix constructor parameter indentation to use tabs only (not mixed spaces)
- Address checkstyle violations per reviewer feedback

These changes improve code readability without affecting functionality.

* Fix remaining indentation issues in method parameters

- Fix `toMessagingMessage()` parameter indentation in `BatchMessagingMessageListenerAdapter`
- Fix `toMessage()` parameter indentation in `BatchMessagingMessageConverter`
- Use single tab indentation consistently per Spring code style

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

* Polish documentation and test code

- Improve Javadoc clarity with technical explanation for method constraints
- Remove redundant phrase from `BatchMessagingMessageConverter` documentation
- Generalize test class description for future extensibility
- Clean up test assertions by removing redundant size checks
- Use descriptive variable names in test data

* Fix Javadoc links in `setMessagingConverter` method

- Add proper links for `SmartMessageConverter` and `BatchMessageConverter`
- Use code formatting for `batchMessageConverter` field reference
- Simplify `setBatchMessageConverter` method link format

Signed-off-by: Jujuwryy <georgemahfoud2@gmail.com>

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

Author: Jujuwryy

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

@@ -32,6 +32,7 @@
 import org.springframework.kafka.support.converter.RecordMessageConverter;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.converter.SmartMessageConverter;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.Assert;
 
@@ -55,6 +56,7 @@
  * @author Venil Noronha
  * @author Wang ZhiYang
  * @author Sanghyeok An
+ * @author George Mahfoud
  * @since 1.1
  */
 public class BatchMessagingMessageListenerAdapter<K, V> extends MessagingMessageListenerAdapter<K, V>
@@ -107,6 +109,30 @@ public void setBatchToRecordAdapter(BatchToRecordAdapter<K, V> batchToRecordAdap
 		this.batchToRecordAdapter = batchToRecordAdapter;
 	}
 
+	/**
+	 * Set the {@link SmartMessageConverter} to use with the batch message converter.
+	 * <p>
+	 * When a {@code SmartMessageConverter} is configured via
+	 * {@code @KafkaListener(contentTypeConverter = "...")}, this method ensures it is
+	 * properly propagated to the batch converter's record converter for message conversion
+	 * in batch listeners.
+	 * <p>
+	 * This method cannot be called after {@link #setBatchMessageConverter(BatchMessageConverter)}
+	 * as it would cause a mutation of the internal {@code batchMessageConverter}. Instead, the
+	 * {@link SmartMessageConverter} has to be provided on the external {@link BatchMessageConverter}.
+	 * 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) {
+		super.setMessagingConverter(messageConverter);
+		if (this.batchMessageConverter instanceof BatchMessagingMessageConverter batchConverter) {
+			batchConverter.setMessagingConverter(messageConverter);
+		}
+	}
+
 	/**
 	 * Return the {@link BatchMessagingMessageConverter} for this listener,
 	 * being able to convert {@link org.springframework.messaging.Message}.

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

@@ -42,6 +42,7 @@
 import org.springframework.kafka.support.serializer.SerializationUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.converter.SmartMessageConverter;
 import org.springframework.messaging.support.MessageBuilder;
 
 /**
@@ -65,6 +66,7 @@
  * @author Hope Kim
  * @author Borahm Lee
  * @author Artem Bilan
+ * @author George Mahfoud
  *
  * @since 1.1
  */
@@ -84,10 +86,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());
 	}
 
 	/**
@@ -136,6 +139,18 @@ public RecordMessageConverter getRecordMessageConverter() {
 		return this.recordConverter;
 	}
 
+	/**
+	 * Set a {@link SmartMessageConverter} to convert the record value to
+	 * the desired type.
+	 * @param messagingConverter the converter.
+	 * @since 3.3.11
+	 */
+	public void setMessagingConverter(@Nullable SmartMessageConverter messagingConverter) {
+		if (this.recordConverter instanceof MessagingMessageConverter messagingRecordConverter) {
+			messagingRecordConverter.setMessagingConverter(messagingConverter);
+		}
+	}
+
 	/**
 	 * Set to true to add the raw {@code List<ConsumerRecord<?, ?>>} as a header
 	 * {@link KafkaHeaders#RAW_DATA}.
@@ -267,14 +282,20 @@ protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, Type type)
 	 * @param type the type - must be a {@link ParameterizedType} with a single generic
 	 * type parameter.
 	 * @param conversionFailures Conversion failures.
-	 * @return the converted payload.
+	 * @return the converted payload, potentially further processed by a {@link SmartMessageConverter}.
 	 */
 	protected Object convert(ConsumerRecord<?, ?> record, Type type, List<ConversionException> conversionFailures) {
 		try {
-			Object payload = this.recordConverter
-					.toMessage(record, null, null, ((ParameterizedType) type).getActualTypeArguments()[0]).getPayload();
-			conversionFailures.add(null);
-			return payload;
+			if (this.recordConverter != null) {
+				Type actualType = ((ParameterizedType) type).getActualTypeArguments()[0];
+				Object payload = this.recordConverter
+						.toMessage(record, null, null, actualType).getPayload();
+				conversionFailures.add(null);
+				return payload;
+			}
+			else {
+				return null;
+			}
 		}
 		catch (ConversionException ex) {
 			byte[] original = null;

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

@@ -0,0 +1,257 @@
+/*
+ * Copyright 2016-present 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.listener;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.config.KafkaListenerContainerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.converter.SmartMessageConverter;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Integration tests for SmartMessageConverter support in batch listeners.
+ *
+ * @author George Mahfoud
+ * @since 3.3.11
+ */
+@SpringJUnitConfig
+@DirtiesContext
+@EmbeddedKafka(partitions = 1, topics = { "smartBatchTopic", "smartBatchTopic2" })
+class BatchSmartMessageConverterTests {
+
+	@Autowired
+	private KafkaTemplate<Integer, byte[]> template;
+
+	@Autowired
+	private Config config;
+
+	@Test
+	void testContentTypeConverterWithBatchListener() throws Exception {
+		BatchListener listener = this.config.batchListener();
+		listener.reset(2);
+
+		this.template.send("smartBatchTopic", "hello".getBytes());
+		this.template.send("smartBatchTopic", "world".getBytes());
+
+		assertThat(listener.latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(listener.received).containsExactly("hello", "world");
+	}
+
+	@Test
+	void testMultipleListenersWithDifferentConverters() throws Exception {
+		BatchListener listener1 = this.config.batchListener();
+		BatchListener2 listener2 = this.config.batchListener2();
+		listener1.reset(1);
+		listener2.reset(1);
+
+		String listener1Data = "listener1Data";
+		String listener2Data = "listener2Data";
+		this.template.send("smartBatchTopic", listener1Data.getBytes());
+		this.template.send("smartBatchTopic2", listener2Data.getBytes());
+
+		assertThat(listener1.latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(listener2.latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(listener1.received).containsExactly(listener1Data);
+		assertThat(listener2.received).containsExactly(listener2Data.toUpperCase());
+	}
+
+	@Configuration
+	@EnableKafka
+	public static class Config {
+
+		@Bean
+		public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory(EmbeddedKafkaBroker embeddedKafka) {
+			ConcurrentKafkaListenerContainerFactory<Integer, byte[]> factory =
+					new ConcurrentKafkaListenerContainerFactory<>();
+			factory.setConsumerFactory(consumerFactory(embeddedKafka));
+			factory.setBatchListener(true);
+			return factory;
+		}
+
+		@Bean
+		public DefaultKafkaConsumerFactory<Integer, byte[]> consumerFactory(EmbeddedKafkaBroker embeddedKafka) {
+			return new DefaultKafkaConsumerFactory<>(consumerConfigs(embeddedKafka));
+		}
+
+		@Bean
+		public Map<String, Object> consumerConfigs(EmbeddedKafkaBroker embeddedKafka) {
+			Map<String, Object> consumerProps =
+					KafkaTestUtils.consumerProps("smartBatchGroup", "false", embeddedKafka);
+			consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+			consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
+			return consumerProps;
+		}
+
+		@Bean
+		public KafkaTemplate<Integer, byte[]> template(EmbeddedKafkaBroker embeddedKafka) {
+			return new KafkaTemplate<>(producerFactory(embeddedKafka));
+		}
+
+		@Bean
+		public ProducerFactory<Integer, byte[]> producerFactory(EmbeddedKafkaBroker embeddedKafka) {
+			return new DefaultKafkaProducerFactory<>(producerConfigs(embeddedKafka));
+		}
+
+		@Bean
+		public Map<String, Object> producerConfigs(EmbeddedKafkaBroker embeddedKafka) {
+			Map<String, Object> props = KafkaTestUtils.producerProps(embeddedKafka);
+			props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
+			props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
+			return props;
+		}
+
+		@Bean
+		public SmartMessageConverter 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 CountDownLatch latch = new CountDownLatch(2);
+
+		private final List<String> received = new ArrayList<>();
+
+		@KafkaListener(
+				id = "batchSmartListener",
+				topics = "smartBatchTopic",
+				groupId = "smartBatchGroup",
+				contentTypeConverter = "byteArrayToStringConverter",
+				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);
+		}
+
+	}
+
+	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 using a function.
+	 */
+	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) ? this.converter.apply(bytes) : payload;
+		}
+
+		@Override
+		public Object fromMessage(Message<?> message, Class<?> targetClass, Object conversionHint) {
+			return fromMessage(message, targetClass);
+		}
+
+		@Override
+		public Message<?> toMessage(Object payload, MessageHeaders headers) {
+			return MessageBuilder.withPayload(payload).copyHeaders(headers).build();
+		}
+
+		@Override
+		public Message<?> toMessage(Object payload, MessageHeaders headers, Object conversionHint) {
+			return toMessage(payload, headers);
+		}
+
+	}
+}