Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Consistent CacheErrorHandler processing for @Cacheable(sync=true)

Browse Source 
Closes gh-34708
This commit is contained in:
Juergen Hoeller
2025-04-04 00:22:12 +02:00
parent e7db15b325
commit 4e5979c75a
3 changed files with 298 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

277
spring-context/src/test/java/org/springframework/cache/annotation/ReactiveCachingTests.java vendored
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2024 the original author or authors.
* Copyright 2002-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -19,7 +19,9 @@ package org.springframework.cache.annotation;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -40,6 +42,7 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.lang.Nullable;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.assertj.core.api.AssertionsForClassTypes.catchThrowable;
/**
@@ -58,8 +61,8 @@ class ReactiveCachingTests {
LateCacheHitDeterminationWithValueWrapperConfig.class})
void cacheHitDetermination(Class<?> configClass) {
AnnotationConfigApplicationContext ctx =
new AnnotationConfigApplicationContext(configClass, ReactiveCacheableService.class);
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(
configClass, ReactiveCacheableService.class);
ReactiveCacheableService service = ctx.getBean(ReactiveCacheableService.class);
Object key = new Object();
@@ -119,68 +122,6 @@ class ReactiveCachingTests {
ctx.close();
}
@Test
void cacheErrorHandlerWithLoggingCacheErrorHandler() {
AnnotationConfigApplicationContext ctx =
new AnnotationConfigApplicationContext(ExceptionCacheManager.class, ReactiveCacheableService.class, ErrorHandlerCachingConfiguration.class);
ReactiveCacheableService service = ctx.getBean(ReactiveCacheableService.class);
Object key = new Object();
Long r1 = service.cacheFuture(key).join();
assertThat(r1).isNotNull();
assertThat(r1).as("cacheFuture").isEqualTo(0L);
key = new Object();
r1 = service.cacheMono(key).block();
assertThat(r1).isNotNull();
assertThat(r1).as("cacheMono").isEqualTo(1L);
key = new Object();
r1 = service.cacheFlux(key).blockFirst();
assertThat(r1).isNotNull();
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 =
new AnnotationConfigApplicationContext(ExceptionCacheManager.class, ReactiveCacheableService.class);
ReactiveCacheableService service = ctx.getBean(ReactiveCacheableService.class);
Throwable completableFuturThrowable = catchThrowable(() -> service.cacheFuture(new Object()).join());
assertThat(completableFuturThrowable).isInstanceOf(CompletionException.class)
.extracting(Throwable::getCause)
.isInstanceOf(UnsupportedOperationException.class);
Throwable monoThrowable = catchThrowable(() -> service.cacheMono(new Object()).block());
assertThat(monoThrowable).isInstanceOf(UnsupportedOperationException.class);
Throwable fluxThrowable = catchThrowable(() -> service.cacheFlux(new Object()).blockFirst());
assertThat(fluxThrowable).isInstanceOf(UnsupportedOperationException.class);
}
@ParameterizedTest
@ValueSource(classes = {EarlyCacheHitDeterminationConfig.class,
EarlyCacheHitDeterminationWithoutNullValuesConfig.class,
@@ -188,8 +129,8 @@ class ReactiveCachingTests {
LateCacheHitDeterminationWithValueWrapperConfig.class})
void fluxCacheDoesntDependOnFirstRequest(Class<?> configClass) {
AnnotationConfigApplicationContext ctx =
new AnnotationConfigApplicationContext(configClass, ReactiveCacheableService.class);
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(
configClass, ReactiveCacheableService.class);
ReactiveCacheableService service = ctx.getBean(ReactiveCacheableService.class);
Object key = new Object();
@@ -207,6 +148,117 @@ class ReactiveCachingTests {
ctx.close();
}
@Test
void cacheErrorHandlerWithSimpleCacheErrorHandler() {
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(
ExceptionCacheManager.class, ReactiveCacheableService.class);
ReactiveCacheableService service = ctx.getBean(ReactiveCacheableService.class);
Throwable completableFutureThrowable = catchThrowable(() -> service.cacheFuture(new Object()).join());
assertThat(completableFutureThrowable).isInstanceOf(CompletionException.class)
.extracting(Throwable::getCause)
.isInstanceOf(UnsupportedOperationException.class);
Throwable monoThrowable = catchThrowable(() -> service.cacheMono(new Object()).block());
assertThat(monoThrowable).isInstanceOf(UnsupportedOperationException.class);
Throwable fluxThrowable = catchThrowable(() -> service.cacheFlux(new Object()).blockFirst());
assertThat(fluxThrowable).isInstanceOf(UnsupportedOperationException.class);
}
@Test
void cacheErrorHandlerWithSimpleCacheErrorHandlerAndSync() {
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(
ExceptionCacheManager.class, ReactiveSyncCacheableService.class);
ReactiveSyncCacheableService service = ctx.getBean(ReactiveSyncCacheableService.class);
Throwable completableFutureThrowable = catchThrowable(() -> service.cacheFuture(new Object()).join());
assertThat(completableFutureThrowable).isInstanceOf(CompletionException.class)
.extracting(Throwable::getCause)
.isInstanceOf(UnsupportedOperationException.class);
Throwable monoThrowable = catchThrowable(() -> service.cacheMono(new Object()).block());
assertThat(monoThrowable).isInstanceOf(UnsupportedOperationException.class);
Throwable fluxThrowable = catchThrowable(() -> service.cacheFlux(new Object()).blockFirst());
assertThat(fluxThrowable).isInstanceOf(UnsupportedOperationException.class);
}
@Test
void cacheErrorHandlerWithLoggingCacheErrorHandler() {
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(
ExceptionCacheManager.class, ReactiveCacheableService.class, ErrorHandlerCachingConfiguration.class);
ReactiveCacheableService service = ctx.getBean(ReactiveCacheableService.class);
Long r1 = service.cacheFuture(new Object()).join();
assertThat(r1).isNotNull();
assertThat(r1).as("cacheFuture").isEqualTo(0L);
r1 = service.cacheMono(new Object()).block();
assertThat(r1).isNotNull();
assertThat(r1).as("cacheMono").isEqualTo(1L);
r1 = service.cacheFlux(new Object()).blockFirst();
assertThat(r1).isNotNull();
assertThat(r1).as("cacheFlux blockFirst").isEqualTo(2L);
}
@Test
void cacheErrorHandlerWithLoggingCacheErrorHandlerAndSync() {
AnnotationConfigApplicationContext ctx =
new AnnotationConfigApplicationContext(ExceptionCacheManager.class, ReactiveSyncCacheableService.class, ErrorHandlerCachingConfiguration.class);
ReactiveSyncCacheableService service = ctx.getBean(ReactiveSyncCacheableService.class);
Long r1 = service.cacheFuture(new Object()).join();
assertThat(r1).isNotNull();
assertThat(r1).as("cacheFuture").isEqualTo(0L);
r1 = service.cacheMono(new Object()).block();
assertThat(r1).isNotNull();
assertThat(r1).as("cacheMono").isEqualTo(1L);
r1 = service.cacheFlux(new Object()).blockFirst();
assertThat(r1).isNotNull();
assertThat(r1).as("cacheFlux blockFirst").isEqualTo(2L);
}
@Test
void cacheErrorHandlerWithLoggingCacheErrorHandlerAndOperationException() {
AnnotationConfigApplicationContext ctx =
new AnnotationConfigApplicationContext(EarlyCacheHitDeterminationConfig.class, ReactiveFailureCacheableService.class, ErrorHandlerCachingConfiguration.class);
ReactiveFailureCacheableService service = ctx.getBean(ReactiveFailureCacheableService.class);
assertThatExceptionOfType(CompletionException.class).isThrownBy(() -> service.cacheFuture(new Object()).join())
.withMessage(IllegalStateException.class.getName() + ": future service error");
StepVerifier.create(service.cacheMono(new Object()))
.expectErrorMessage("mono service error")
.verify();
StepVerifier.create(service.cacheFlux(new Object()))
.expectErrorMessage("flux service error")
.verify();
}
@Test
void cacheErrorHandlerWithLoggingCacheErrorHandlerAndOperationExceptionAndSync() {
AnnotationConfigApplicationContext ctx =
new AnnotationConfigApplicationContext(EarlyCacheHitDeterminationConfig.class, ReactiveSyncFailureCacheableService.class, ErrorHandlerCachingConfiguration.class);
ReactiveSyncFailureCacheableService service = ctx.getBean(ReactiveSyncFailureCacheableService.class);
assertThatExceptionOfType(CompletionException.class).isThrownBy(() -> service.cacheFuture(new Object()).join())
.withMessage(IllegalStateException.class.getName() + ": future service error");
StepVerifier.create(service.cacheMono(new Object()))
.expectErrorMessage("mono service error")
.verify();
StepVerifier.create(service.cacheFlux(new Object()))
.expectErrorMessage("flux service error")
.verify();
}
@CacheConfig(cacheNames = "first")
static class ReactiveCacheableService {
@@ -232,16 +284,94 @@ class ReactiveCachingTests {
}
}
@CacheConfig(cacheNames = "first")
static class ReactiveFailureCacheableService extends ReactiveCacheableService {
static class ReactiveSyncCacheableService {
private final AtomicLong counter = new AtomicLong();
@Cacheable(sync = true)
CompletableFuture<Long> cacheFuture(Object arg) {
return CompletableFuture.completedFuture(this.counter.getAndIncrement());
}
@Cacheable(sync = true)
Mono<Long> cacheMono(Object arg) {
return Mono.defer(() -> Mono.just(this.counter.getAndIncrement()));
}
@Cacheable(sync = true)
Flux<Long> cacheFlux(Object arg) {
return Flux.defer(() -> Flux.just(this.counter.getAndIncrement(), 0L, -1L, -2L, -3L));
}
}
@CacheConfig(cacheNames = "first")
static class ReactiveFailureCacheableService {
private final AtomicBoolean cacheFutureInvoked = new AtomicBoolean();
private final AtomicBoolean cacheMonoInvoked = new AtomicBoolean();
private final AtomicBoolean cacheFluxInvoked = new AtomicBoolean();
@Cacheable
CompletableFuture<Long> cacheFuture(Object arg) {
if (!this.cacheFutureInvoked.compareAndSet(false, true)) {
return CompletableFuture.failedFuture(new IllegalStateException("future service invoked twice"));
}
return CompletableFuture.failedFuture(new IllegalStateException("future service error"));
}
@Cacheable
Mono<Long> cacheMono(Object arg) {
if (!this.cacheMonoInvoked.compareAndSet(false, true)) {
return Mono.error(new IllegalStateException("mono service invoked twice"));
}
return Mono.error(new IllegalStateException("mono service error"));
}
@Cacheable
Flux<Long> cacheFlux(Object arg) {
if (!this.cacheFluxInvoked.compareAndSet(false, true)) {
return Flux.error(new IllegalStateException("flux service invoked twice"));
}
return Flux.error(new IllegalStateException("flux service error"));
}
}
@CacheConfig(cacheNames = "first")
static class ReactiveSyncFailureCacheableService {
private final AtomicBoolean cacheFutureInvoked = new AtomicBoolean();
private final AtomicBoolean cacheMonoInvoked = new AtomicBoolean();
private final AtomicBoolean cacheFluxInvoked = new AtomicBoolean();
@Cacheable(sync = true)
CompletableFuture<Long> cacheFuture(Object arg) {
if (!this.cacheFutureInvoked.compareAndSet(false, true)) {
return CompletableFuture.failedFuture(new IllegalStateException("future service invoked twice"));
}
return CompletableFuture.failedFuture(new IllegalStateException("future service error"));
}
@Cacheable(sync = true)
Mono<Long> cacheMono(Object arg) {
if (!this.cacheMonoInvoked.compareAndSet(false, true)) {
return Mono.error(new IllegalStateException("mono service invoked twice"));
}
return Mono.error(new IllegalStateException("mono service error"));
}
@Cacheable(sync = true)
Flux<Long> cacheFlux(Object arg) {
if (!this.cacheFluxInvoked.compareAndSet(false, true)) {
return Flux.error(new IllegalStateException("flux service invoked twice"));
}
return Flux.error(new IllegalStateException("flux service error"));
}
}
@@ -323,6 +453,7 @@ class ReactiveCachingTests {
}
}
@Configuration
static class ErrorHandlerCachingConfiguration implements CachingConfigurer {
@@ -333,6 +464,7 @@ class ReactiveCachingTests {
}
}
@Configuration(proxyBeanMethods = false)
@EnableCaching
static class ExceptionCacheManager {
@@ -345,11 +477,12 @@ class ReactiveCachingTests {
return new ConcurrentMapCache(name, isAllowNullValues()) {
@Override
public CompletableFuture<?> retrieve(Object key) {
return CompletableFuture.supplyAsync(() -> {
throw new UnsupportedOperationException("Test exception on retrieve");
});
return CompletableFuture.failedFuture(new UnsupportedOperationException("Test exception on retrieve"));
}
@Override
public <T> CompletableFuture<T> retrieve(Object key, Supplier<CompletableFuture<T>> valueLoader) {
return CompletableFuture.failedFuture(new UnsupportedOperationException("Test exception on retrieve"));
}
@Override
public void put(Object key, @Nullable Object value) {
throw new UnsupportedOperationException("Test exception on put");