Support for always executing specific listeners in original thread

See gh-30244

spring-context/src/main/java/org/springframework/context/ApplicationListener.java

@@ -48,6 +48,18 @@ public interface ApplicationListener<E extends ApplicationEvent> extends EventLi
 	 */
 	void onApplicationEvent(E event);
 
+	/**
+	 * Return whether this listener supports asynchronous execution.
+	 * @return {@code true} if this listener instance can be executed asynchronously
+	 * depending on the multicaster configuration (the default), or {@code false} if it
+	 * needs to immediately run within the original thread which published the event
+	 * @since 6.1
+	 * @see org.springframework.context.event.SimpleApplicationEventMulticaster#setTaskExecutor
+	 */
+	default boolean supportsAsyncExecution() {
+		return true;
+	}
+
 
 	/**
 	 * Create a new {@code ApplicationListener} for the given payload consumer.

spring-context/src/main/java/org/springframework/context/event/ApplicationEventMulticaster.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -110,17 +110,23 @@ public interface ApplicationEventMulticaster {
 	 * Multicast the given application event to appropriate listeners.
 	 * <p>Consider using {@link #multicastEvent(ApplicationEvent, ResolvableType)}
 	 * if possible as it provides better support for generics-based events.
+	 * <p>If a matching {@code ApplicationListener} does not support asynchronous
+	 * execution, it must be run within the calling thread of this multicast call.
 	 * @param event the event to multicast
+	 * @see ApplicationListener#supportsAsyncExecution()
 	 */
 	void multicastEvent(ApplicationEvent event);
 
 	/**
 	 * Multicast the given application event to appropriate listeners.
 	 * <p>If the {@code eventType} is {@code null}, a default type is built
 	 * based on the {@code event} instance.
+	 * <p>If a matching {@code ApplicationListener} does not support asynchronous
+	 * execution, it must be run within the calling thread of this multicast call.
 	 * @param event the event to multicast
 	 * @param eventType the type of event (can be {@code null})
 	 * @since 4.2
+	 * @see ApplicationListener#supportsAsyncExecution()
 	 */
 	void multicastEvent(ApplicationEvent event, @Nullable ResolvableType eventType);
 

spring-context/src/main/java/org/springframework/context/event/SimpleApplicationEventMulticaster.java

@@ -79,10 +79,15 @@ public SimpleApplicationEventMulticaster(BeanFactory beanFactory) {
 	 * to invoke each listener with.
 	 * <p>Default is equivalent to {@link org.springframework.core.task.SyncTaskExecutor},
 	 * executing all listeners synchronously in the calling thread.
-	 * <p>Consider specifying an asynchronous task executor here to not block the
-	 * caller until all listeners have been executed. However, note that asynchronous
-	 * execution will not participate in the caller's thread context (class loader,
-	 * transaction association) unless the TaskExecutor explicitly supports this.
+	 * <p>Consider specifying an asynchronous task executor here to not block the caller
+	 * until all listeners have been executed. However, note that asynchronous execution
+	 * will not participate in the caller's thread context (class loader, transaction context)
+	 * unless the TaskExecutor explicitly supports this.
+	 * <p>{@link ApplicationListener} instances which declare no support for asynchronous
+	 * execution ({@link ApplicationListener#supportsAsyncExecution()} always run within
+	 * the original thread which published the event, e.g. the transaction-synchronized
+	 * {@link org.springframework.transaction.event.TransactionalApplicationListener}.
+	 * @since 2.0
 	 * @see org.springframework.core.task.SyncTaskExecutor
 	 * @see org.springframework.core.task.SimpleAsyncTaskExecutor
 	 */
@@ -92,6 +97,7 @@ public void setTaskExecutor(@Nullable Executor taskExecutor) {
 
 	/**
 	 * Return the current task executor for this multicaster.
+	 * @since 2.0
 	 */
 	@Nullable
 	protected Executor getTaskExecutor() {
@@ -136,7 +142,7 @@ public void multicastEvent(ApplicationEvent event, @Nullable ResolvableType even
 		ResolvableType type = (eventType != null ? eventType : ResolvableType.forInstance(event));
 		Executor executor = getTaskExecutor();
 		for (ApplicationListener<?> listener : getApplicationListeners(event, type)) {
-			if (executor != null) {
+			if (executor != null && listener.supportsAsyncExecution()) {
 				executor.execute(() -> invokeListener(listener, event));
 			}
 			else {

spring-context/src/test/java/org/springframework/context/event/ApplicationContextEventTests.java

@@ -20,6 +20,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.aopalliance.intercept.MethodInvocation;
 import org.junit.jupiter.api.Test;
@@ -54,6 +55,7 @@
 import static org.assertj.core.api.Assertions.assertThatRuntimeException;
 import static org.mockito.ArgumentMatchers.isA;
 import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.willReturn;
 import static org.mockito.BDDMockito.willThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -137,19 +139,44 @@ private void multicastEvent(boolean match, Class<?> listenerType, ApplicationEve
 	public void simpleApplicationEventMulticasterWithTaskExecutor() {
 		@SuppressWarnings("unchecked")
 		ApplicationListener<ApplicationEvent> listener = mock();
+		willReturn(true).given(listener).supportsAsyncExecution();
 		ApplicationEvent evt = new ContextClosedEvent(new StaticApplicationContext());
 
 		SimpleApplicationEventMulticaster smc = new SimpleApplicationEventMulticaster();
+		AtomicBoolean invoked = new AtomicBoolean();
 		smc.setTaskExecutor(command -> {
+			invoked.set(true);
 			command.run();
 			command.run();
 		});
 		smc.addApplicationListener(listener);
 
 		smc.multicastEvent(evt);
+		assertThat(invoked.get()).isTrue();
 		verify(listener, times(2)).onApplicationEvent(evt);
 	}
 
+	@Test
+	public void simpleApplicationEventMulticasterWithTaskExecutorAndNonAsyncListener() {
+		@SuppressWarnings("unchecked")
+		ApplicationListener<ApplicationEvent> listener = mock();
+		willReturn(false).given(listener).supportsAsyncExecution();
+		ApplicationEvent evt = new ContextClosedEvent(new StaticApplicationContext());
+
+		SimpleApplicationEventMulticaster smc = new SimpleApplicationEventMulticaster();
+		AtomicBoolean invoked = new AtomicBoolean();
+		smc.setTaskExecutor(command -> {
+			invoked.set(true);
+			command.run();
+			command.run();
+		});
+		smc.addApplicationListener(listener);
+
+		smc.multicastEvent(evt);
+		assertThat(invoked.get()).isFalse();
+		verify(listener, times(1)).onApplicationEvent(evt);
+	}
+
 	@Test
 	public void simpleApplicationEventMulticasterWithException() {
 		@SuppressWarnings("unchecked")