Add AmqpHeaders.CHANNEL in error hander

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

@@ -26,13 +26,15 @@
 import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
 import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException;
 import org.springframework.amqp.support.AmqpHeaderMapper;
+import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.amqp.support.converter.MessageConversionException;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.support.converter.MessagingMessageConverter;
 import org.springframework.core.MethodParameter;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.remoting.support.RemoteInvocationResult;
 import org.springframework.util.Assert;
 
@@ -142,6 +144,9 @@ public void onMessage(org.springframework.amqp.core.Message amqpMessage, Channel
 		catch (ListenerExecutionFailedException e) {
 			if (this.errorHandler != null) {
 				try {
+					message = MessageBuilder.fromMessage(message)
+							.setHeader(AmqpHeaders.CHANNEL, channel)
+							.build();
 					Object errorResult = this.errorHandler.handleError(amqpMessage, message, e);
 					if (errorResult != null) {
 						handleResult(new InvocationResult(errorResult, null, null), amqpMessage, channel, message);

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java

@@ -765,6 +765,8 @@ public void listenerErrorHandlerException() {
 		catch (Exception e) {
 			assertThat(e.getCause().getMessage()).isEqualTo("from error handler");
 			assertThat(e.getCause().getCause().getMessage()).isEqualTo("return this");
+			EnableRabbitConfig config = this.context.getBean(EnableRabbitConfig.class);
+			assertThat(config.errorHandlerChannel).isNotNull();
 		}
 	}
 
@@ -1351,6 +1353,8 @@ public static class EnableRabbitConfig {
 
 		private final CountDownLatch noListenerLatch = new CountDownLatch(1);
 
+		private volatile Channel errorHandlerChannel;
+
 		@Bean
 		public ConnectionNameStrategy cns() {
 			return new SimplePropertyValueConnectionNameStrategy("spring.application.name");
@@ -1552,6 +1556,7 @@ public RabbitListenerErrorHandler alwaysBARHandler() {
 		@Bean
 		public RabbitListenerErrorHandler throwANewException() {
 			return (m, sm, e) -> {
+				this.errorHandlerChannel = sm.getHeaders().get(AmqpHeaders.CHANNEL, Channel.class);
 				throw new RuntimeException("from error handler", e.getCause());
 			};
 		}

src/reference/asciidoc/amqp.adoc

@@ -2756,6 +2756,21 @@ If you use JSON, consider using an `errorHandler` to return some other Jackson-f
 
 IMPORTANT: In version 2.1, this interface moved from package `o.s.amqp.rabbit.listener` to `o.s.amqp.rabbit.listener.api`.
 
+Starting with version 2.1.7, the `Channel` is available in a messaging message header; this allows you to ack or nack the failed messasge when using `AcknowledgeMode.MANUAL`:
+
+====
+[source, java]
+----
+public Object handleError(Message amqpMessage, org.springframework.messaging.Message<?> message,
+          ListenerExecutionFailedException exception) {
+              ...
+              message.getHeaders().get(AmqpHeaders.CHANNEL, Channel.class)
+                  .basicReject(message.getHeaders().get(AmqpHeaders.DELIVERY_TAG, Long.class),
+                               true);
+          }
+----
+====
+
 ====== Container Management
 
 Containers created for annotations are not registered with the application context.