Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3005: Fix SimpleMLC.killOrRestart for closed AC

Browse Source 
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
This commit is contained in:
Artem Bilan
2025-03-10 14:20:22 -04:00
parent 88c04bb557
commit 7f2315eacb
2 changed files with 14 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ObservableListenerContainer.java
View File

@@ -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;

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

@@ -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));
}
}