Fix @ConcurrencyLimit leak for non-singleton beans

spring-aop/src/main/java/org/springframework/aop/framework/AbstractAdvisingBeanPostProcessor.java

@@ -94,17 +94,18 @@ public Object postProcessAfterInitialization(Object bean, String beanName) {
 		if (bean instanceof Advised advised) {
 			if (!advised.isFrozen() && isEligible(AopUtils.getTargetClass(bean))) {
 				// Add our local Advisor to the existing proxy's Advisor chain.
+				Advisor advisor = getAdvisor(this.advisor, bean, beanName);
 				if (this.beforeExistingAdvisors) {
-					advised.addAdvisor(0, this.advisor);
+					advised.addAdvisor(0, advisor);
 				}
 				else if (advised.getTargetSource() == AdvisedSupport.EMPTY_TARGET_SOURCE &&
 						advised.getAdvisorCount() > 0) {
 					// No target, leave last Advisor in place and add new Advisor right before.
-					advised.addAdvisor(advised.getAdvisorCount() - 1, this.advisor);
+					advised.addAdvisor(advised.getAdvisorCount() - 1, advisor);
 					return bean;
 				}
 				else {
-					advised.addAdvisor(this.advisor);
+					advised.addAdvisor(advisor);
 				}
 				return bean;
 			}
@@ -115,7 +116,7 @@ else if (advised.getTargetSource() == AdvisedSupport.EMPTY_TARGET_SOURCE &&
 			if (!proxyFactory.isProxyTargetClass() && !proxyFactory.hasUserSuppliedInterfaces()) {
 				evaluateProxyInterfaces(bean.getClass(), proxyFactory);
 			}
-			proxyFactory.addAdvisor(this.advisor);
+			proxyFactory.addAdvisor(getAdvisor(this.advisor, bean, beanName));
 			customizeProxyFactory(proxyFactory);
 			proxyFactory.setFrozen(isFrozen());
 			proxyFactory.setPreFiltered(true);
@@ -133,6 +134,21 @@ else if (advised.getTargetSource() == AdvisedSupport.EMPTY_TARGET_SOURCE &&
 		return bean;
 	}
 
+	/**
+	 * Return the {@link Advisor} to apply to the given eligible bean.
+	 * <p>The default implementation returns this post-processor's shared
+	 * {@link Advisor}. Can be overridden to return a dedicated {@link Advisor}
+	 * per bean, for example, if its advice keeps state for each proxy instance.
+	 * @param advisor this post-processor's shared {@link Advisor}
+	 * @param bean the bean instance (possibly an existing proxy)
+	 * @param beanName the name of the bean
+	 * @return the {@link Advisor} to add to the bean's proxy
+	 * @since 7.1
+	 */
+	protected Advisor getAdvisor(Advisor advisor, Object bean, String beanName) {
+		return advisor;
+	}
+
 	/**
 	 * Check whether the given bean is eligible for advising with this
 	 * post-processor's {@link Advisor}.

spring-context/src/main/java/org/springframework/resilience/annotation/ConcurrencyLimitBeanPostProcessor.java

@@ -17,16 +17,16 @@
 package org.springframework.resilience.annotation;
 
 import java.lang.reflect.Method;
-import java.util.Collections;
-import java.util.IdentityHashMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
 import org.jspecify.annotations.Nullable;
 
+import org.springframework.aop.Advisor;
 import org.springframework.aop.Pointcut;
+import org.springframework.aop.PointcutAdvisor;
 import org.springframework.aop.ProxyMethodInvocation;
 import org.springframework.aop.framework.autoproxy.AbstractBeanFactoryAwareAdvisingPostProcessor;
 import org.springframework.aop.interceptor.ConcurrencyThrottleInterceptor;
@@ -74,11 +74,20 @@ public void setEmbeddedValueResolver(StringValueResolver resolver) {
 		this.embeddedValueResolver = resolver;
 	}
 
+	@Override
+	protected Advisor getAdvisor(Advisor advisor, Object bean, String beanName) {
+		// Dedicated interceptor per proxy: its concurrency state is released with
+		// the proxy instead of being retained for every non-singleton bean instance.
+		return new DefaultPointcutAdvisor(((PointcutAdvisor) advisor).getPointcut(),
+				new ConcurrencyLimitInterceptor());
+	}
+
 
 	private class ConcurrencyLimitInterceptor implements MethodInterceptor {
 
-		private final Map<Object, ConcurrencyThrottleHolder> holderPerInstance =
-				Collections.synchronizedMap(new IdentityHashMap<>(16));
+		private final Map<Method, MethodInterceptor> methodInterceptors = new ConcurrentHashMap<>();
+
+		private @Nullable MethodInterceptor classInterceptor;
 
 		@Override
 		public @Nullable Object invoke(MethodInvocation invocation) throws Throwable {
@@ -92,23 +101,19 @@ private class ConcurrencyLimitInterceptor implements MethodInterceptor {
 			}
 			Assert.state(instance != null, "Unique instance required - use a ProxyMethodInvocation");
 
-			// Build unique ConcurrencyThrottleHolder instance per target object
-			ConcurrencyThrottleHolder holder = this.holderPerInstance.computeIfAbsent(instance,
-					k -> new ConcurrencyThrottleHolder());
-
 			// Determine method-specific interceptor instance with isolated concurrency count
-			MethodInterceptor interceptor = holder.methodInterceptors.get(method);
+			MethodInterceptor interceptor = this.methodInterceptors.get(method);
 			if (interceptor == null) {
-				synchronized (holder) {
-					interceptor = holder.methodInterceptors.get(method);
+				synchronized (this) {
+					interceptor = this.methodInterceptors.get(method);
 					if (interceptor == null) {
 						boolean perMethod = false;
 						ConcurrencyLimit annotation = AnnotatedElementUtils.findMergedAnnotation(method, ConcurrencyLimit.class);
 						if (annotation != null) {
 							perMethod = true;
 						}
 						else {
-							interceptor = holder.classInterceptor;
+							interceptor = this.classInterceptor;
 							if (interceptor == null) {
 								annotation = AnnotatedElementUtils.findMergedAnnotation(targetClass, ConcurrencyLimit.class);
 							}
@@ -124,10 +129,10 @@ private class ConcurrencyLimitInterceptor implements MethodInterceptor {
 									new RejectingConcurrencyThrottleInterceptor(concurrencyLimit, name, instance) :
 									new ResilienceConcurrencyThrottleInterceptor(concurrencyLimit, name, instance));
 							if (!perMethod) {
-								holder.classInterceptor = interceptor;
+								this.classInterceptor = interceptor;
 							}
 						}
-						holder.methodInterceptors.put(method, interceptor);
+						this.methodInterceptors.put(method, interceptor);
 					}
 				}
 			}
@@ -148,14 +153,6 @@ private int parseInt(int value, String stringValue) {
 	}
 
 
-	private static class ConcurrencyThrottleHolder {
-
-		final Map<Method, MethodInterceptor> methodInterceptors = new ConcurrentHashMap<>();
-
-		@Nullable MethodInterceptor classInterceptor;
-	}
-
-
 	private static class ResilienceConcurrencyThrottleInterceptor extends ConcurrencyThrottleInterceptor {
 
 		private final String identifier;

spring-context/src/test/java/org/springframework/resilience/ConcurrencyLimitTests.java

@@ -25,10 +25,12 @@
 import org.junit.jupiter.api.Test;
 
 import org.springframework.aop.config.AopConfigUtils;
+import org.springframework.aop.framework.Advised;
 import org.springframework.aop.framework.AopProxyUtils;
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.interceptor.ConcurrencyThrottleInterceptor;
 import org.springframework.aop.support.AopUtils;
+import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.support.DefaultListableBeanFactory;
 import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
@@ -132,6 +134,32 @@ void withPostProcessorForMethodWithPlainInterfaceAndDefaultTargetClass() {
 		assertThat(target.current).hasValue(0);
 	}
 
+	@Test
+	void withPostProcessorForPrototypeBeans() {
+		DefaultListableBeanFactory bf = new DefaultListableBeanFactory();
+		RootBeanDefinition bd = new RootBeanDefinition(AnnotatedMethodBean.class);
+		bd.setScope(BeanDefinition.SCOPE_PROTOTYPE);
+		bf.registerBeanDefinition("bean", bd);
+		ConcurrencyLimitBeanPostProcessor bpp = new ConcurrencyLimitBeanPostProcessor();
+		bpp.setBeanFactory(bf);
+		bf.addBeanPostProcessor(bpp);
+
+		AnnotatedMethodBean proxy1 = bf.getBean(AnnotatedMethodBean.class);
+		AnnotatedMethodBean proxy2 = bf.getBean(AnnotatedMethodBean.class);
+		// Concurrency state must be held per proxy, not in a shared registry
+		// which would retain every prototype instance
+		assertThat(((Advised) proxy1).getAdvisors()[0].getAdvice())
+				.isNotSameAs(((Advised) proxy2).getAdvisors()[0].getAdvice());
+
+		AnnotatedMethodBean target = (AnnotatedMethodBean) AopProxyUtils.getSingletonTarget(proxy1);
+		List<CompletableFuture<?>> futures = new ArrayList<>(10);
+		for (int i = 0; i < 10; i++) {
+			futures.add(CompletableFuture.runAsync(proxy1::concurrentOperation));
+		}
+		CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+		assertThat(target.current).hasValue(0);
+	}
+
 	@Test
 	void withPostProcessorForMethodWithUnboundedConcurrency() {
 		AnnotatedMethodBean proxy = createProxy(AnnotatedMethodBean.class);