From e32a2f339d611053c2b53f0bb3d5f322e4df24d7 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Simon=20Basl=C3=A9?= <simon.basle@broadcom.com>
Date: Fri, 27 Sep 2024 15:32:20 +0200
Subject: [PATCH] Propagate method error in some cases of reactive
 `findInCaches` errors

In a Cacheable reactive method, if an exception is propagated from
both the method and the caching infrastructure, an NPE could previously
surface due to the `CacheAspectSupport` attempting to perform an
`onErrorResume` with a `null`. This change ensures that in such a case
the user-level exception from the method is propagated instead.

Closes gh-33492
---
 .../cache/interceptor/CacheAspectSupport.java |  6 ++--
 .../annotation/ReactiveCachingTests.java      | 32 +++++++++++++++++++
 2 files changed, 36 insertions(+), 2 deletions(-)

diff --git a/spring-context/src/main/java/org/springframework/cache/interceptor/CacheAspectSupport.java b/spring-context/src/main/java/org/springframework/cache/interceptor/CacheAspectSupport.java
index 568aea4780b8..7c88d009d179 100644
--- a/spring-context/src/main/java/org/springframework/cache/interceptor/CacheAspectSupport.java
+++ b/spring-context/src/main/java/org/springframework/cache/interceptor/CacheAspectSupport.java
@@ -1141,7 +1141,8 @@ public Object findInCaches(CacheOperationContext context, Cache cache, Object ke
 							.onErrorResume(RuntimeException.class, ex -> {
 								try {
 									getErrorHandler().handleCacheGetError((RuntimeException) ex, cache, key);
-									return evaluate(null, invoker, method, contexts);
+									Object e = evaluate(null, invoker, method, contexts);
+									return (e != null ? e : Flux.error((RuntimeException) ex));
 								}
 								catch (RuntimeException exception) {
 									return Flux.error(exception);
@@ -1155,7 +1156,8 @@ public Object findInCaches(CacheOperationContext context, Cache cache, Object ke
 							.onErrorResume(RuntimeException.class, ex -> {
 								try {
 									getErrorHandler().handleCacheGetError((RuntimeException) ex, cache, key);
-									return evaluate(null, invoker, method, contexts);
+									Object e = evaluate(null, invoker, method, contexts);
+									return (e != null ? e : Mono.error((RuntimeException) ex));
 								}
 								catch (RuntimeException exception) {
 									return Mono.error(exception);
diff --git a/spring-context/src/test/java/org/springframework/cache/annotation/ReactiveCachingTests.java b/spring-context/src/test/java/org/springframework/cache/annotation/ReactiveCachingTests.java
index 6adf6a2ebf0e..c43df31de414 100644
--- a/spring-context/src/test/java/org/springframework/cache/annotation/ReactiveCachingTests.java
+++ b/spring-context/src/test/java/org/springframework/cache/annotation/ReactiveCachingTests.java
@@ -26,6 +26,7 @@
 import org.junit.jupiter.params.provider.ValueSource;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
 
 import org.springframework.cache.Cache;
 import org.springframework.cache.CacheManager;
@@ -145,6 +146,23 @@ void cacheErrorHandlerWithLoggingCacheErrorHandler() {
 		assertThat(r1).as("cacheFlux blockFirst").isEqualTo(2L);
 	}
 
+	@Test
+	void cacheErrorHandlerWithLoggingCacheErrorHandlerAndMethodError() {
+		AnnotationConfigApplicationContext ctx =
+				new AnnotationConfigApplicationContext(ExceptionCacheManager.class, ReactiveFailureCacheableService.class, ErrorHandlerCachingConfiguration.class);
+		ReactiveCacheableService service = ctx.getBean(ReactiveCacheableService.class);
+
+		Object key = new Object();
+		StepVerifier.create(service.cacheMono(key))
+				.expectErrorMessage("mono service error")
+				.verify();
+
+		key = new Object();
+		StepVerifier.create(service.cacheFlux(key))
+				.expectErrorMessage("flux service error")
+				.verify();
+	}
+
 	@Test
 	void cacheErrorHandlerWithSimpleCacheErrorHandler() {
 		AnnotationConfigApplicationContext ctx =
@@ -214,6 +232,20 @@ Flux<Long> cacheFlux(Object arg) {
 		}
 	}
 
+	@CacheConfig(cacheNames = "first")
+	static class ReactiveFailureCacheableService extends ReactiveCacheableService {
+
+		@Cacheable
+		Mono<Long> cacheMono(Object arg) {
+			return Mono.error(new IllegalStateException("mono service error"));
+		}
+
+		@Cacheable
+		Flux<Long> cacheFlux(Object arg) {
+			return Flux.error(new IllegalStateException("flux service error"));
+		}
+	}
+
 
 	@Configuration(proxyBeanMethods = false)
 	@EnableCaching