GH-2638: Fix for Kotlin suspend returns

Fixes: #2638

The `AbstractAdaptableMessageListener.asyncSuccess()` assumes only `Mono` & `CompletableFuture` return types.
With Kotlin `suspend` it is just plain `Object`.

* Fix `AbstractAdaptableMessageListener.asyncSuccess()` to check if `returnType` is not simple `Object.class`
before going deep for its generic actual argument.

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

Author: artembilan

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2023 the original author or authors.
+ * Copyright 2014-2024 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.
@@ -418,9 +418,10 @@ private void asyncSuccess(InvocationResult resultArg, Message request, Channel c
 			}
 		}
 		else {
-			// We only get here with Mono<?> and ListenableFuture<?> which have exactly one type argument
 			Type returnType = resultArg.getReturnType();
-			if (returnType != null) {
+			// We only get here with Mono<?> and CompletableFuture<?> which have exactly one type argument
+			// Otherwise it might be Kotlin suspend function
+			if (returnType != null && !Object.class.getName().equals(returnType.getTypeName())) {
 				Type[] actualTypeArguments = ((ParameterizedType) returnType).getActualTypeArguments();
 				if (actualTypeArguments.length > 0) {
 					returnType = actualTypeArguments[0]; // NOSONAR

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

@@ -21,7 +21,6 @@ import assertk.assertions.isEqualTo
 import assertk.assertions.isTrue
 import org.junit.jupiter.api.Test
 import org.springframework.amqp.core.AcknowledgeMode
-import org.springframework.amqp.core.MessageListener
 import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory
 import org.springframework.amqp.rabbit.core.RabbitTemplate
@@ -62,8 +61,9 @@ class EnableRabbitKotlinTests {
 	@Test
 	fun `send and wait for consume`(@Autowired registry: RabbitListenerEndpointRegistry) {
 		val template = RabbitTemplate(this.config.cf())
-		template.convertAndSend("kotlinQueue", "test")
-		assertThat(this.config.latch.await(10, TimeUnit.SECONDS)).isTrue()
+		template.setReplyTimeout(10_000)
+		val result = template.convertSendAndReceive("kotlinQueue", "test")
+		assertThat(result).isEqualTo("TEST")
 		val listener = registry.getListenerContainer("single").messageListener
 		assertThat(TestUtils.getPropertyValue(listener, "messagingMessageConverter.inferredArgumentType").toString())
 				.isEqualTo("class java.lang.String")
@@ -82,11 +82,9 @@ class EnableRabbitKotlinTests {
 	@EnableRabbit
 	class Config {
 
-		val latch = CountDownLatch(1)
-
 		@RabbitListener(id = "single", queues = ["kotlinQueue"])
-		suspend fun handle(@Suppress("UNUSED_PARAMETER") data: String) {
-			this.latch.countDown()
+		suspend fun handle(@Suppress("UNUSED_PARAMETER") data: String) : String? {
+			return data.uppercase()
 		}
 
 		@Bean