GH-3393: Fix RabbitAmqpListenerContainer for fatal errors

Fixes: #3393

If `ErrorHandler` is configured on the container to rethrow an exception,
the final decision must be done in the listener container itself.

* Use `ConditionalRejectingErrorHandler.DefaultExceptionStrategy`
in the `SimpleMessageListenerContainer` and `RabbitAmqpListenerContainer`
to determine if listener container has to be stopped due to fatal error, e.g. `NoSuchMethodException`
* Fix `RabbitAmqpListenerContainer` to emit `ListenerContainerConsumerFailedEvent`
in case of forced stop
* Propagate `ErrorHandler` and `ApplicationEventPublisher` from the `RabbitAmqpListenerContainerFactory`
* Use `log4j-slf4j-impl` dependency instead of `log4j-slf4j2-impl`
to satisfy third-party libraries
* Make `spotbugs-annotations` as `compileOnly` and `testImplementation` dependency

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

Author: artembilan

build.gradle

@@ -180,8 +180,8 @@ configure(javaProjects) { subproject ->
 	// dependencies that are common across all java projects
 	dependencies {
 		def spotBugs = 'com.github.spotbugs:spotbugs-annotations:4.9.8'
-		runtimeOnly spotBugs
-		testRuntimeOnly spotBugs
+		compileOnly spotBugs
+		testImplementation spotBugs
 
 		testImplementation 'org.apache.logging.log4j:log4j-core'
 		testImplementation "org.hamcrest:hamcrest-core:$hamcrestVersion"
@@ -198,7 +198,7 @@ configure(javaProjects) { subproject ->
 		testImplementation "com.willowtreeapps.assertk:assertk-jvm:$assertkVersion"
 
 		testRuntimeOnly 'org.apache.logging.log4j:log4j-jcl'
-		testRuntimeOnly 'org.apache.logging.log4j:log4j-slf4j2-impl'
+		testRuntimeOnly 'org.apache.logging.log4j:log4j-slf4j-impl'
 	}
 
 	// enable all compiler warnings; individual projects may customize further

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

@@ -142,6 +142,9 @@ public abstract class AbstractMessageListenerContainer extends ObservableListene
 
 	protected final AtomicBoolean stopNow = new AtomicBoolean(); // NOSONAR
 
+	protected final FatalExceptionStrategy exceptionStrategy =
+			new ConditionalRejectingErrorHandler.DefaultExceptionStrategy();
+
 	private ContainerDelegate proxy = this.delegate;
 
 	private long shutdownTimeout = DEFAULT_SHUTDOWN_TIMEOUT;

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

@@ -1486,7 +1486,7 @@ private void mainLoop() throws Exception { // NOSONAR Exception
 			}
 			catch (ListenerExecutionFailedException ex) {
 				// Continue to process, otherwise re-throw
-				if (causeContainsNoSuchMethod(ex.getCause())) {
+				if (SimpleMessageListenerContainer.this.exceptionStrategy.isFatal(ex)) {
 					throw new FatalListenerExecutionException("Invalid listener", ex);
 				}
 			}
@@ -1651,22 +1651,6 @@ else if (logger.isWarnEnabled()) {
 			publishConsumerFailedEvent("Consumer raised exception, attempting restart", false, t);
 		}
 
-		private static boolean causeContainsNoSuchMethod(@Nullable Throwable throwable) {
-			if (throwable instanceof NoSuchMethodException) {
-				return true;
-			}
-
-			if (throwable != null) {
-				Throwable cause = throwable.getCause();
-				if (cause == null || cause == throwable) {
-					return false;
-				}
-				return causeContainsNoSuchMethod(cause);
-			}
-
-			return false;
-		}
-
 	}
 
 }

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/config/RabbitAmqpListenerContainerFactory.java

@@ -31,7 +31,10 @@
 import org.springframework.amqp.rabbitmq.client.listener.RabbitAmqpMessageListenerAdapter;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.utils.JavaUtils;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.scheduling.TaskScheduler;
+import org.springframework.util.ErrorHandler;
 
 /**
  * Factory for {@link RabbitAmqpListenerContainer}.
@@ -44,7 +47,8 @@
  *
  */
 public class RabbitAmqpListenerContainerFactory
-		extends BaseRabbitListenerContainerFactory<RabbitAmqpListenerContainer> {
+		extends BaseRabbitListenerContainerFactory<RabbitAmqpListenerContainer>
+		implements ApplicationEventPublisherAware {
 
 	private final AmqpConnectionFactory connectionFactory;
 
@@ -60,6 +64,10 @@ public class RabbitAmqpListenerContainerFactory
 
 	private @Nullable MessageConverter messageConverter;
 
+	private @Nullable ErrorHandler errorHandler;
+
+	private @Nullable ApplicationEventPublisher applicationEventPublisher;
+
 	/**
 	 * Construct an instance using the provided {@link AmqpConnectionFactory}.
 	 * @param connectionFactory the connection.
@@ -68,6 +76,11 @@ public RabbitAmqpListenerContainerFactory(AmqpConnectionFactory connectionFactor
 		this.connectionFactory = connectionFactory;
 	}
 
+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		this.applicationEventPublisher = applicationEventPublisher;
+	}
+
 	/**
 	 * Set a {@link ContainerCustomizer} that is invoked after a container is created and
 	 * configured to enable further customization of the container.
@@ -130,6 +143,16 @@ public void setMessageConverter(MessageConverter messageConverter) {
 		this.messageConverter = messageConverter;
 	}
 
+	/**
+	 * Set the error handler for the container.
+	 * @param errorHandler The error handler.
+	 * @since 4.0.3
+	 * @see RabbitAmqpListenerContainer#setErrorHandler(ErrorHandler)
+	 */
+	public void setErrorHandler(ErrorHandler errorHandler) {
+		this.errorHandler = errorHandler;
+	}
+
 	@Override
 	public RabbitAmqpListenerContainer createListenerContainer(@Nullable RabbitListenerEndpoint endpoint) {
 		if (endpoint instanceof MethodRabbitListenerEndpoint methodRabbitListenerEndpoint) {
@@ -152,7 +175,9 @@ public RabbitAmqpListenerContainer createListenerContainer(@Nullable RabbitListe
 				.acceptIfNotNull(this.afterReceivePostProcessors, container::setAfterReceivePostProcessors)
 				.acceptIfNotNull(this.batchSize, container::setBatchSize)
 				.acceptIfNotNull(this.batchReceiveTimeout, container::setBatchReceiveTimeout)
-				.acceptIfNotNull(this.taskScheduler, container::setTaskScheduler);
+				.acceptIfNotNull(this.taskScheduler, container::setTaskScheduler)
+				.acceptIfNotNull(this.errorHandler, container::setErrorHandler)
+				.acceptIfNotNull(this.applicationEventPublisher, container::setApplicationEventPublisher);
 
 		applyCommonOverrides(endpoint, container);
 

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerContainer.java

@@ -43,13 +43,19 @@
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.listener.ConditionalRejectingErrorHandler;
 import org.springframework.amqp.listener.ContainerUtils;
+import org.springframework.amqp.listener.FatalExceptionStrategy;
+import org.springframework.amqp.rabbit.listener.ListenerContainerConsumerFailedEvent;
+import org.springframework.amqp.rabbit.listener.ListenerContainerConsumerTerminatedEvent;
 import org.springframework.amqp.rabbitmq.client.AmqpConnectionFactory;
 import org.springframework.amqp.rabbitmq.client.RabbitAmqpUtils;
 import org.springframework.amqp.support.postprocessor.MessagePostProcessorUtils;
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.support.DefaultPointcutAdvisor;
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.DisposableBean;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -69,7 +75,8 @@
  */
 @SuppressWarnings("removal")
 public class RabbitAmqpListenerContainer
-		implements org.springframework.amqp.rabbit.listener.MessageListenerContainer, BeanNameAware, DisposableBean {
+		implements org.springframework.amqp.rabbit.listener.MessageListenerContainer, BeanNameAware, DisposableBean,
+		ApplicationEventPublisherAware {
 
 	private static final LogAccessor LOG = new LogAccessor(LogFactory.getLog(RabbitAmqpListenerContainer.class));
 
@@ -79,6 +86,11 @@ public class RabbitAmqpListenerContainer
 
 	private final MultiValueMap<String, Consumer> queueToConsumers = new LinkedMultiValueMap<>();
 
+	private final FatalExceptionStrategy exceptionStrategy =
+			new ConditionalRejectingErrorHandler.DefaultExceptionStrategy();
+
+	private @Nullable ApplicationEventPublisher applicationEventPublisher;
+
 	private String @Nullable [] queues;
 
 	private Advice @Nullable [] adviceChain;
@@ -129,6 +141,11 @@ public RabbitAmqpListenerContainer(AmqpConnectionFactory connectionFactory) {
 		this.connectionFactory = connectionFactory;
 	}
 
+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		this.applicationEventPublisher = applicationEventPublisher;
+	}
+
 	@Override
 	public void setQueueNames(String... queueNames) {
 		this.queues = Arrays.copyOf(queueNames, queueNames.length);
@@ -446,6 +463,12 @@ private void handleListenerError(Exception ex, Consumer.Context context, Object
 	}
 
 	private boolean handleSpecialErrors(Exception ex, Consumer.Context context) {
+		if (this.exceptionStrategy.isFatal(ex)) {
+			stop();
+			context.requeue();
+			publishConsumerFailedEvent("Consumer received fatal exception during processing", true, ex);
+			return true;
+		}
 		if (ContainerUtils.shouldRequeue(this.defaultRequeue, ex, LOG.getLog())) {
 			context.requeue();
 			return true;
@@ -461,6 +484,16 @@ private boolean handleSpecialErrors(Exception ex, Consumer.Context context) {
 		return false;
 	}
 
+	private void publishConsumerFailedEvent(@Nullable String reason, boolean fatal, @Nullable Throwable t) {
+		if (this.applicationEventPublisher != null) {
+			ApplicationEvent event =
+					t == null
+							? new ListenerContainerConsumerTerminatedEvent(this, reason) :
+							new ListenerContainerConsumerFailedEvent(this, reason, t, fatal);
+			this.applicationEventPublisher.publishEvent(event);
+		}
+	}
+
 	@Override
 	public void stop() {
 		stop(() -> {

spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerTests.java

@@ -34,12 +34,15 @@
 import org.springframework.amqp.core.Binding;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageListenerContainer;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.core.QueueBuilder;
 import org.springframework.amqp.rabbit.annotation.EnableRabbit;
+import org.springframework.amqp.rabbit.annotation.RabbitHandler;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor;
+import org.springframework.amqp.rabbit.listener.ListenerContainerConsumerFailedEvent;
 import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
 import org.springframework.amqp.rabbitmq.client.AmqpConnectionFactory;
 import org.springframework.amqp.rabbitmq.client.RabbitAmqpTestBase;
@@ -49,14 +52,17 @@
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
 import org.springframework.amqp.utils.test.TestUtils;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationListener;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.messaging.handler.annotation.SendTo;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.util.MultiValueMap;
+import org.springframework.util.ReflectionUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 /**
  * @author Artem Bilan
@@ -184,6 +190,16 @@ void verifyReplyOnAnotherQueue() {
 				.isEqualTo("Reply for 'test data4' via 'e1' and 'k4'");
 	}
 
+	@Test
+	void noSuchMethodErrorStopsContainer(@Autowired NoSuchMethodService noSuchMethodService)
+			throws InterruptedException {
+
+		this.template.convertAndSend("no.such.method", "test data");
+		MessageListenerContainer container = this.rabbitListenerEndpointRegistry.getListenerContainer("no.such.method");
+		assertThat(noSuchMethodService.eventLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		await().until(() -> !container.isRunning());
+	}
+
 	@Configuration
 	@EnableRabbit
 	static class Config {
@@ -318,6 +334,44 @@ String replyViaSendTo(String data) {
 			return "Reply for '%s' via 'e1' and 'k4'".formatted(data);
 		}
 
+		@Bean
+		public RabbitAmqpListenerContainerFactory noSuchMethodListenerContainerFactory(
+				AmqpConnectionFactory connectionFactory) {
+
+			RabbitAmqpListenerContainerFactory rabbitAmqpListenerContainerFactory =
+					new RabbitAmqpListenerContainerFactory(connectionFactory);
+			rabbitAmqpListenerContainerFactory.setErrorHandler(ReflectionUtils::rethrowRuntimeException);
+			return rabbitAmqpListenerContainerFactory;
+		}
+
+		@Bean
+		Queue noSuchMethodQueue() {
+			return new Queue("no.such.method");
+		}
+
+		@Bean
+		NoSuchMethodService noSuchMethodService() {
+			return new NoSuchMethodService();
+		}
+
+	}
+
+	@RabbitListener(id = "no.such.method", queues = "no.such.method",
+			containerFactory = "noSuchMethodListenerContainerFactory")
+	static class NoSuchMethodService implements ApplicationListener<ListenerContainerConsumerFailedEvent> {
+
+		CountDownLatch eventLatch = new CountDownLatch(1);
+
+		@RabbitHandler
+		void notReachableHandle(Message message) {
+			throw new RuntimeException("Should not be thrown");
+		}
+
+		@Override
+		public void onApplicationEvent(ListenerContainerConsumerFailedEvent event) {
+			this.eventLatch.countDown();
+		}
+
 	}
 
 }