Fix AOP for Kotlin suspending function with aspect + `@Cacheable`

This change simplifies the CacheInterceptor way of dealing with cached
coroutines, thanks to the fact that lower level support for AOP has been
introduced in c8169e5c. This fix is similar to the one applied for
`@Transactional` in gh-33095.

Closes gh-33210

integration-tests/src/test/kotlin/org/springframework/aop/framework/autoproxy/AspectJAutoProxyInterceptorKotlinIntegrationTests.kt

@@ -28,6 +28,10 @@ import org.junit.jupiter.api.Test
 import org.springframework.aop.framework.autoproxy.AspectJAutoProxyInterceptorKotlinIntegrationTests.InterceptorConfig
 import org.springframework.aop.support.StaticMethodMatcherPointcutAdvisor
 import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.cache.CacheManager
+import org.springframework.cache.annotation.Cacheable
+import org.springframework.cache.annotation.EnableCaching
+import org.springframework.cache.concurrent.ConcurrentMapCacheManager
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import org.springframework.context.annotation.EnableAspectJAutoProxy
@@ -93,9 +97,26 @@ class AspectJAutoProxyInterceptorKotlinIntegrationTests(
 		assertThat(reactiveTransactionManager.commits).`as`("transactional applied").isOne()
 	}
 
+	@Test // gh-33210
+	fun `Aspect and cacheable with suspending function`() {
+		assertThat(countingAspect.counter).isZero()
+		val value = "Hello!"
+		runBlocking {
+			assertThat(echo.suspendingCacheableEcho(value)).isEqualTo("$value 0")
+			assertThat(echo.suspendingCacheableEcho(value)).isEqualTo("$value 0")
+			assertThat(echo.suspendingCacheableEcho(value)).isEqualTo("$value 0")
+			assertThat(countingAspect.counter).`as`("aspect applied once").isOne()
+
+			assertThat(echo.suspendingCacheableEcho("$value bis")).isEqualTo("$value bis 1")
+			assertThat(echo.suspendingCacheableEcho("$value bis")).isEqualTo("$value bis 1")
+		}
+		assertThat(countingAspect.counter).`as`("aspect applied once per key").isEqualTo(2)
+	}
+
     @Configuration
     @EnableAspectJAutoProxy
     @EnableTransactionManagement
+	@EnableCaching
     open class InterceptorConfig {
 
         @Bean
@@ -112,6 +133,11 @@ class AspectJAutoProxyInterceptorKotlinIntegrationTests(
 			return ReactiveCallCountingTransactionManager()
 		}
 
+		@Bean
+		open fun cacheManager(): CacheManager {
+			return ConcurrentMapCacheManager()
+		}
+
         @Bean
         open fun echo(): Echo {
             return Echo()
@@ -155,7 +181,7 @@ class AspectJAutoProxyInterceptorKotlinIntegrationTests(
 		fun logging(joinPoint: ProceedingJoinPoint): Any {
 			return (joinPoint.proceed(joinPoint.args) as Mono<*>).doOnTerminate {
 				counter++
-			}
+			}.checkpoint("CountingAspect")
 		}
 	}
 
@@ -177,6 +203,15 @@ class AspectJAutoProxyInterceptorKotlinIntegrationTests(
 			return value
 		}
 
+		open var cacheCounter: Int = 0
+
+		@Counting
+		@Cacheable("something")
+		open suspend fun suspendingCacheableEcho(value: String): String {
+			delay(1)
+			return "$value ${cacheCounter++}"
+		}
+
     }
 
 }

spring-context/src/main/java/org/springframework/cache/interceptor/CacheInterceptor.java

@@ -19,15 +19,9 @@ package org.springframework.cache.interceptor;
 import java.io.Serializable;
 import java.lang.reflect.Method;
 
-import kotlin.coroutines.Continuation;
-import kotlin.coroutines.CoroutineContext;
-import kotlinx.coroutines.Job;
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
-import org.reactivestreams.Publisher;
 
-import org.springframework.core.CoroutinesUtils;
-import org.springframework.core.KotlinDetector;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -58,9 +52,6 @@ public class CacheInterceptor extends CacheAspectSupport implements MethodInterc
 
 		CacheOperationInvoker aopAllianceInvoker = () -> {
 			try {
-				if (KotlinDetector.isKotlinReflectPresent() && KotlinDetector.isSuspendingFunction(method)) {
-					return KotlinDelegate.invokeSuspendingFunction(method, invocation.getThis(), invocation.getArguments());
-				}
 				return invocation.proceed();
 			}
 			catch (Throwable ex) {
@@ -78,17 +69,4 @@ public class CacheInterceptor extends CacheAspectSupport implements MethodInterc
 		}
 	}
 
-
-	/**
-	 * Inner class to avoid a hard dependency on Kotlin at runtime.
-	 */
-	private static class KotlinDelegate {
-
-		public static Publisher<?> invokeSuspendingFunction(Method method, @Nullable Object target, Object... args) {
-			Continuation<?> continuation = (Continuation<?>) args[args.length - 1];
-			CoroutineContext coroutineContext = continuation.getContext().minusKey(Job.Key);
-			return CoroutinesUtils.invokeSuspendingFunction(coroutineContext, method, target, args);
-		}
-	}
-
 }