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.

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

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);
 		}
+
 	}
 }