Address review feedback: Use integration test and fix adapter implementation

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.

Author: Jujuwryy

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

@@ -56,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>
@@ -109,20 +110,24 @@ public void setBatchToRecordAdapter(BatchToRecordAdapter<K, V> batchToRecordAdap
 	}
 
 	/**
-	 * Set the {@link SmartMessageConverter} to use with both the default record converter
-	 * and the batch message converter.
+	 * 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 both the record converter (via the parent class) and the
-	 * batch converter to support message conversion in batch listeners.
+	 * properly propagated to the batch converter, which will then propagate it to the
+	 * 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.
 	 * @param messageConverter the converter to set
 	 */
 	@Override
 	public void setMessagingConverter(SmartMessageConverter messageConverter) {
-		super.setMessagingConverter(messageConverter);
-		if (this.batchMessageConverter instanceof BatchMessagingMessageConverter batchConverter) {
-			batchConverter.setMessagingConverter(messageConverter);
+		if (this.batchMessageConverter instanceof BatchMessagingMessageConverter) {
+			((BatchMessagingMessageConverter) this.batchMessageConverter)
+					.setMessagingConverter(messageConverter);
 		}
 	}
 

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

@@ -68,6 +68,7 @@
  * @author Borahm Lee
  * @author Artem Bilan
  * @author Soby Chacko
+ * @author George Mahfoud
  *
  * @since 1.1
  */
@@ -79,8 +80,6 @@ public class BatchMessagingMessageConverter implements BatchMessageConverter {
 	@Nullable
 	private final RecordMessageConverter recordConverter;
 
-	private @Nullable SmartMessageConverter messagingConverter;
-
 	private boolean generateMessageId = false;
 
 	private boolean generateTimestamp = false;
@@ -152,8 +151,6 @@ public RecordMessageConverter getRecordMessageConverter() {
 	 * @since 3.3.11
 	 */
 	public void setMessagingConverter(@Nullable SmartMessageConverter messagingConverter) {
-		this.messagingConverter = messagingConverter;
-
 		if (this.recordConverter instanceof MessagingMessageConverter messagingRecordConverter) {
 			messagingRecordConverter.setMessagingConverter(messagingConverter);
 		}

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

@@ -16,140 +16,183 @@
 
 package org.springframework.kafka.listener;
 
-import java.lang.reflect.Type;
-import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.List;
-
-import org.apache.kafka.clients.consumer.ConsumerRecord;
+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.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;
 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;
 
 /**
- * Tests for SmartMessageConverter support in batch listeners.
- * Reproduces the issue described in GH-4097.
+ * Integration tests for SmartMessageConverter support in batch listeners.
+ * Reproduces and verifies the fix for the issue described in GH-4097.
  *
- * @author Jujuwryy
+ * @author George Mahfoud
  * @since 3.3.11
  */
+@SpringJUnitConfig
+@DirtiesContext
+@EmbeddedKafka(partitions = 1, topics = { "smartBatchTopic" })
 class BatchSmartMessageConverterTests {
 
-	@Test
-	void testSmartMessageConverterWorksInBatchConversion() {
-		// Given: A BatchMessagingMessageConverter with a record converter and SmartMessageConverter
-		MessagingMessageConverter recordConverter = new MessagingMessageConverter();
-		BatchMessagingMessageConverter batchConverter = new BatchMessagingMessageConverter(recordConverter);
-
-		// Set up SmartMessageConverter that converts byte[] to String
-		TestStringMessageConverter smartConverter = new TestStringMessageConverter();
-		batchConverter.setMessagingConverter(smartConverter);
-
-		// Create test records with byte[] values that need conversion to String
-		List<ConsumerRecord<?, ?>> records = Arrays.asList(
-				new ConsumerRecord<>("topic", 0, 0, "key", "hello".getBytes()),
-				new ConsumerRecord<>("topic", 0, 1, "key", "world".getBytes())
-		);
-
-		// When: Convert batch with List<String> target type
-		Type targetType = new TestParameterizedType(List.class, new Type[]{String.class});
-		Message<?> result = batchConverter.toMessage(records, null, null, targetType);
-
-		// Then: Verify the SmartMessageConverter was applied and byte[] was converted to String
-		assertThat(result).isNotNull();
-		assertThat(result.getPayload()).isInstanceOf(List.class);
-
-		List<?> payloads = (List<?>) result.getPayload();
-		assertThat(payloads).hasSize(2);
-		assertThat(payloads.get(0)).isEqualTo("hello");
-		assertThat(payloads.get(1)).isEqualTo("world");
-	}
+	@Autowired
+	private KafkaTemplate<Integer, byte[]> template;
+
+	@Autowired
+	private Config config;
 
 	@Test
-	void testBatchConversionWithoutSmartMessageConverter() {
-		// Given: A BatchMessagingMessageConverter without SmartMessageConverter
-		MessagingMessageConverter recordConverter = new MessagingMessageConverter();
-		BatchMessagingMessageConverter batchConverter = new BatchMessagingMessageConverter(recordConverter);
-
-		// Create test records with byte[] values
-		List<ConsumerRecord<?, ?>> records = Arrays.asList(
-				new ConsumerRecord<>("topic", 0, 0, "key", "test".getBytes())
-		);
-
-		// When: Convert batch
-		Type targetType = new TestParameterizedType(List.class, new Type[]{String.class});
-		Message<?> result = batchConverter.toMessage(records, null, null, targetType);
-
-		// Then: Should work but payloads remain as byte[]
-		assertThat(result).isNotNull();
-		List<?> payloads = (List<?>) result.getPayload();
-		assertThat(payloads.get(0)).isInstanceOf(byte[].class);
+	void testContentTypeConverterWithBatchListener() throws Exception {
+		// Given: A batch listener with contentTypeConverter configured
+		BatchListener listener = this.config.batchListener();
+
+		// 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 SmartMessageConverter that converts byte[] to String.
-	 */
-	static class TestStringMessageConverter implements SmartMessageConverter {
+	@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);
+			// Set up batch converter with record converter - framework will propagate SmartMessageConverter
+			factory.setBatchMessageConverter(new BatchMessagingMessageConverter(new MessagingMessageConverter()));
+			return factory;
+		}
 
-		@Override
-		public Object fromMessage(Message<?> message, Class<?> targetClass) {
-			return convertPayload(message.getPayload());
+		@Bean
+		public DefaultKafkaConsumerFactory<Integer, byte[]> consumerFactory(EmbeddedKafkaBroker embeddedKafka) {
+			return new DefaultKafkaConsumerFactory<>(consumerConfigs(embeddedKafka));
 		}
 
-		@Override
-		public Object fromMessage(Message<?> message, Class<?> targetClass, Object conversionHint) {
-			return convertPayload(message.getPayload());
+		@Bean
+		public Map<String, Object> consumerConfigs(EmbeddedKafkaBroker embeddedKafka) {
+			Map<String, Object> consumerProps =
+					KafkaTestUtils.consumerProps(embeddedKafka, "smartBatchGroup", false);
+			consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+			consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
+			return consumerProps;
 		}
 
-		@Override
-		public Message<?> toMessage(Object payload, MessageHeaders headers) {
-			return MessageBuilder.withPayload(payload).copyHeaders(headers).build();
+		@Bean
+		public KafkaTemplate<Integer, byte[]> template(EmbeddedKafkaBroker embeddedKafka) {
+			return new KafkaTemplate<>(producerFactory(embeddedKafka));
 		}
 
-		@Override
-		public Message<?> toMessage(Object payload, MessageHeaders headers, Object conversionHint) {
-			return toMessage(payload, headers);
+		@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 ByteArrayToStringConverter();
 		}
 
-		private Object convertPayload(Object payload) {
-			// Convert byte[] to String - this is the core functionality being tested
-			if (payload instanceof byte[] bytes) {
-				return new String(bytes);
-			}
-			return payload;
+		@Bean
+		public BatchListener batchListener() {
+			return new BatchListener();
 		}
+
 	}
 
-	/**
-	 * Helper class for creating parameterized types for testing.
-	 */
-	static class TestParameterizedType implements java.lang.reflect.ParameterizedType {
+	public static class BatchListener {
+
+		private final CountDownLatch latch = new CountDownLatch(2);
 
-		private final Type rawType;
+		private final List<String> received = new ArrayList<>();
 
-		private final Type[] typeArguments;
+		@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();
+			});
+		}
+
+	}
+
+	/**
+	 * Simple SmartMessageConverter for testing that converts byte[] to String.
+	 */
+	static class ByteArrayToStringConverter implements SmartMessageConverter {
 
-		TestParameterizedType(Type rawType, Type[] typeArguments) {
-			this.rawType = rawType;
-			this.typeArguments = typeArguments;
+		@Override
+		public Object fromMessage(Message<?> message, Class<?> targetClass) {
+			Object payload = message.getPayload();
+			return (payload instanceof byte[] bytes) ? new String(bytes) : payload;
 		}
 
-		public Type[] getActualTypeArguments() {
-			return typeArguments;
+		@Override
+		public Object fromMessage(Message<?> message, Class<?> targetClass, Object conversionHint) {
+			return fromMessage(message, targetClass);
 		}
 
-		public Type getRawType() {
-			return rawType;
+		@Override
+		public Message<?> toMessage(Object payload, MessageHeaders headers) {
+			return MessageBuilder.withPayload(payload).copyHeaders(headers).build();
 		}
 
-		public Type getOwnerType() {
-			return null;
+		@Override
+		public Message<?> toMessage(Object payload, MessageHeaders headers, Object conversionHint) {
+			return toMessage(payload, headers);
 		}
+
 	}
 }