GH-2652: Make RabbitListenerErrorHandler aware of Channel

Fixes: #2652

Whenever a `MessageConversionException` occurs,
the raw Spring message returned in `RabbitListenerErrorHandler#handleError` is null.
Channel information is being stored inside of that raw message,
therefore it is not possible to manually nack just failed message

* Introduce a new `handleError(Message, Channel, Message<?>, ListenerExecutionFailedException)`
contract to make a `Channel` access independent of the `Message<?>` result
* Deprecate existing method with the plan to remove in the next version (see #2654)

Author: artembilan

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2023 the original author or authors.
+ * Copyright 2002-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.
@@ -182,10 +182,11 @@ private void handleException(org.springframework.amqp.core.Message amqpMessage,
 				Message<?> messageWithChannel = null;
 				if (message != null) {
 					messageWithChannel = MessageBuilder.fromMessage(message)
+							// TODO won't be necessary starting with version 3.2
 							.setHeader(AmqpHeaders.CHANNEL, channel)
 							.build();
 				}
-				Object errorResult = this.errorHandler.handleError(amqpMessage, messageWithChannel, e);
+				Object errorResult = this.errorHandler.handleError(amqpMessage, channel, messageWithChannel, e);
 				if (errorResult != null) {
 					Object payload = message == null ? null : message.getPayload();
 					InvocationResult invResult = payload == null

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/RabbitListenerErrorHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-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.
@@ -20,6 +20,7 @@
 import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException;
 import org.springframework.lang.Nullable;
 
+import com.rabbitmq.client.Channel;
 /**
  * An error handler which is called when a {code @RabbitListener} method
  * throws an exception. This is invoked higher up the stack than the
@@ -41,8 +42,31 @@ public interface RabbitListenerErrorHandler {
 	 * {@link ListenerExecutionFailedException}.
 	 * @return the return value to be sent to the sender.
 	 * @throws Exception an exception which may be the original or different.
+	 * @deprecated in favor of
+	 * {@link #handleError(Message, Channel, org.springframework.messaging.Message, ListenerExecutionFailedException)}
 	 */
+	@Deprecated(forRemoval = true, since = "3.1.3")
 	Object handleError(Message amqpMessage, @Nullable org.springframework.messaging.Message<?> message,
 			ListenerExecutionFailedException exception) throws Exception; // NOSONAR
 
+	/**
+	 * Handle the error. If an exception is not thrown, the return value is returned to
+	 * the sender using normal {@code replyTo/@SendTo} semantics.
+	 * @param amqpMessage the raw message received.
+	 * @param channel AMQP channel for manual acks.
+	 * @param message the converted spring-messaging message (if available).
+	 * @param exception the exception the listener threw, wrapped in a
+	 * {@link ListenerExecutionFailedException}.
+	 * @return the return value to be sent to the sender.
+	 * @throws Exception an exception which may be the original or different.
+	 * @since 3.1.3
+	 */
+	@SuppressWarnings("deprecation")
+	default Object handleError(Message amqpMessage, Channel channel,
+			@Nullable org.springframework.messaging.Message<?> message,
+			ListenerExecutionFailedException exception) throws Exception { // NOSONAR
+
+		return handleError(amqpMessage, message, exception);
+	}
+
 }