GH-3163: Propagate EH error down to the super.asyncFailure

Fixes: #3163

The `MessagingMessageListenerAdapter` in its `asyncFailure()` calls the `handleException()`,
but uses the same input `Throwable` for the `super.asyncFailure()`.
This may not lead to expected state of the system, when an `AmqpRejectAndDontRequeueException`
might be expected as a trigger to DLQ the failed message.

* Fix `MessagingMessageListenerAdapter.asyncFailure()` to override error with an exception thrown
from the `handleException()` before calling `super.asyncFailure()`

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

Author: artembilan

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

@@ -172,15 +172,18 @@ public void onMessage(org.springframework.amqp.core.Message amqpMessage, @Nullab
 	protected void asyncFailure(org.springframework.amqp.core.Message request, @Nullable Channel channel, Throwable t,
 			@Nullable Object source) {
 
+		Throwable throwableToHandle = t;
+
 		try {
 			handleException(request, channel, (Message<?>) source,
-					new ListenerExecutionFailedException("Async Fail", t, request));
+					new ListenerExecutionFailedException("Async Fail", throwableToHandle, request));
 			return;
 		}
 		catch (Exception ex) {
-			// Ignore
+			throwableToHandle = ex;
 		}
-		super.asyncFailure(request, channel, t, source);
+
+		super.asyncFailure(request, channel, throwableToHandle, source);
 	}
 
 	protected void handleException(org.springframework.amqp.core.Message amqpMessage, @Nullable Channel channel,

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

@@ -23,10 +23,13 @@ import assertk.assertions.isInstanceOf
 import assertk.assertions.isNotNull
 import assertk.assertions.isTrue
 import org.junit.jupiter.api.Test
+import org.springframework.amqp.AmqpRejectAndDontRequeueException
 import org.springframework.amqp.core.AcknowledgeMode
 import org.springframework.amqp.core.Message
+import org.springframework.amqp.core.QueueBuilder
 import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory
+import org.springframework.amqp.rabbit.core.RabbitAdmin
 import org.springframework.amqp.rabbit.core.RabbitTemplate
 import org.springframework.amqp.rabbit.junit.RabbitAvailable
 import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition
@@ -56,7 +59,7 @@ import java.util.concurrent.TimeUnit
  *
  */
 @SpringJUnitConfig
-@RabbitAvailable(queues = ["kotlinQueue", "kotlinBatchQueue", "kotlinQueue1", "kotlinReplyQueue"])
+@RabbitAvailable(queues = ["kotlinDLQ", "kotlinBatchQueue", "kotlinQueue1", "kotlinReplyQueue"])
 @DirtiesContext
 class EnableRabbitKotlinTests {
 
@@ -67,7 +70,7 @@ class EnableRabbitKotlinTests {
 	private lateinit var registry: RabbitListenerEndpointRegistry
 
 	@Test
-	fun `send and wait for consume`() {
+	fun `send and wait for consume and DLQ on error`() {
 		val template = RabbitTemplate(this.config.cf())
 		template.setReplyTimeout(10_000)
 		val result = template.convertSendAndReceive("kotlinQueue", "test")
@@ -83,6 +86,11 @@ class EnableRabbitKotlinTests {
 			)
 				.isEqualTo("class java.lang.String")
 		}
+
+		template.convertAndSend("kotlinQueue", "junk")
+
+		var dlqResult = template.receiveAndConvert("kotlinDLQ", 30_000)
+		assertThat(dlqResult).isEqualTo("junk")
 	}
 
 	@Test
@@ -114,9 +122,25 @@ class EnableRabbitKotlinTests {
 	@EnableRabbit
 	class Config {
 
-		@RabbitListener(id = "single", queues = ["kotlinQueue"])
-		suspend fun handle(@Suppress("UNUSED_PARAMETER") data: String): String? {
-			return data.uppercase()
+		@Bean
+		fun kotlinQueue() =
+			QueueBuilder.durable("kotlinQueue")
+				.deadLetterExchange("")
+				.deadLetterRoutingKey("kotlinDLQ")
+				.build()
+
+		@RabbitListener(id = "single", queues = ["kotlinQueue"], errorHandler = "dontRequeueErrorHandler")
+		suspend fun handle(data: String): String? {
+			if ("junk" == data) {
+				throw IllegalArgumentException("Illegal data: $data")
+			} else {
+				return data.uppercase()
+			}
+		}
+
+		@Bean
+		fun dontRequeueErrorHandler() = RabbitListenerErrorHandler { _, _, _, ex ->
+			throw AmqpRejectAndDontRequeueException(ex)
 		}
 
 		val batchReceived = CountDownLatch(1)
@@ -132,6 +156,9 @@ class EnableRabbitKotlinTests {
 			batchReceived.countDown()
 		}
 
+		@Bean
+		fun rabbitAdmin(cf: CachingConnectionFactory) = RabbitAdmin(cf)
+
 		@Bean
 		fun rabbitListenerContainerFactory(cf: CachingConnectionFactory) =
 			SimpleRabbitListenerContainerFactory().also {