GH-3005: Fix SimpleMLC.killOrRestart for closed AC

Fixes: #3005
Issue link: #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`

# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ObservableListenerContainer.java

Author: artembilan

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,19 +24,24 @@
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.event.ContextClosedEvent;
 import org.springframework.lang.Nullable;
 import org.springframework.util.ClassUtils;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 3.0.5
  *
  */
 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;
 
@@ -52,6 +57,8 @@ public abstract class ObservableListenerContainer extends RabbitAccessor
 
 	private String listenerId;
 
+	private volatile boolean contextClosed;
+
 	@Nullable
 	protected final ApplicationContext getApplicationContext() {
 		return this.applicationContext;
@@ -60,6 +67,9 @@ protected final ApplicationContext getApplicationContext() {
 	@Override
 	public final void setApplicationContext(ApplicationContext applicationContext) {
 		this.applicationContext = applicationContext;
+		if (applicationContext instanceof ConfigurableApplicationContext configurableApplicationContext) {
+			configurableApplicationContext.addApplicationListener(new ApplicationClosedListener());
+		}
 	}
 
 	protected MicrometerHolder getMicrometerHolder() {
@@ -119,6 +129,10 @@ protected void checkObservation() {
 		}
 	}
 
+	protected boolean isApplicationContextClosed() {
+		return this.contextClosed;
+	}
+
 	@Override
 	public void setBeanName(String beanName) {
 		this.beanName = beanName;
@@ -152,4 +166,15 @@ public void destroy() {
 		}
 	}
 
+	private final class ApplicationClosedListener implements ApplicationListener<ContextClosedEvent> {
+
+		@Override
+		public void onApplicationEvent(ContextClosedEvent event) {
+			if (event.getApplicationContext().equals(ObservableListenerContainer.this.applicationContext)) {
+				ObservableListenerContainer.this.contextClosed = true;
+			}
+		}
+
+	}
+
 }

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

@@ -57,6 +57,7 @@
 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;
@@ -797,7 +798,6 @@ protected void adjustConsumers(int deltaArg) {
 		}
 	}
 
-
 	/**
 	 * Start up to delta consumers, limited by {@link #setMaxConcurrentConsumers(int)}.
 	 * @param delta the consumers to add.
@@ -862,7 +862,6 @@ private void considerAddingAConsumer() {
 		}
 	}
 
-
 	private void considerStoppingAConsumer(BlockingQueueConsumer consumer) {
 		this.consumersLock.lock();
 		try {
@@ -1262,7 +1261,6 @@ private final class AsyncMessageProcessingConsumer implements Runnable {
 
 		private boolean failedExclusive;
 
-
 		AsyncMessageProcessingConsumer(BlockingQueueConsumer consumer) {
 			this.consumer = consumer;
 			this.start = new CountDownLatch(1);
@@ -1527,8 +1525,9 @@ private void killOrRestart(boolean aborted) {
 				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));
 					}
 				}