GH-4097: Fix SmartMessageConverter support in batch listeners

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

@@ -117,9 +117,10 @@ public void setBatchToRecordAdapter(BatchToRecordAdapter<K, V> batchToRecordAdap
 	 * properly propagated to the batch converter's record converter for message conversion
 	 * in batch listeners.
 	 * <p>
-	 * 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
+	 * This method cannot be called after {@link #setBatchMessageConverter(BatchMessageConverter)
+	 * setBatchMessageConverter()} as it would cause a mutation of the internal
+	 * batchMessageConverter. Instead, the SmartMessageConverter has to be provided on the
+	 * external 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

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

@@ -146,7 +146,7 @@ public RecordMessageConverter getRecordMessageConverter() {
 	}
 
 	/**
-	 * Set a spring-messaging {@link SmartMessageConverter} to convert the record value to
+	 * Set a {@link SmartMessageConverter} to convert the record value to
 	 * the desired type.
 	 * @param messagingConverter the converter.
 	 * @since 3.3.11

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

@@ -55,7 +55,6 @@
 
 /**
  * Integration tests for SmartMessageConverter support in batch listeners.
- * Reproduces and verifies the fix for the issue described in GH-4097.
  *
  * @author George Mahfoud
  * @since 3.3.11
@@ -80,7 +79,7 @@ void testContentTypeConverterWithBatchListener() throws Exception {
 		this.template.send("smartBatchTopic", "world".getBytes());
 
 		assertThat(listener.latch.await(10, TimeUnit.SECONDS)).isTrue();
-		assertThat(listener.received).hasSize(2).containsExactly("hello", "world");
+		assertThat(listener.received).containsExactly("hello", "world");
 	}
 
 	@Test
@@ -90,13 +89,15 @@ void testMultipleListenersWithDifferentConverters() throws Exception {
 		listener1.reset(1);
 		listener2.reset(1);
 
-		this.template.send("smartBatchTopic", "foo".getBytes());
-		this.template.send("smartBatchTopic2", "bar".getBytes());
+		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).hasSize(1).containsExactly("foo");
-		assertThat(listener2.received).hasSize(1).containsExactly("BAR");
+		assertThat(listener1.received).containsExactly(listener1Data);
+		assertThat(listener2.received).containsExactly(listener2Data.toUpperCase());
 	}
 
 	@Configuration