From 7f2315eacb94e9e1324b0de6b63cd686775aee8c Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 10 Mar 2025 14:20:22 -0400 Subject: [PATCH] GH-3005: Fix `SimpleMLC.killOrRestart` for closed AC Fixes: #3005 Issue link: https://github.com/spring-projects/spring-amqp/issues/3005 The `SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.killOrRestart()` is also called during application context shutdown. At this moment we cannot emit events into an application context. Otherwise, it fails with: ``` Exception in thread "rabbitListenerExecutor1" org.springframework.beans.factory.BeanCreationNotAllowedException: Error creating bean with name 'refreshEventListener': Singleton bean creation not allowed while singletons of this factory are in destruction (Do not request a bean from a BeanFactory in a destroy method implementation!) ``` * Introduce `ObservableListenerContainer.isApplicationContextClosed()` and call it as additional condition in the `SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.killOrRestart()` before trying to emit `AsyncConsumerStoppedEvent` The fix for `3.1.x` requires a slightly different approach via `ContextClosedEvent` # Conflicts: # spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ObservableListenerContainer.java --- .../rabbit/listener/ObservableListenerContainer.java | 12 ++++++++++-- .../listener/SimpleMessageListenerContainer.java | 9 ++++----- 2 files changed, 14 insertions(+), 7 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ObservableListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ObservableListenerContainer.java index e9a7b7ba..63eb5551 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ObservableListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ObservableListenerContainer.java @@ -1,5 +1,5 @@ /* - * Copyright 2023 the original author or authors. + * Copyright 2023-2025 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. @@ -24,11 +24,14 @@ import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.DisposableBean; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.lang.Nullable; import org.springframework.util.ClassUtils; /** * @author Gary Russell + * @author Artem Bilan + * * @since 3.0.5 * */ @@ -36,7 +39,7 @@ public abstract class ObservableListenerContainer extends RabbitAccessor implements MessageListenerContainer, ApplicationContextAware, BeanNameAware, DisposableBean { private static final boolean MICROMETER_PRESENT = ClassUtils.isPresent( - "io.micrometer.core.instrument.MeterRegistry", AbstractMessageListenerContainer.class.getClassLoader()); + "io.micrometer.core.instrument.MeterRegistry", AbstractMessageListenerContainer.class.getClassLoader()); private ApplicationContext applicationContext; @@ -119,6 +122,11 @@ public abstract class ObservableListenerContainer extends RabbitAccessor } } + protected boolean isApplicationContextClosed() { + return this.applicationContext instanceof ConfigurableApplicationContext configurableCtx + && configurableCtx.isClosed(); + } + @Override public void setBeanName(String beanName) { this.beanName = beanName; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index 8f4c293e..41a58509 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java @@ -61,6 +61,7 @@ import org.springframework.amqp.rabbit.support.ListenerContainerAware; import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException; import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator; import org.springframework.amqp.support.ConsumerTagStrategy; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.log.LogMessage; import org.springframework.jmx.export.annotation.ManagedMetric; import org.springframework.jmx.support.MetricType; @@ -798,7 +799,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } } - /** * Start up to delta consumers, limited by {@link #setMaxConcurrentConsumers(int)}. * @param delta the consumers to add. @@ -863,7 +863,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } } - private void considerStoppingAConsumer(BlockingQueueConsumer consumer) { this.consumersLock.lock(); try { @@ -1266,7 +1265,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private boolean failedExclusive; - AsyncMessageProcessingConsumer(BlockingQueueConsumer consumer) { this.consumer = consumer; this.start = new CountDownLatch(1); @@ -1531,8 +1529,9 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta try { this.consumer.stop(); SimpleMessageListenerContainer.this.cancellationLock.release(this.consumer); - if (getApplicationEventPublisher() != null) { - getApplicationEventPublisher().publishEvent( + ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher(); + if (applicationEventPublisher != null && !isApplicationContextClosed()) { + applicationEventPublisher.publishEvent( new AsyncConsumerStoppedEvent(SimpleMessageListenerContainer.this, this.consumer)); } }