AMQP-794: CRErrorHandler: Traverse cause tree

JIRA: https://jira.spring.io/browse/AMQP-794

In the `ConditionalRejectingErrorHandleri.DefaultExceptionStrategy`,
if the `cause` of the `ListenerExecutionFailedException` is a `MessagingException`,
traverse the `cause` tree to find the root cause of all such exceptions unless
it's a fatal cause itself (`MethodArgumentResolutionException` or `MessageConversionException`).
The final fatal check is performed on the root cause.

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

@@ -22,8 +22,10 @@
 import org.springframework.amqp.AmqpRejectAndDontRequeueException;
 import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
 import org.springframework.amqp.support.converter.MessageConversionException;
+import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.handler.annotation.support.MethodArgumentNotValidException;
 import org.springframework.messaging.handler.annotation.support.MethodArgumentTypeMismatchException;
+import org.springframework.messaging.handler.invocation.MethodArgumentResolutionException;
 import org.springframework.util.ErrorHandler;
 
 /**
@@ -102,8 +104,14 @@ public static class DefaultExceptionStrategy implements FatalExceptionStrategy {
 
 		@Override
 		public boolean isFatal(Throwable t) {
-			if (t instanceof ListenerExecutionFailedException
-					&& isCauseFatal(t.getCause())) {
+			Throwable cause = t.getCause();
+			while (cause instanceof MessagingException
+					&& !(cause instanceof
+							org.springframework.messaging.converter.MessageConversionException)
+					&& !(cause instanceof MethodArgumentResolutionException)) {
+				cause = cause.getCause();
+			}
+			if (t instanceof ListenerExecutionFailedException && isCauseFatal(cause)) {
 				if (this.logger.isWarnEnabled()) {
 					this.logger.warn(
 							"Fatal message conversion error; message rejected; "
@@ -118,8 +126,7 @@ && isCauseFatal(t.getCause())) {
 		private boolean isCauseFatal(Throwable cause) {
 			return cause instanceof MessageConversionException
 					|| cause instanceof org.springframework.messaging.converter.MessageConversionException
-					|| cause instanceof MethodArgumentNotValidException
-					|| cause instanceof MethodArgumentTypeMismatchException
+					|| cause instanceof MethodArgumentResolutionException
 					|| cause instanceof NoSuchMethodException
 					|| cause instanceof ClassCastException
 					|| isUserCauseFatal(cause);

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ErrorHandlerTests.java

@@ -27,12 +27,14 @@
 import org.junit.Test;
 
 import org.springframework.amqp.AmqpRejectAndDontRequeueException;
+import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
 import org.springframework.amqp.support.converter.MessageConversionException;
 import org.springframework.amqp.utils.test.TestUtils;
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.core.MethodParameter;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.handler.annotation.support.MethodArgumentNotValidException;
 import org.springframework.messaging.handler.annotation.support.MethodArgumentTypeMismatchException;
 
@@ -92,6 +94,39 @@ public void testFatalsAreRejected() throws Exception {
 		}
 	}
 
+	@Test
+	public void testSimple() {
+		Throwable cause = new ClassCastException();
+		try {
+			doTest(cause);
+			fail("Expected exception");
+		}
+		catch (AmqpRejectAndDontRequeueException e) {
+			// noop
+		}
+	}
+
+	@Test
+	public void testMessagingException() {
+		Throwable cause = new MessageHandlingException(null, "test",
+				new MessageHandlingException(null, "test", new ClassCastException()));
+		try {
+			doTest(cause);
+			fail("Expected exception");
+		}
+		catch (AmqpRejectAndDontRequeueException e) {
+			// noop
+		}
+	}
+
+	private void doTest(Throwable cause) {
+		ConditionalRejectingErrorHandler handler = new ConditionalRejectingErrorHandler();
+		handler.handleError(
+				new ListenerExecutionFailedException("test", cause,
+						new org.springframework.amqp.core.Message(new byte[0],
+				new MessageProperties())));
+	}
+
 	private static class Foo {
 
 		@SuppressWarnings("unused")