GH-3161: Fix HandlerAdapter for Kotlin suspend functions

Fixes: #3161

The Kotlin `suspend` are async by nature.
However, the `HandlerAdapter` is missing to detect that for its `asyncReplies` property
which is used to set `acknowledgeMode` in the container to the `MANUAL` mode.

* Add `KotlinDetector.isSuspendingFunction()` check to the `HandlerAdapter` for its `asyncReplies` property
* Modify `EnableRabbitKotlinTests` to remove `setAcknowledgeMode(AcknowledgeMode.MANUAL)` from listener container factory beans.
Instead, verify in the respective tests that listener container is switched to the `AcknowledgeMode.MANUAL` due to `true`
outcome from the `KotlinDetector.isSuspendingFunction()` for those `suspend` functions in the test configuration

# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java
#	spring-rabbit/src/test/kotlin/org/springframework/amqp/rabbit/annotation/EnableRabbitKotlinTests.kt

Author: artembilan

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

@@ -21,6 +21,7 @@
 import java.util.concurrent.CompletableFuture;
 
 import org.springframework.lang.Nullable;
+import org.springframework.core.KotlinDetector;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
 
@@ -30,6 +31,8 @@
  * underlying handler.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 1.5
  *
  */
@@ -48,9 +51,11 @@ public class HandlerAdapter {
 	public HandlerAdapter(InvocableHandlerMethod invokerHandlerMethod) {
 		this.invokerHandlerMethod = invokerHandlerMethod;
 		this.delegatingHandler = null;
+		Method method = invokerHandlerMethod.getMethod();
 		this.asyncReplies = (AbstractAdaptableMessageListener.monoPresent
-				&& MonoHandler.isMono(invokerHandlerMethod.getMethod().getReturnType()))
-			|| CompletableFuture.class.isAssignableFrom(invokerHandlerMethod.getMethod().getReturnType());
+				&& MonoHandler.isMono(method.getReturnType()))
+				|| CompletableFuture.class.isAssignableFrom(method.getReturnType())
+				|| KotlinDetector.isSuspendingFunction(method);
 	}
 
 	/**

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

@@ -29,6 +29,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory
 import org.springframework.amqp.rabbit.core.RabbitTemplate
 import org.springframework.amqp.rabbit.junit.RabbitAvailable
 import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition
+import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer
 import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry
 import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler
 import org.springframework.amqp.utils.test.TestUtils
@@ -43,7 +44,6 @@ import org.springframework.test.annotation.DirtiesContext
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
-import java.util.concurrent.atomic.AtomicBoolean
 
 /**
  * Kotlin Annotated listener tests.
@@ -62,14 +62,24 @@ class EnableRabbitKotlinTests {
 	@Autowired
 	private lateinit var config: Config
 
+	@Autowired
+	private lateinit var registry: RabbitListenerEndpointRegistry
+
 	@Test
-	fun `send and wait for consume`(@Autowired registry: RabbitListenerEndpointRegistry) {
+	fun `send and wait for consume`() {
 		val template = RabbitTemplate(this.config.cf())
 		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())
+		var listenerContainer = registry.getListenerContainer("single") as AbstractMessageListenerContainer
+		assertThat(listenerContainer.acknowledgeMode).isEqualTo(AcknowledgeMode.MANUAL)
+		val listener = listenerContainer?.messageListener
+		assertThat(listener).isNotNull()
+		listener?.let { nonNullableListener ->
+			assertThat(
+				TestUtils.getPropertyValue(nonNullableListener, "messagingMessageConverter.inferredArgumentType")
+					.toString()
+			)
 				.isEqualTo("class java.lang.String")
 	}
 
@@ -81,6 +91,9 @@ class EnableRabbitKotlinTests {
 		assertThat(this.config.batchReceived.await(10, TimeUnit.SECONDS)).isTrue()
 		assertThat(this.config.batch[0]).isInstanceOf(Message::class.java)
 		assertThat(this.config.batch.map { m -> String(m.body) }).containsOnly("test1", "test2")
+
+		var listenerContainer = registry.getListenerContainer("batch") as AbstractMessageListenerContainer
+		assertThat(listenerContainer.acknowledgeMode).isEqualTo(AcknowledgeMode.MANUAL)
 	}
 
 	@Test
@@ -90,45 +103,48 @@ class EnableRabbitKotlinTests {
 		assertThat(this.config.ehLatch.await(10, TimeUnit.SECONDS)).isTrue()
 		val reply = template.receiveAndConvert("kotlinReplyQueue", 10_000)
 		assertThat(reply).isEqualTo("error processed")
+
+		var listenerContainer = registry.getListenerContainer("multi") as AbstractMessageListenerContainer
+		assertThat(listenerContainer.acknowledgeMode).isEqualTo(AcknowledgeMode.AUTO)
 	}
 
 	@Configuration
 	@EnableRabbit
 	class Config {
 
 		@RabbitListener(id = "single", queues = ["kotlinQueue"])
-		suspend fun handle(@Suppress("UNUSED_PARAMETER") data: String) : String? {
+		suspend fun handle(@Suppress("UNUSED_PARAMETER") data: String): String? {
 			return data.uppercase()
 		}
 
 		val batchReceived = CountDownLatch(1)
 
 		lateinit var batch: List<Message>
 
-		@RabbitListener(id = "batch", queues = ["kotlinBatchQueue"],
-				containerFactory = "batchRabbitListenerContainerFactory")
+		@RabbitListener(
+			id = "batch", queues = ["kotlinBatchQueue"],
+			containerFactory = "batchRabbitListenerContainerFactory"
+		)
 		suspend fun receiveBatch(messages: List<Message>) {
 			batch = messages
 			batchReceived.countDown()
 		}
 
 		@Bean
 		fun rabbitListenerContainerFactory(cf: CachingConnectionFactory) =
-				SimpleRabbitListenerContainerFactory().also {
-					it.setAcknowledgeMode(AcknowledgeMode.MANUAL)
-					it.setReceiveTimeout(10)
-					it.setConnectionFactory(cf)
-				}
+			SimpleRabbitListenerContainerFactory().also {
+				it.setReceiveTimeout(10)
+				it.setConnectionFactory(cf)
+			}
 
 		@Bean
 		fun batchRabbitListenerContainerFactory(cf: CachingConnectionFactory) =
-				SimpleRabbitListenerContainerFactory().also {
-					it.setAcknowledgeMode(AcknowledgeMode.MANUAL)
-					it.setConsumerBatchEnabled(true)
-					it.setDeBatchingEnabled(true)
-					it.setBatchSize(3)
-					it.setConnectionFactory(cf)
-				}
+			SimpleRabbitListenerContainerFactory().also {
+				it.setConsumerBatchEnabled(true)
+				it.setDeBatchingEnabled(true)
+				it.setBatchSize(3)
+				it.setConnectionFactory(cf)
+			}
 
 		@Bean
 		fun cf() = CachingConnectionFactory(RabbitAvailableCondition.getBrokerRunning().connectionFactory)