Introduce "spring.cache.reactivestreams.ignore" escape hatch

Closes gh-31861

spring-context-support/src/main/java/org/springframework/cache/caffeine/CaffeineCacheManager.java

@@ -204,10 +204,17 @@ public void setAsyncCacheLoader(AsyncCacheLoader<Object, Object> cacheLoader) {
 	 * {@link #setAllowNullValues setAllowNullValues(false)}. This makes the semantics
 	 * of CompletableFuture-based access simpler and optimizes retrieval performance
 	 * since a Caffeine-provided CompletableFuture handle does not have to get wrapped.
+	 * <p>If you come here for the adaptation of reactive types such as a Reactor
+	 * {@code Mono} or {@code Flux} onto asynchronous caching, we recommend the standard
+	 * arrangement for caching the produced values asynchronously in 6.1 through enabling
+	 * this Caffeine mode. If this is not immediately possible/desirable for existing
+	 * apps, you may set the system property "spring.cache.reactivestreams.ignore=true"
+	 * to restore 6.0 behavior where reactive handles are treated as regular values.
 	 * @since 6.1
 	 * @see Caffeine#buildAsync()
 	 * @see Cache#retrieve(Object)
 	 * @see Cache#retrieve(Object, Supplier)
+	 * @see org.springframework.cache.interceptor.CacheAspectSupport#IGNORE_REACTIVESTREAMS_PROPERTY_NAME
 	 */
 	public void setAsyncCacheMode(boolean asyncCacheMode) {
 		if (this.asyncCacheMode != asyncCacheMode) {

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

@@ -52,6 +52,7 @@
 import org.springframework.core.KotlinDetector;
 import org.springframework.core.ReactiveAdapter;
 import org.springframework.core.ReactiveAdapterRegistry;
+import org.springframework.core.SpringProperties;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
 import org.springframework.lang.Nullable;
@@ -94,9 +95,31 @@
 public abstract class CacheAspectSupport extends AbstractCacheInvoker
 		implements BeanFactoryAware, InitializingBean, SmartInitializingSingleton {
 
+	/**
+	 * System property that instructs Spring's caching infrastructure to ignore the
+	 * presence of Reactive Streams, in particular Reactor's {@link Mono}/{@link Flux}
+	 * in {@link org.springframework.cache.annotation.Cacheable} method return type
+	 * declarations.
+	 * <p>By default, as of 6.1, Reactive Streams Publishers such as Reactor's
+	 * {@link Mono}/{@link Flux} will be specifically processed for asynchronous
+	 * caching of their produced values rather than trying to cache the returned
+	 * {@code Publisher} instances themselves.
+	 * <p>Switch this flag to "true" in order to ignore Reactive Streams Publishers
+	 * and process them as regular return values through synchronous caching,
+	 * restoring 6.0 behavior. Note that this is not recommended and only works in
+	 * very limited scenarios, e.g. with manual `Mono.cache()`/`Flux.cache()` calls.
+	 * @since 6.1.3
+	 * @see org.reactivestreams.Publisher
+	 */
+	public static final String IGNORE_REACTIVESTREAMS_PROPERTY_NAME = "spring.cache.reactivestreams.ignore";
+
+	private static final boolean shouldIgnoreReactiveStreams =
+			SpringProperties.getFlag(IGNORE_REACTIVESTREAMS_PROPERTY_NAME);
+
 	private static final boolean reactiveStreamsPresent = ClassUtils.isPresent(
 			"org.reactivestreams.Publisher", CacheAspectSupport.class.getClassLoader());
 
+
 	protected final Log logger = LogFactory.getLog(getClass());
 
 	private final Map<CacheOperationCacheKey, CacheOperationMetadata> metadataCache = new ConcurrentHashMap<>(1024);
@@ -124,7 +147,8 @@ public abstract class CacheAspectSupport extends AbstractCacheInvoker
 
 
 	protected CacheAspectSupport() {
-		this.reactiveCachingHandler = (reactiveStreamsPresent ? new ReactiveCachingHandler() : null);
+		this.reactiveCachingHandler =
+				(reactiveStreamsPresent && !shouldIgnoreReactiveStreams ? new ReactiveCachingHandler() : null);
 	}