Handle low-level errors for sync/flux/mono/future gets

This change adds 3 protected methods to `AbstractCacheInvoker` that wrap
additional `Cache#retrieve` and `Cache#get` calls with
`handleCacheGetError` in case the Cache call itself fails.

For example, if the cache is remote and a connection to it cannot be
established.

Closes gh-21590

Author: simonbasle

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

@@ -16,6 +16,10 @@
 
 package org.springframework.cache.interceptor;
 
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
+
 import org.springframework.cache.Cache;
 import org.springframework.lang.Nullable;
 import org.springframework.util.function.SingletonSupplier;
@@ -78,6 +82,52 @@ protected Cache.ValueWrapper doGet(Cache cache, Object key) {
 		}
 	}
 
+	@Nullable
+	protected <T> T doGet(Cache cache, Object key, Callable<T> valueLoader) {
+		try {
+			return cache.get(key, valueLoader);
+		}
+		catch (Cache.ValueRetrievalException ex) {
+			throw ex;
+		}
+		catch (RuntimeException ex) {
+			getErrorHandler().handleCacheGetError(ex, cache, key);
+			try {
+				return valueLoader.call();
+			}
+			catch (Exception ex2) {
+				throw new RuntimeException(ex2);
+			}
+		}
+	}
+
+	@Nullable
+	protected CompletableFuture<?> doRetrieve(Cache cache, Object key) {
+		try {
+			return cache.retrieve(key);
+		}
+		catch (Cache.ValueRetrievalException ex) {
+			throw ex;
+		}
+		catch (RuntimeException ex) {
+			getErrorHandler().handleCacheGetError(ex, cache, key);
+			return null;
+		}
+	}
+
+	protected <T> CompletableFuture<T> doRetrieve(Cache cache, Object key, Supplier<CompletableFuture<T>> valueLoader) {
+		try {
+			return cache.retrieve(key, valueLoader);
+		}
+		catch (Cache.ValueRetrievalException ex) {
+			throw ex;
+		}
+		catch (RuntimeException ex) {
+			getErrorHandler().handleCacheGetError(ex, cache, key);
+			return valueLoader.get();
+		}
+	}
+
 	/**
 	 * Execute {@link Cache#put(Object, Object)} on the specified {@link Cache}
 	 * and invoke the error handler if an exception occurs.

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

@@ -456,7 +456,7 @@ private Object executeSynchronized(CacheOperationInvoker invoker, Method method,
 			Object key = generateKey(context, CacheOperationExpressionEvaluator.NO_RESULT);
 			Cache cache = context.getCaches().iterator().next();
 			if (CompletableFuture.class.isAssignableFrom(method.getReturnType())) {
-				return cache.retrieve(key, () -> (CompletableFuture<?>) invokeOperation(invoker));
+				return doRetrieve(cache, key, () -> (CompletableFuture<?>) invokeOperation(invoker));
 			}
 			if (this.reactiveCachingHandler != null) {
 				Object returnValue = this.reactiveCachingHandler.executeSynchronized(invoker, method, cache, key);
@@ -465,7 +465,7 @@ private Object executeSynchronized(CacheOperationInvoker invoker, Method method,
 				}
 			}
 			try {
-				return wrapCacheValue(method, cache.get(key, () -> unwrapReturnValue(invokeOperation(invoker))));
+				return wrapCacheValue(method, doGet(cache, key, () -> unwrapReturnValue(invokeOperation(invoker))));
 			}
 			catch (Cache.ValueRetrievalException ex) {
 				// Directly propagate ThrowableWrapper from the invoker,
@@ -515,7 +515,7 @@ private Object findInCaches(CacheOperationContext context, Object key,
 
 		for (Cache cache : context.getCaches()) {
 			if (CompletableFuture.class.isAssignableFrom(context.getMethod().getReturnType())) {
-				CompletableFuture<?> result = cache.retrieve(key);
+				CompletableFuture<?> result = doRetrieve(cache, key);
 				if (result != null) {
 					return result.exceptionally(ex -> {
 						getErrorHandler().handleCacheGetError((RuntimeException) ex, cache, key);
@@ -1144,7 +1144,7 @@ public Object findInCaches(CacheOperationContext context, Cache cache, Object ke
 
 			ReactiveAdapter adapter = this.registry.getAdapter(context.getMethod().getReturnType());
 			if (adapter != null) {
-				CompletableFuture<?> cachedFuture = cache.retrieve(key);
+				CompletableFuture<?> cachedFuture = doRetrieve(cache, key);
 				if (cachedFuture == null) {
 					return null;
 				}

spring-context/src/test/java/org/springframework/cache/interceptor/CacheErrorHandlerTests.java

@@ -17,11 +17,15 @@
 package org.springframework.cache.interceptor;
 
 import java.util.Collections;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import org.springframework.cache.Cache;
 import org.springframework.cache.CacheManager;
@@ -39,6 +43,8 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willReturn;
 import static org.mockito.BDDMockito.willThrow;
@@ -83,11 +89,56 @@ void getFail() {
 		willThrow(exception).given(this.cache).get(0L);
 
 		Object result = this.simpleService.get(0L);
-		verify(this.errorHandler).handleCacheGetError(exception, cache, 0L);
+		verify(this.errorHandler).handleCacheGetError(exception, this.cache, 0L);
 		verify(this.cache).get(0L);
 		verify(this.cache).put(0L, result); // result of the invocation
 	}
 
+	@Test
+	public void getSyncFail() {
+		UnsupportedOperationException exception = new UnsupportedOperationException("Test exception on get");
+		willThrow(exception).given(this.cache).get(eq(0L), any(Callable.class));
+
+		Object result = this.simpleService.getSync(0L);
+		assertThat(result).isEqualTo(0L);
+		verify(this.errorHandler).handleCacheGetError(exception, this.cache, 0L);
+		verify(this.cache).get(eq(0L), any(Callable.class));
+	}
+
+	@Test
+	public void getCompletableFutureFail() {
+		UnsupportedOperationException exception = new UnsupportedOperationException("Test exception on get");
+		willThrow(exception).given(this.cache).retrieve(eq(0L));
+
+		Object result = this.simpleService.getFuture(0L).join();
+		assertThat(result).isEqualTo(0L);
+		verify(this.errorHandler).handleCacheGetError(exception, this.cache, 0L);
+		verify(this.cache).retrieve(eq(0L));
+	}
+
+	@Test
+	public void getMonoFail() {
+		UnsupportedOperationException exception = new UnsupportedOperationException("Test exception on get");
+		willThrow(exception).given(this.cache).retrieve(eq(0L));
+
+		Object result = this.simpleService.getMono(0L).block();
+		assertThat(result).isEqualTo(0L);
+		verify(this.errorHandler).handleCacheGetError(exception, this.cache, 0L);
+		verify(this.cache).retrieve(eq(0L));
+	}
+
+
+	@Test
+	public void getFluxFail() {
+		UnsupportedOperationException exception = new UnsupportedOperationException("Test exception on get");
+		willThrow(exception).given(this.cache).retrieve(eq(0L));
+
+		Object result = this.simpleService.getFlux(0L).blockLast();
+		assertThat(result).isEqualTo(0L);
+		verify(this.errorHandler).handleCacheGetError(exception, this.cache, 0L);
+		verify(this.cache).retrieve(eq(0L));
+	}
+
 	@Test
 	void getAndPutFail() {
 		UnsupportedOperationException exception = new UnsupportedOperationException("Test exception on get");
@@ -220,6 +271,26 @@ public Object get(long id) {
 			return this.counter.getAndIncrement();
 		}
 
+		@Cacheable(sync = true)
+		public Object getSync(long id) {
+			return this.counter.getAndIncrement();
+		}
+
+		@Cacheable
+		public CompletableFuture<Long> getFuture(long id) {
+			return CompletableFuture.completedFuture(this.counter.getAndIncrement());
+		}
+
+		@Cacheable
+		public Mono<Long> getMono(long id) {
+			return Mono.just(this.counter.getAndIncrement());
+		}
+
+		@Cacheable
+		public Flux<Long> getFlux(long id) {
+			return Flux.just(this.counter.getAndIncrement(), 0L);
+		}
+
 		@CachePut
 		public Object put(long id) {
 			return this.counter.getAndIncrement();