GH-2890: Fix MessagingMessageListenerAdapter for batch in Kotlin

Fixes: #2890
Issue link: #2890

The Kotlin function with signature `receiveBatch(messages: List<Message>)`
produced a `WildCardType` for the generic of the `List` argument.

* Fix `MessagingMessageListenerAdapter` to use `TypeUtils.isAssignable()`
to determine if the `Type` has a part as expected type

**Auto-cherry-pick to `3.1.x`**

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java

@@ -41,6 +41,7 @@
 import org.springframework.messaging.handler.annotation.Headers;
 import org.springframework.messaging.handler.annotation.Payload;
 import org.springframework.util.Assert;
+import org.springframework.util.TypeUtils;
 
 import com.rabbitmq.client.Channel;
 
@@ -456,17 +457,18 @@ private Type extractGenericParameterTypFromMethodParameter(MethodParameter metho
 				if (parameterizedType.getRawType().equals(Message.class)) {
 					genericParameterType = ((ParameterizedType) genericParameterType).getActualTypeArguments()[0];
 				}
-				else if (this.isBatch
-						&& ((parameterizedType.getRawType().equals(List.class)
-						|| parameterizedType.getRawType().equals(Collection.class))
-						&& parameterizedType.getActualTypeArguments().length == 1)) {
+				else if (this.isBatch &&
+						(parameterizedType.getRawType().equals(List.class) ||
+								(parameterizedType.getRawType().equals(Collection.class) &&
+										parameterizedType.getActualTypeArguments().length == 1))) {
 
 					this.isCollection = true;
 					Type paramType = parameterizedType.getActualTypeArguments()[0];
 					boolean messageHasGeneric = paramType instanceof ParameterizedType pType
 							&& pType.getRawType().equals(Message.class);
-					this.isMessageList = paramType.equals(Message.class) || messageHasGeneric;
-					this.isAmqpMessageList = paramType.equals(org.springframework.amqp.core.Message.class);
+					this.isMessageList = TypeUtils.isAssignable(paramType, Message.class) || messageHasGeneric;
+					this.isAmqpMessageList =
+							TypeUtils.isAssignable(paramType, org.springframework.amqp.core.Message.class);
 					if (messageHasGeneric) {
 						genericParameterType = ((ParameterizedType) paramType).getActualTypeArguments()[0];
 					}

spring-rabbit/src/test/kotlin/org/springframework/amqp/rabbit/annotation/EnableRabbitKotlinTests.kt

@@ -19,9 +19,11 @@ package org.springframework.amqp.rabbit.annotation
 import assertk.assertThat
 import assertk.assertions.containsOnly
 import assertk.assertions.isEqualTo
+import assertk.assertions.isInstanceOf
 import assertk.assertions.isTrue
 import org.junit.jupiter.api.Test
 import org.springframework.amqp.core.AcknowledgeMode
+import org.springframework.amqp.core.Message
 import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory
 import org.springframework.amqp.rabbit.core.RabbitTemplate
@@ -77,7 +79,8 @@ class EnableRabbitKotlinTests {
 		template.convertAndSend("kotlinBatchQueue", "test1")
 		template.convertAndSend("kotlinBatchQueue", "test2")
 		assertThat(this.config.batchReceived.await(10, TimeUnit.SECONDS)).isTrue()
-		assertThat(this.config.batch).containsOnly("test1", "test2")
+		assertThat(this.config.batch[0]).isInstanceOf(Message::class.java)
+		assertThat(this.config.batch.map { m -> String(m.body) }).containsOnly("test1", "test2")
 	}
 
 	@Test
@@ -100,11 +103,11 @@ class EnableRabbitKotlinTests {
 
 		val batchReceived = CountDownLatch(1)
 
-		lateinit var batch: List<String>
+		lateinit var batch: List<Message>
 
 		@RabbitListener(id = "batch", queues = ["kotlinBatchQueue"],
 				containerFactory = "batchRabbitListenerContainerFactory")
-		suspend fun receiveBatch(messages: List<String>) {
+		suspend fun receiveBatch(messages: List<Message>) {
 			batch = messages
 			batchReceived.countDown()
 		}