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`

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

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,22 @@
 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
  *
  */
 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 @@ protected void checkObservation() {
 		}
 	}
 
+	protected boolean isApplicationContextClosed() {
+		return this.applicationContext instanceof ConfigurableApplicationContext configurableCtx
+				&& configurableCtx.isClosed();
+	}
+
 	@Override
 	public void setBeanName(String beanName) {
 		this.beanName = beanName;

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

@@ -61,6 +61,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;
@@ -798,7 +799,6 @@ protected void adjustConsumers(int deltaArg) {
 		}
 	}
 
-
 	/**
 	 * Start up to delta consumers, limited by {@link #setMaxConcurrentConsumers(int)}.
 	 * @param delta the consumers to add.
@@ -863,7 +863,6 @@ private void considerAddingAConsumer() {
 		}
 	}
 
-
 	private void considerStoppingAConsumer(BlockingQueueConsumer consumer) {
 		this.consumersLock.lock();
 		try {
@@ -1266,7 +1265,6 @@ private final class AsyncMessageProcessingConsumer implements Runnable {
 
 		private boolean failedExclusive;
 
-
 		AsyncMessageProcessingConsumer(BlockingQueueConsumer consumer) {
 			this.consumer = consumer;
 			this.start = new CountDownLatch(1);
@@ -1531,8 +1529,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));
 					}
 				}