mirror of
https://github.com/spring-projects/spring-framework
synced 2026-06-08 17:33:33 +00:00
Introduce @ConcurrencyLimit annotation based on ConcurrencyThrottleInterceptor
Moves @Retryable infrastructure to resilience package in spring-context module. Includes duration parsing and placeholder resolution for @Retryable attributes. Provides convenient @EnableResilientMethods for @Retryable + @ConcurrencyLimit. Closes gh-35133 See gh-34529
This commit is contained in:
Juergen Hoeller
+188
@@ -0,0 +1,188 @@
|
||||
/*
|
||||
* Copyright 2002-present 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.resilience;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.aop.framework.AopProxyUtils;
|
||||
import org.springframework.aop.framework.ProxyFactory;
|
||||
import org.springframework.aop.interceptor.ConcurrencyThrottleInterceptor;
|
||||
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
|
||||
import org.springframework.beans.factory.support.RootBeanDefinition;
|
||||
import org.springframework.resilience.annotation.ConcurrencyLimit;
|
||||
import org.springframework.resilience.annotation.ConcurrencyLimitBeanPostProcessor;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* @author Juergen Hoeller
|
||||
* @since 7.0
|
||||
*/
|
||||
public class ConcurrencyLimitTests {
|
||||
|
||||
@Test
|
||||
void withSimpleInterceptor() {
|
||||
NonAnnotatedBean target = new NonAnnotatedBean();
|
||||
ProxyFactory pf = new ProxyFactory();
|
||||
pf.setTarget(target);
|
||||
pf.addAdvice(new ConcurrencyThrottleInterceptor(2));
|
||||
NonAnnotatedBean proxy = (NonAnnotatedBean) pf.getProxy();
|
||||
|
||||
List<CompletableFuture<?>> futures = new ArrayList<>(10);
|
||||
for (int i = 0; i < 10; i++) {
|
||||
futures.add(CompletableFuture.runAsync(proxy::concurrentOperation));
|
||||
}
|
||||
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
|
||||
assertThat(target.counter).hasValue(0);
|
||||
}
|
||||
|
||||
@Test
|
||||
void withPostProcessorForMethod() {
|
||||
DefaultListableBeanFactory bf = new DefaultListableBeanFactory();
|
||||
bf.registerBeanDefinition("bean", new RootBeanDefinition(AnnotatedMethodBean.class));
|
||||
ConcurrencyLimitBeanPostProcessor bpp = new ConcurrencyLimitBeanPostProcessor();
|
||||
bpp.setBeanFactory(bf);
|
||||
bf.addBeanPostProcessor(bpp);
|
||||
AnnotatedMethodBean proxy = bf.getBean(AnnotatedMethodBean.class);
|
||||
AnnotatedMethodBean target = (AnnotatedMethodBean) AopProxyUtils.getSingletonTarget(proxy);
|
||||
|
||||
List<CompletableFuture<?>> futures = new ArrayList<>(10);
|
||||
for (int i = 0; i < 10; i++) {
|
||||
futures.add(CompletableFuture.runAsync(proxy::concurrentOperation));
|
||||
}
|
||||
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
|
||||
assertThat(target.current).hasValue(0);
|
||||
}
|
||||
|
||||
@Test
|
||||
void withPostProcessorForClass() {
|
||||
DefaultListableBeanFactory bf = new DefaultListableBeanFactory();
|
||||
bf.registerBeanDefinition("bean", new RootBeanDefinition(AnnotatedClassBean.class));
|
||||
ConcurrencyLimitBeanPostProcessor bpp = new ConcurrencyLimitBeanPostProcessor();
|
||||
bpp.setBeanFactory(bf);
|
||||
bf.addBeanPostProcessor(bpp);
|
||||
AnnotatedClassBean proxy = bf.getBean(AnnotatedClassBean.class);
|
||||
AnnotatedClassBean target = (AnnotatedClassBean) AopProxyUtils.getSingletonTarget(proxy);
|
||||
|
||||
List<CompletableFuture<?>> futures = new ArrayList<>(30);
|
||||
for (int i = 0; i < 10; i++) {
|
||||
futures.add(CompletableFuture.runAsync(proxy::concurrentOperation));
|
||||
}
|
||||
for (int i = 0; i < 10; i++) {
|
||||
futures.add(CompletableFuture.runAsync(proxy::otherOperation));
|
||||
}
|
||||
for (int i = 0; i < 10; i++) {
|
||||
futures.add(CompletableFuture.runAsync(proxy::overrideOperation));
|
||||
}
|
||||
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
|
||||
assertThat(target.current).hasValue(0);
|
||||
}
|
||||
|
||||
|
||||
public static class NonAnnotatedBean {
|
||||
|
||||
AtomicInteger counter = new AtomicInteger();
|
||||
|
||||
public void concurrentOperation() {
|
||||
if (counter.incrementAndGet() > 2) {
|
||||
throw new IllegalStateException();
|
||||
}
|
||||
try {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
catch (InterruptedException ex) {
|
||||
throw new IllegalStateException(ex);
|
||||
}
|
||||
counter.decrementAndGet();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public static class AnnotatedMethodBean {
|
||||
|
||||
AtomicInteger current = new AtomicInteger();
|
||||
|
||||
@ConcurrencyLimit(2)
|
||||
public void concurrentOperation() {
|
||||
if (current.incrementAndGet() > 2) {
|
||||
throw new IllegalStateException();
|
||||
}
|
||||
try {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
catch (InterruptedException ex) {
|
||||
throw new IllegalStateException(ex);
|
||||
}
|
||||
current.decrementAndGet();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ConcurrencyLimit(2)
|
||||
public static class AnnotatedClassBean {
|
||||
|
||||
AtomicInteger current = new AtomicInteger();
|
||||
|
||||
AtomicInteger currentOverride = new AtomicInteger();
|
||||
|
||||
public void concurrentOperation() {
|
||||
if (current.incrementAndGet() > 2) {
|
||||
throw new IllegalStateException();
|
||||
}
|
||||
try {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
catch (InterruptedException ex) {
|
||||
throw new IllegalStateException(ex);
|
||||
}
|
||||
current.decrementAndGet();
|
||||
}
|
||||
|
||||
public void otherOperation() {
|
||||
if (current.incrementAndGet() > 2) {
|
||||
throw new IllegalStateException();
|
||||
}
|
||||
try {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
catch (InterruptedException ex) {
|
||||
throw new IllegalStateException(ex);
|
||||
}
|
||||
current.decrementAndGet();
|
||||
}
|
||||
|
||||
@ConcurrencyLimit(1)
|
||||
public void overrideOperation() {
|
||||
if (currentOverride.incrementAndGet() > 1) {
|
||||
throw new IllegalStateException();
|
||||
}
|
||||
try {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
catch (InterruptedException ex) {
|
||||
throw new IllegalStateException(ex);
|
||||
}
|
||||
currentOverride.decrementAndGet();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
+338
@@ -0,0 +1,338 @@
|
||||
/*
|
||||
* Copyright 2002-present 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.resilience;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.lang.reflect.Method;
|
||||
import java.nio.file.AccessDeniedException;
|
||||
import java.time.Duration;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import org.springframework.aop.framework.AopProxyUtils;
|
||||
import org.springframework.aop.framework.ProxyFactory;
|
||||
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
|
||||
import org.springframework.beans.factory.support.RootBeanDefinition;
|
||||
import org.springframework.resilience.annotation.RetryAnnotationBeanPostProcessor;
|
||||
import org.springframework.resilience.annotation.Retryable;
|
||||
import org.springframework.resilience.retry.MethodRetryPredicate;
|
||||
import org.springframework.resilience.retry.MethodRetrySpec;
|
||||
import org.springframework.resilience.retry.SimpleRetryInterceptor;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
|
||||
import static org.assertj.core.api.Assertions.assertThatRuntimeException;
|
||||
|
||||
/**
|
||||
* @author Juergen Hoeller
|
||||
* @since 7.0
|
||||
*/
|
||||
public class ReactiveRetryInterceptorTests {
|
||||
|
||||
@Test
|
||||
void withSimpleInterceptor() {
|
||||
NonAnnotatedBean target = new NonAnnotatedBean();
|
||||
ProxyFactory pf = new ProxyFactory();
|
||||
pf.setTarget(target);
|
||||
pf.addAdvice(new SimpleRetryInterceptor(
|
||||
new MethodRetrySpec((m, t) -> true, 5, Duration.ofMillis(10))));
|
||||
NonAnnotatedBean proxy = (NonAnnotatedBean) pf.getProxy();
|
||||
|
||||
assertThatIllegalStateException().isThrownBy(() -> proxy.retryOperation().block())
|
||||
.withCauseInstanceOf(IOException.class).havingCause().withMessage("6");
|
||||
assertThat(target.counter.get()).isEqualTo(6);
|
||||
}
|
||||
|
||||
@Test
|
||||
void withPostProcessorForMethod() {
|
||||
DefaultListableBeanFactory bf = new DefaultListableBeanFactory();
|
||||
bf.registerBeanDefinition("bean", new RootBeanDefinition(AnnotatedMethodBean.class));
|
||||
RetryAnnotationBeanPostProcessor bpp = new RetryAnnotationBeanPostProcessor();
|
||||
bpp.setBeanFactory(bf);
|
||||
bf.addBeanPostProcessor(bpp);
|
||||
AnnotatedMethodBean proxy = bf.getBean(AnnotatedMethodBean.class);
|
||||
AnnotatedMethodBean target = (AnnotatedMethodBean) AopProxyUtils.getSingletonTarget(proxy);
|
||||
|
||||
assertThatIllegalStateException().isThrownBy(() -> proxy.retryOperation().block())
|
||||
.withCauseInstanceOf(IOException.class).havingCause().withMessage("6");
|
||||
assertThat(target.counter.get()).isEqualTo(6);
|
||||
}
|
||||
|
||||
@Test
|
||||
void withPostProcessorForClass() {
|
||||
DefaultListableBeanFactory bf = new DefaultListableBeanFactory();
|
||||
bf.registerBeanDefinition("bean", new RootBeanDefinition(AnnotatedClassBean.class));
|
||||
RetryAnnotationBeanPostProcessor bpp = new RetryAnnotationBeanPostProcessor();
|
||||
bpp.setBeanFactory(bf);
|
||||
bf.addBeanPostProcessor(bpp);
|
||||
AnnotatedClassBean proxy = bf.getBean(AnnotatedClassBean.class);
|
||||
AnnotatedClassBean target = (AnnotatedClassBean) AopProxyUtils.getSingletonTarget(proxy);
|
||||
|
||||
assertThatRuntimeException().isThrownBy(() -> proxy.retryOperation().block())
|
||||
.withCauseInstanceOf(IOException.class).havingCause().withMessage("3");
|
||||
assertThat(target.counter.get()).isEqualTo(3);
|
||||
assertThatRuntimeException().isThrownBy(() -> proxy.otherOperation().block())
|
||||
.withCauseInstanceOf(IOException.class);
|
||||
assertThat(target.counter.get()).isEqualTo(4);
|
||||
assertThatIllegalStateException().isThrownBy(() -> proxy.overrideOperation().blockFirst())
|
||||
.withCauseInstanceOf(IOException.class);
|
||||
assertThat(target.counter.get()).isEqualTo(6);
|
||||
}
|
||||
|
||||
@Test
|
||||
void adaptReactiveResultWithMinimalRetrySpec() {
|
||||
// Test minimal retry configuration: maxAttempts=1, delay=0, jitter=0, multiplier=1.0, maxDelay=0
|
||||
MinimalRetryBean target = new MinimalRetryBean();
|
||||
ProxyFactory pf = new ProxyFactory();
|
||||
pf.setTarget(target);
|
||||
pf.addAdvice(new SimpleRetryInterceptor(
|
||||
new MethodRetrySpec((m, t) -> true, 1, Duration.ZERO, Duration.ZERO, 1.0, Duration.ZERO)));
|
||||
MinimalRetryBean proxy = (MinimalRetryBean) pf.getProxy();
|
||||
|
||||
// Should execute only 2 times, because maxAttempts=1 means 1 call + 1 retry
|
||||
assertThatIllegalStateException().isThrownBy(() -> proxy.retryOperation().block())
|
||||
.withCauseInstanceOf(IOException.class).havingCause().withMessage("2");
|
||||
assertThat(target.counter.get()).isEqualTo(2);
|
||||
}
|
||||
|
||||
@Test
|
||||
void adaptReactiveResultWithZeroDelayAndJitter() {
|
||||
// Test case where delay=0 and jitter>0
|
||||
ZeroDelayJitterBean target = new ZeroDelayJitterBean();
|
||||
ProxyFactory pf = new ProxyFactory();
|
||||
pf.setTarget(target);
|
||||
pf.addAdvice(new SimpleRetryInterceptor(
|
||||
new MethodRetrySpec((m, t) -> true, 3, Duration.ZERO, Duration.ofMillis(10), 2.0, Duration.ofMillis(100))));
|
||||
ZeroDelayJitterBean proxy = (ZeroDelayJitterBean) pf.getProxy();
|
||||
|
||||
assertThatIllegalStateException().isThrownBy(() -> proxy.retryOperation().block())
|
||||
.withCauseInstanceOf(IOException.class).havingCause().withMessage("4");
|
||||
assertThat(target.counter.get()).isEqualTo(4);
|
||||
}
|
||||
|
||||
@Test
|
||||
void adaptReactiveResultWithJitterGreaterThanDelay() {
|
||||
// Test case where jitter > delay
|
||||
JitterGreaterThanDelayBean target = new JitterGreaterThanDelayBean();
|
||||
ProxyFactory pf = new ProxyFactory();
|
||||
pf.setTarget(target);
|
||||
pf.addAdvice(new SimpleRetryInterceptor(
|
||||
new MethodRetrySpec((m, t) -> true, 3, Duration.ofMillis(5), Duration.ofMillis(20), 1.5, Duration.ofMillis(50))));
|
||||
JitterGreaterThanDelayBean proxy = (JitterGreaterThanDelayBean) pf.getProxy();
|
||||
|
||||
assertThatIllegalStateException().isThrownBy(() -> proxy.retryOperation().block())
|
||||
.withCauseInstanceOf(IOException.class).havingCause().withMessage("4");
|
||||
assertThat(target.counter.get()).isEqualTo(4);
|
||||
}
|
||||
|
||||
@Test
|
||||
void adaptReactiveResultWithFluxMultiValue() {
|
||||
// Test Flux multi-value stream case
|
||||
FluxMultiValueBean target = new FluxMultiValueBean();
|
||||
ProxyFactory pf = new ProxyFactory();
|
||||
pf.setTarget(target);
|
||||
pf.addAdvice(new SimpleRetryInterceptor(
|
||||
new MethodRetrySpec((m, t) -> true, 3, Duration.ofMillis(10), Duration.ofMillis(5), 2.0, Duration.ofMillis(100))));
|
||||
FluxMultiValueBean proxy = (FluxMultiValueBean) pf.getProxy();
|
||||
|
||||
assertThatIllegalStateException().isThrownBy(() -> proxy.retryOperation().blockFirst())
|
||||
.withCauseInstanceOf(IOException.class).havingCause().withMessage("4");
|
||||
assertThat(target.counter.get()).isEqualTo(4);
|
||||
}
|
||||
|
||||
@Test
|
||||
void adaptReactiveResultWithSuccessfulOperation() {
|
||||
// Test successful return case, ensuring retry mechanism doesn't activate
|
||||
SuccessfulOperationBean target = new SuccessfulOperationBean();
|
||||
ProxyFactory pf = new ProxyFactory();
|
||||
pf.setTarget(target);
|
||||
pf.addAdvice(new SimpleRetryInterceptor(
|
||||
new MethodRetrySpec((m, t) -> true, 5, Duration.ofMillis(10), Duration.ofMillis(5), 2.0, Duration.ofMillis(100))));
|
||||
SuccessfulOperationBean proxy = (SuccessfulOperationBean) pf.getProxy();
|
||||
|
||||
String result = proxy.retryOperation().block();
|
||||
assertThat(result).isEqualTo("success");
|
||||
// Should execute only once because of successful return
|
||||
assertThat(target.counter.get()).isEqualTo(1);
|
||||
}
|
||||
|
||||
@Test
|
||||
void adaptReactiveResultWithImmediateFailure() {
|
||||
// Test immediate failure case
|
||||
ImmediateFailureBean target = new ImmediateFailureBean();
|
||||
ProxyFactory pf = new ProxyFactory();
|
||||
pf.setTarget(target);
|
||||
pf.addAdvice(new SimpleRetryInterceptor(
|
||||
new MethodRetrySpec((m, t) -> true, 3, Duration.ofMillis(10), Duration.ofMillis(5), 1.5, Duration.ofMillis(50))));
|
||||
ImmediateFailureBean proxy = (ImmediateFailureBean) pf.getProxy();
|
||||
|
||||
assertThatIllegalStateException().isThrownBy(() -> proxy.retryOperation().block())
|
||||
.withCauseInstanceOf(RuntimeException.class).havingCause().withMessage("immediate failure");
|
||||
assertThat(target.counter.get()).isEqualTo(4);
|
||||
}
|
||||
|
||||
|
||||
public static class NonAnnotatedBean {
|
||||
|
||||
AtomicInteger counter = new AtomicInteger();
|
||||
|
||||
public Mono<Object> retryOperation() {
|
||||
return Mono.fromCallable(() -> {
|
||||
counter.incrementAndGet();
|
||||
throw new IOException(counter.toString());
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public static class AnnotatedMethodBean {
|
||||
|
||||
AtomicInteger counter = new AtomicInteger();
|
||||
|
||||
@Retryable(maxAttempts = 5, delay = 10)
|
||||
public Mono<Object> retryOperation() {
|
||||
return Mono.fromCallable(() -> {
|
||||
counter.incrementAndGet();
|
||||
throw new IOException(counter.toString());
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@Retryable(delay = 10, jitter = 5, multiplier = 2.0, maxDelay = 40,
|
||||
includes = IOException.class, excludes = AccessDeniedException.class,
|
||||
predicate = CustomPredicate.class)
|
||||
public static class AnnotatedClassBean {
|
||||
|
||||
AtomicInteger counter = new AtomicInteger();
|
||||
|
||||
public Mono<Object> retryOperation() {
|
||||
return Mono.fromCallable(() -> {
|
||||
counter.incrementAndGet();
|
||||
throw new IOException(counter.toString());
|
||||
});
|
||||
}
|
||||
|
||||
public Mono<Object> otherOperation() {
|
||||
return Mono.fromCallable(() -> {
|
||||
counter.incrementAndGet();
|
||||
throw new AccessDeniedException(counter.toString());
|
||||
});
|
||||
}
|
||||
|
||||
@Retryable(value = IOException.class, maxAttempts = 1, delay = 10)
|
||||
public Flux<Object> overrideOperation() {
|
||||
return Flux.from(Mono.fromCallable(() -> {
|
||||
counter.incrementAndGet();
|
||||
throw new AccessDeniedException(counter.toString());
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private static class CustomPredicate implements MethodRetryPredicate {
|
||||
|
||||
@Override
|
||||
public boolean shouldRetry(Method method, Throwable throwable) {
|
||||
return !"3".equals(throwable.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// Bean classes for boundary testing
|
||||
|
||||
public static class MinimalRetryBean {
|
||||
|
||||
AtomicInteger counter = new AtomicInteger();
|
||||
|
||||
public Mono<Object> retryOperation() {
|
||||
return Mono.fromCallable(() -> {
|
||||
counter.incrementAndGet();
|
||||
throw new IOException(counter.toString());
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public static class ZeroDelayJitterBean {
|
||||
|
||||
AtomicInteger counter = new AtomicInteger();
|
||||
|
||||
public Mono<Object> retryOperation() {
|
||||
return Mono.fromCallable(() -> {
|
||||
counter.incrementAndGet();
|
||||
throw new IOException(counter.toString());
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public static class JitterGreaterThanDelayBean {
|
||||
|
||||
AtomicInteger counter = new AtomicInteger();
|
||||
|
||||
public Mono<Object> retryOperation() {
|
||||
return Mono.fromCallable(() -> {
|
||||
counter.incrementAndGet();
|
||||
throw new IOException(counter.toString());
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public static class FluxMultiValueBean {
|
||||
|
||||
AtomicInteger counter = new AtomicInteger();
|
||||
|
||||
public Flux<Object> retryOperation() {
|
||||
return Flux.from(Mono.fromCallable(() -> {
|
||||
counter.incrementAndGet();
|
||||
throw new IOException(counter.toString());
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public static class SuccessfulOperationBean {
|
||||
|
||||
AtomicInteger counter = new AtomicInteger();
|
||||
|
||||
public Mono<String> retryOperation() {
|
||||
return Mono.fromCallable(() -> {
|
||||
counter.incrementAndGet();
|
||||
return "success";
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public static class ImmediateFailureBean {
|
||||
|
||||
AtomicInteger counter = new AtomicInteger();
|
||||
|
||||
public Mono<Object> retryOperation() {
|
||||
return Mono.fromCallable(() -> {
|
||||
counter.incrementAndGet();
|
||||
throw new RuntimeException("immediate failure");
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
+254
@@ -0,0 +1,254 @@
|
||||
/*
|
||||
* Copyright 2002-present 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.resilience;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.lang.reflect.Method;
|
||||
import java.nio.file.AccessDeniedException;
|
||||
import java.time.Duration;
|
||||
import java.util.Properties;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.aop.framework.AopProxyUtils;
|
||||
import org.springframework.aop.framework.ProxyFactory;
|
||||
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
|
||||
import org.springframework.beans.factory.support.RootBeanDefinition;
|
||||
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
|
||||
import org.springframework.context.support.GenericApplicationContext;
|
||||
import org.springframework.core.env.PropertiesPropertySource;
|
||||
import org.springframework.resilience.annotation.ConcurrencyLimit;
|
||||
import org.springframework.resilience.annotation.EnableResilientMethods;
|
||||
import org.springframework.resilience.annotation.RetryAnnotationBeanPostProcessor;
|
||||
import org.springframework.resilience.annotation.Retryable;
|
||||
import org.springframework.resilience.retry.MethodRetryPredicate;
|
||||
import org.springframework.resilience.retry.MethodRetrySpec;
|
||||
import org.springframework.resilience.retry.SimpleRetryInterceptor;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatIOException;
|
||||
|
||||
/**
|
||||
* @author Juergen Hoeller
|
||||
* @since 7.0
|
||||
*/
|
||||
public class RetryInterceptorTests {
|
||||
|
||||
@Test
|
||||
void withSimpleInterceptor() {
|
||||
NonAnnotatedBean target = new NonAnnotatedBean();
|
||||
ProxyFactory pf = new ProxyFactory();
|
||||
pf.setTarget(target);
|
||||
pf.addAdvice(new SimpleRetryInterceptor(
|
||||
new MethodRetrySpec((m, t) -> true, 5, Duration.ofMillis(10))));
|
||||
NonAnnotatedBean proxy = (NonAnnotatedBean) pf.getProxy();
|
||||
|
||||
assertThatIOException().isThrownBy(proxy::retryOperation).withMessage("6");
|
||||
assertThat(target.counter).isEqualTo(6);
|
||||
}
|
||||
|
||||
@Test
|
||||
void withPostProcessorForMethod() {
|
||||
DefaultListableBeanFactory bf = new DefaultListableBeanFactory();
|
||||
bf.registerBeanDefinition("bean", new RootBeanDefinition(AnnotatedMethodBean.class));
|
||||
RetryAnnotationBeanPostProcessor bpp = new RetryAnnotationBeanPostProcessor();
|
||||
bpp.setBeanFactory(bf);
|
||||
bf.addBeanPostProcessor(bpp);
|
||||
AnnotatedMethodBean proxy = bf.getBean(AnnotatedMethodBean.class);
|
||||
AnnotatedMethodBean target = (AnnotatedMethodBean) AopProxyUtils.getSingletonTarget(proxy);
|
||||
|
||||
assertThatIOException().isThrownBy(proxy::retryOperation).withMessage("6");
|
||||
assertThat(target.counter).isEqualTo(6);
|
||||
}
|
||||
|
||||
@Test
|
||||
void withPostProcessorForClass() {
|
||||
DefaultListableBeanFactory bf = new DefaultListableBeanFactory();
|
||||
bf.registerBeanDefinition("bean", new RootBeanDefinition(AnnotatedClassBean.class));
|
||||
RetryAnnotationBeanPostProcessor bpp = new RetryAnnotationBeanPostProcessor();
|
||||
bpp.setBeanFactory(bf);
|
||||
bf.addBeanPostProcessor(bpp);
|
||||
AnnotatedClassBean proxy = bf.getBean(AnnotatedClassBean.class);
|
||||
AnnotatedClassBean target = (AnnotatedClassBean) AopProxyUtils.getSingletonTarget(proxy);
|
||||
|
||||
assertThatIOException().isThrownBy(proxy::retryOperation).withMessage("3");
|
||||
assertThat(target.counter).isEqualTo(3);
|
||||
assertThatIOException().isThrownBy(proxy::otherOperation);
|
||||
assertThat(target.counter).isEqualTo(4);
|
||||
assertThatIOException().isThrownBy(proxy::overrideOperation);
|
||||
assertThat(target.counter).isEqualTo(6);
|
||||
}
|
||||
|
||||
@Test
|
||||
void withPostProcessorForClassWithStrings() {
|
||||
Properties props = new Properties();
|
||||
props.setProperty("delay", "10");
|
||||
props.setProperty("jitter", "5");
|
||||
props.setProperty("multiplier", "2.0");
|
||||
props.setProperty("maxDelay", "40");
|
||||
props.setProperty("limitedAttempts", "1");
|
||||
|
||||
GenericApplicationContext ctx = new GenericApplicationContext();
|
||||
ctx.getEnvironment().getPropertySources().addFirst(new PropertiesPropertySource("props", props));
|
||||
ctx.registerBeanDefinition("bean", new RootBeanDefinition(AnnotatedClassBeanWithStrings.class));
|
||||
ctx.registerBeanDefinition("bpp", new RootBeanDefinition(RetryAnnotationBeanPostProcessor.class));
|
||||
ctx.refresh();
|
||||
AnnotatedClassBeanWithStrings proxy = ctx.getBean(AnnotatedClassBeanWithStrings.class);
|
||||
AnnotatedClassBeanWithStrings target = (AnnotatedClassBeanWithStrings) AopProxyUtils.getSingletonTarget(proxy);
|
||||
|
||||
assertThatIOException().isThrownBy(proxy::retryOperation).withMessage("3");
|
||||
assertThat(target.counter).isEqualTo(3);
|
||||
assertThatIOException().isThrownBy(proxy::otherOperation);
|
||||
assertThat(target.counter).isEqualTo(4);
|
||||
assertThatIOException().isThrownBy(proxy::overrideOperation);
|
||||
assertThat(target.counter).isEqualTo(6);
|
||||
}
|
||||
|
||||
@Test
|
||||
void withEnableAnnotation() throws Exception {
|
||||
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext();
|
||||
ctx.registerBeanDefinition("bean", new RootBeanDefinition(DoubleAnnotatedBean.class));
|
||||
ctx.registerBeanDefinition("config", new RootBeanDefinition(EnablingConfig.class));
|
||||
ctx.refresh();
|
||||
DoubleAnnotatedBean proxy = ctx.getBean(DoubleAnnotatedBean.class);
|
||||
DoubleAnnotatedBean target = (DoubleAnnotatedBean) AopProxyUtils.getSingletonTarget(proxy);
|
||||
|
||||
Thread thread = new Thread(() -> assertThatIOException().isThrownBy(proxy::retryOperation));
|
||||
thread.start();
|
||||
assertThatIOException().isThrownBy(proxy::retryOperation);
|
||||
thread.join();
|
||||
assertThat(target.counter).hasValue(6);
|
||||
assertThat(target.threadChange).hasValue(2);
|
||||
}
|
||||
|
||||
|
||||
public static class NonAnnotatedBean {
|
||||
|
||||
int counter = 0;
|
||||
|
||||
public void retryOperation() throws IOException {
|
||||
counter++;
|
||||
throw new IOException(Integer.toString(counter));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public static class AnnotatedMethodBean {
|
||||
|
||||
int counter = 0;
|
||||
|
||||
@Retryable(maxAttempts = 5, delay = 10)
|
||||
public void retryOperation() throws IOException {
|
||||
counter++;
|
||||
throw new IOException(Integer.toString(counter));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@Retryable(delay = 10, jitter = 5, multiplier = 2.0, maxDelay = 40,
|
||||
includes = IOException.class, excludes = AccessDeniedException.class,
|
||||
predicate = CustomPredicate.class)
|
||||
public static class AnnotatedClassBean {
|
||||
|
||||
int counter = 0;
|
||||
|
||||
public void retryOperation() throws IOException {
|
||||
counter++;
|
||||
throw new IOException(Integer.toString(counter));
|
||||
}
|
||||
|
||||
public void otherOperation() throws IOException {
|
||||
counter++;
|
||||
throw new AccessDeniedException(Integer.toString(counter));
|
||||
}
|
||||
|
||||
@Retryable(value = IOException.class, maxAttempts = 1, delay = 10)
|
||||
public void overrideOperation() throws IOException {
|
||||
counter++;
|
||||
throw new AccessDeniedException(Integer.toString(counter));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@Retryable(delayString = "${delay}", jitterString = "${jitter}",
|
||||
multiplierString = "${multiplier}", maxDelayString = "${maxDelay}",
|
||||
includes = IOException.class, excludes = AccessDeniedException.class,
|
||||
predicate = CustomPredicate.class)
|
||||
public static class AnnotatedClassBeanWithStrings {
|
||||
|
||||
int counter = 0;
|
||||
|
||||
public void retryOperation() throws IOException {
|
||||
counter++;
|
||||
throw new IOException(Integer.toString(counter));
|
||||
}
|
||||
|
||||
public void otherOperation() throws IOException {
|
||||
counter++;
|
||||
throw new AccessDeniedException(Integer.toString(counter));
|
||||
}
|
||||
|
||||
@Retryable(value = IOException.class, maxAttemptsString = "${limitedAttempts}", delayString = "10ms")
|
||||
public void overrideOperation() throws IOException {
|
||||
counter++;
|
||||
throw new AccessDeniedException(Integer.toString(counter));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private static class CustomPredicate implements MethodRetryPredicate {
|
||||
|
||||
@Override
|
||||
public boolean shouldRetry(Method method, Throwable throwable) {
|
||||
return !"3".equals(throwable.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public static class DoubleAnnotatedBean {
|
||||
|
||||
AtomicInteger current = new AtomicInteger();
|
||||
|
||||
AtomicInteger counter = new AtomicInteger();
|
||||
|
||||
AtomicInteger threadChange = new AtomicInteger();
|
||||
|
||||
volatile String lastThreadName;
|
||||
|
||||
@ConcurrencyLimit(1)
|
||||
@Retryable(maxAttempts = 2, delay = 10)
|
||||
public void retryOperation() throws IOException, InterruptedException {
|
||||
if (current.incrementAndGet() > 1) {
|
||||
throw new IllegalStateException();
|
||||
}
|
||||
Thread.sleep(100);
|
||||
current.decrementAndGet();
|
||||
if (!Thread.currentThread().getName().equals(lastThreadName)) {
|
||||
lastThreadName = Thread.currentThread().getName();
|
||||
threadChange.incrementAndGet();
|
||||
}
|
||||
throw new IOException(Integer.toString(counter.incrementAndGet()));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@EnableResilientMethods
|
||||
public static class EnablingConfig {
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user