GH-2437: Fix Fatal When No Matching RabbitHandler

Resolves https://github.com/spring-projects/spring-amqp/issues/2437

Failure to find a `@RabbitHandler` method for the payload must be treated
as a fatal exception to avoid an infinite loop.

Also, fix CREH cause traversal to stop when **any** fatal exception is found.
Previously, traversal only stopped for the original subset of exceptions.

**cherry-pick to 2.4.x**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2022 the original author or authors.
+ * Copyright 2014-2023 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.amqp.rabbit.listener;
 
+import java.lang.reflect.UndeclaredThrowableException;
 import java.util.List;
 import java.util.Map;
 
@@ -200,9 +201,9 @@ public class ConditionalRejectingErrorHandler implements ErrorHandler {
 		@Override
 		public boolean isFatal(Throwable t) {
 			Throwable cause = t.getCause();
-			while (cause instanceof MessagingException
-					&& !(cause instanceof org.springframework.messaging.converter.MessageConversionException)
-					&& !(cause instanceof MethodArgumentResolutionException)) {
+			while ((cause instanceof MessagingException || cause instanceof  UndeclaredThrowableException)
+					&& !isCauseFatal(cause)) {
+
 				cause = cause.getCause();
 			}
 			if (t instanceof ListenerExecutionFailedException lefe && isCauseFatal(cause)) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2022 the original author or authors.
+ * Copyright 2015-2023 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.
@@ -46,6 +46,7 @@ import org.springframework.messaging.handler.annotation.SendTo;
 import org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver;
 import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
 import org.springframework.util.Assert;
+import org.springframework.util.ReflectionUtils;
 import org.springframework.validation.Validator;
 
 
@@ -204,7 +205,9 @@ public class DelegatingInvocableHandler {
 		if (handler == null) {
 			handler = findHandlerForPayload(payloadClass);
 			if (handler == null) {
-				throw new AmqpException("No method found for " + payloadClass);
+				ReflectionUtils.rethrowRuntimeException(
+						new NoSuchMethodException("No listener method found in " + this.bean.getClass().getName()
+								+ " for " + payloadClass));
 			}
 			this.cachedHandlers.putIfAbsent(payloadClass, handler); //NOSONAR
 			setupReplyTo(handler);

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

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2023 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbit.listener.adapter;
+
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.mockito.Mockito.mock;
+
+import java.lang.reflect.Method;
+import java.lang.reflect.UndeclaredThrowableException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.config.BeanExpressionContext;
+import org.springframework.beans.factory.config.BeanExpressionResolver;
+import org.springframework.format.support.DefaultFormattingConversionService;
+import org.springframework.messaging.converter.GenericMessageConverter;
+import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
+import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
+import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
+
+/**
+ * @author Gary Russell
+ * @since 2.4.12
+ *
+ */
+public class DelegatingInvocableHandlerTests {
+
+	@Test
+	void multiNoMatch() throws Exception {
+		List<InvocableHandlerMethod> methods = new ArrayList<>();
+		Object bean = new Multi();
+		Method method = Multi.class.getDeclaredMethod("listen", Integer.class);
+		methods.add(messageHandlerFactory().createInvocableHandlerMethod(bean, method));
+		BeanExpressionResolver resolver = mock(BeanExpressionResolver.class);
+		BeanExpressionContext context = mock(BeanExpressionContext.class);
+		DelegatingInvocableHandler handler = new DelegatingInvocableHandler(methods, bean, resolver, context);
+		assertThatExceptionOfType(UndeclaredThrowableException.class).isThrownBy(() ->
+				handler.getHandlerForPayload(Long.class))
+				.withCauseExactlyInstanceOf(NoSuchMethodException.class)
+				.withStackTraceContaining("No listener method found in");
+	}
+
+	private MessageHandlerMethodFactory messageHandlerFactory() {
+		DefaultMessageHandlerMethodFactory defaultFactory = new DefaultMessageHandlerMethodFactory();
+		DefaultFormattingConversionService cs = new DefaultFormattingConversionService();
+		defaultFactory.setConversionService(cs);
+		GenericMessageConverter messageConverter = new GenericMessageConverter(cs);
+		defaultFactory.setMessageConverter(messageConverter);
+		defaultFactory.afterPropertiesSet();
+		return defaultFactory;
+	}
+
+	public static class Multi {
+
+		void listen(Integer in) {
+		}
+
+	}
+
+}