Trigger cancellation on context close for non-managed objects only

Specifically for prototype/scoped beans and FactoryBean-exposed objects.

Closes gh-33009

spring-context/src/main/java/org/springframework/scheduling/annotation/ScheduledAnnotationBeanPostProcessor.java

@@ -44,6 +44,7 @@
 import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.beans.factory.SmartInitializingSingleton;
 import org.springframework.beans.factory.config.DestructionAwareBeanPostProcessor;
+import org.springframework.beans.factory.config.SingletonBeanRegistry;
 import org.springframework.beans.factory.support.MergedBeanDefinitionPostProcessor;
 import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.context.ApplicationContext;
@@ -155,6 +156,8 @@ public class ScheduledAnnotationBeanPostProcessor
 
 	private final Map<Object, List<Runnable>> reactiveSubscriptions = new IdentityHashMap<>(16);
 
+	private final Set<Object> manualCancellationOnContextClose = Collections.newSetFromMap(new IdentityHashMap<>(16));
+
 
 	/**
 	 * Create a default {@code ScheduledAnnotationBeanPostProcessor}.
@@ -305,6 +308,12 @@ public Object postProcessAfterInitialization(Object bean, String beanName) {
 					logger.trace(annotatedMethods.size() + " @Scheduled methods processed on bean '" + beanName +
 							"': " + annotatedMethods);
 				}
+				if ((this.beanFactory != null && !this.beanFactory.isSingleton(beanName)) ||
+						(this.beanFactory instanceof SingletonBeanRegistry sbr && sbr.containsSingleton(beanName))) {
+					// Either a prototype/scoped bean or a FactoryBean with a pre-existing managed singleton
+					// -> trigger manual cancellation when ContextClosedEvent comes in
+					this.manualCancellationOnContextClose.add(bean);
+				}
 			}
 		}
 		return bean;
@@ -595,6 +604,18 @@ public Set<ScheduledTask> getScheduledTasks() {
 
 	@Override
 	public void postProcessBeforeDestruction(Object bean, String beanName) {
+		cancelScheduledTasks(bean);
+		this.manualCancellationOnContextClose.remove(bean);
+	}
+
+	@Override
+	public boolean requiresDestruction(Object bean) {
+		synchronized (this.scheduledTasks) {
+			return (this.scheduledTasks.containsKey(bean) || this.reactiveSubscriptions.containsKey(bean));
+		}
+	}
+
+	private void cancelScheduledTasks(Object bean) {
 		Set<ScheduledTask> tasks;
 		List<Runnable> liveSubscriptions;
 		synchronized (this.scheduledTasks) {
@@ -613,13 +634,6 @@ public void postProcessBeforeDestruction(Object bean, String beanName) {
 		}
 	}
 
-	@Override
-	public boolean requiresDestruction(Object bean) {
-		synchronized (this.scheduledTasks) {
-			return (this.scheduledTasks.containsKey(bean) || this.reactiveSubscriptions.containsKey(bean));
-		}
-	}
-
 	@Override
 	public void destroy() {
 		synchronized (this.scheduledTasks) {
@@ -636,7 +650,10 @@ public void destroy() {
 					liveSubscription.run();  // equivalent to cancelling the subscription
 				}
 			}
+			this.reactiveSubscriptions.clear();
+			this.manualCancellationOnContextClose.clear();
 		}
+
 		this.registrar.destroy();
 		if (this.localScheduler != null) {
 			this.localScheduler.destroy();
@@ -659,15 +676,10 @@ public void onApplicationEvent(ApplicationContextEvent event) {
 				finishRegistration();
 			}
 			else if (event instanceof ContextClosedEvent) {
-				synchronized (this.scheduledTasks) {
-					Collection<Set<ScheduledTask>> allTasks = this.scheduledTasks.values();
-					for (Set<ScheduledTask> tasks : allTasks) {
-						for (ScheduledTask task : tasks) {
-							// At this early point, let in-progress tasks complete still
-							task.cancel(false);
-						}
-					}
+				for (Object bean : this.manualCancellationOnContextClose) {
+					cancelScheduledTasks(bean);
 				}
+				this.manualCancellationOnContextClose.clear();
 			}
 		}
 	}