Add resilience: Spring Framework 7 @Retryable and @ConcurrencyLimit
Companion code for "Spring Framework 7's Built-in Resilience: @Retryable, @ConcurrencyLimit, and What's Left for Resilience4j". Every retry counted by recording real invocations: defaults, backoff and jitter, timeout, reactive and CompletableFuture returns, the concurrency limit's BLOCK and REJECT policies, retries around transactions, composition with Resilience4j 2.4.0, and the annotation API across 7.0.0-7.0.9. Co-Authored-By: Claude Opus 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01C3TETMrqVUWeFkNtz3Jbo3
This commit is contained in:
Claude Opus 5
@@ -0,0 +1,12 @@
|
||||
package com.ankurm.resilience;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
@SpringBootApplication
|
||||
public class ResilienceApplication {
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(ResilienceApplication.class, args);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
package com.ankurm.resilience;
|
||||
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.resilience.annotation.EnableResilientMethods;
|
||||
|
||||
/**
|
||||
* Spring Boot 4.1 does not switch the resilience annotations on for you. Without this,
|
||||
* {@code @Retryable} and {@code @ConcurrencyLimit} are inert - no proxy, no error, no log line.
|
||||
* The {@code demo.resilience.enabled=false} switch exists only so the article can show that
|
||||
* (docs/02-enabling.md).
|
||||
*/
|
||||
@Configuration
|
||||
@EnableResilientMethods
|
||||
@ConditionalOnBooleanProperty(name = "demo.resilience.enabled", matchIfMissing = true)
|
||||
public class ResilienceConfig {
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
package com.ankurm.resilience.limit;
|
||||
|
||||
import com.ankurm.resilience.support.CallLog;
|
||||
|
||||
import org.springframework.resilience.annotation.ConcurrencyLimit;
|
||||
import org.springframework.resilience.annotation.Retryable;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/**
|
||||
* An expensive operation that must not run more than twice at once. The default policy BLOCKS
|
||||
* callers beyond the limit; REJECT (added in 7.0.3) throws
|
||||
* {@code InvocationRejectedException} instead. See docs/04-concurrency-limit.md.
|
||||
*/
|
||||
@Service
|
||||
public class ReportService {
|
||||
|
||||
private final CallLog log;
|
||||
|
||||
public ReportService(CallLog log) {
|
||||
this.log = log;
|
||||
}
|
||||
|
||||
@ConcurrencyLimit(2)
|
||||
public String blocking() {
|
||||
return work();
|
||||
}
|
||||
|
||||
@ConcurrencyLimit(limit = 2, policy = ConcurrencyLimit.ThrottlePolicy.REJECT)
|
||||
public String rejecting() {
|
||||
return work();
|
||||
}
|
||||
|
||||
private final ThreadLocal<Integer> attemptsOnThisThread = ThreadLocal.withInitial(() -> 0);
|
||||
|
||||
/**
|
||||
* Both annotations on one method. Every caller's first attempt fails and is retried after
|
||||
* 300 ms. Whether the permit is held during that 300 ms depends on which interceptor is
|
||||
* outermost - measured in docs/output/limit-and-retry.txt.
|
||||
*/
|
||||
@ConcurrencyLimit(1)
|
||||
@Retryable(maxRetries = 1, delay = 300)
|
||||
public String limitedAndRetried() {
|
||||
int attempt = attemptsOnThisThread.get() + 1;
|
||||
attemptsOnThisThread.set(attempt);
|
||||
log.enter();
|
||||
log.record();
|
||||
try {
|
||||
Thread.sleep(50);
|
||||
if (attempt == 1) {
|
||||
throw new IllegalStateException("first attempt fails");
|
||||
}
|
||||
attemptsOnThisThread.remove();
|
||||
return "ok";
|
||||
}
|
||||
catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
return "interrupted";
|
||||
}
|
||||
finally {
|
||||
log.exit();
|
||||
}
|
||||
}
|
||||
|
||||
public String unlimited() {
|
||||
return work();
|
||||
}
|
||||
|
||||
private String work() {
|
||||
log.enter();
|
||||
log.record();
|
||||
try {
|
||||
Thread.sleep(200);
|
||||
}
|
||||
catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
finally {
|
||||
log.exit();
|
||||
}
|
||||
return "rendered";
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,95 @@
|
||||
package com.ankurm.resilience.r4j;
|
||||
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
import com.ankurm.resilience.support.CallLog;
|
||||
import com.ankurm.resilience.support.TransientException;
|
||||
import io.github.resilience4j.bulkhead.annotation.Bulkhead;
|
||||
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
|
||||
import io.github.resilience4j.retry.annotation.Retry;
|
||||
import io.github.resilience4j.timelimiter.annotation.TimeLimiter;
|
||||
|
||||
import org.springframework.resilience.annotation.Retryable;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/**
|
||||
* The same kind of downstream, guarded by Resilience4j 2.4.0 instead - the things Spring Framework
|
||||
* 7 does not do (circuit breaking, time limiting, a rejecting bulkhead with metrics), and one
|
||||
* method that carries annotations from both libraries to show how they nest.
|
||||
* Instances are configured in application.yaml. See docs/06-what-is-left-for-resilience4j.md.
|
||||
*/
|
||||
@Service
|
||||
public class PaymentClient {
|
||||
|
||||
private final CallLog log;
|
||||
private volatile boolean failing = true;
|
||||
|
||||
public PaymentClient(CallLog log) {
|
||||
this.log = log;
|
||||
}
|
||||
|
||||
public void setFailing(boolean failing) {
|
||||
this.failing = failing;
|
||||
}
|
||||
|
||||
private String call() {
|
||||
int n = log.record();
|
||||
if (failing) {
|
||||
throw new TransientException("payment gateway 503 (call " + n + ")");
|
||||
}
|
||||
return "charged";
|
||||
}
|
||||
|
||||
/** Resilience4j retry: maxAttempts counts the FIRST call too. */
|
||||
@Retry(name = "payments")
|
||||
public String r4jRetry() {
|
||||
return call();
|
||||
}
|
||||
|
||||
@CircuitBreaker(name = "payments")
|
||||
public String charge() {
|
||||
return call();
|
||||
}
|
||||
|
||||
/** Both libraries on one method. Which one is on the outside decides what the breaker counts. */
|
||||
@CircuitBreaker(name = "combo")
|
||||
@Retryable(maxRetries = 3, delay = 10)
|
||||
public String chargeWithBoth() {
|
||||
return call();
|
||||
}
|
||||
|
||||
@Bulkhead(name = "reports")
|
||||
public String bulkheadReport() {
|
||||
log.enter();
|
||||
log.record();
|
||||
try {
|
||||
Thread.sleep(200);
|
||||
}
|
||||
catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
finally {
|
||||
log.exit();
|
||||
}
|
||||
return "rendered";
|
||||
}
|
||||
|
||||
/** TimeLimiter only applies to asynchronous return types. */
|
||||
@TimeLimiter(name = "slow")
|
||||
public CompletableFuture<String> slowAsync() {
|
||||
return CompletableFuture.supplyAsync(() -> {
|
||||
log.record();
|
||||
sleep(2000);
|
||||
return "too late";
|
||||
});
|
||||
}
|
||||
|
||||
private static void sleep(long ms) {
|
||||
try {
|
||||
Thread.sleep(ms);
|
||||
}
|
||||
catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
package com.ankurm.resilience.retry;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.UncheckedIOException;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import com.ankurm.resilience.support.CallLog;
|
||||
import com.ankurm.resilience.support.TransientException;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import org.springframework.resilience.annotation.Retryable;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/**
|
||||
* A downstream that fails a configurable number of times before succeeding. Every method records
|
||||
* each real invocation in {@link CallLog}, so the demos count what actually ran.
|
||||
*
|
||||
* <p>Each method isolates one behaviour of Spring Framework 7's {@code @Retryable}; the
|
||||
* docs/03-retryable-measured.md chapter has the transcript for each.
|
||||
*/
|
||||
@Service
|
||||
public class FlakyGateway {
|
||||
|
||||
private final CallLog log;
|
||||
private final AtomicInteger failuresLeft = new AtomicInteger();
|
||||
|
||||
public FlakyGateway(CallLog log) {
|
||||
this.log = log;
|
||||
}
|
||||
|
||||
/** Fail the next {@code n} invocations, then succeed. */
|
||||
public void failNext(int n) {
|
||||
failuresLeft.set(n);
|
||||
}
|
||||
|
||||
private String attempt() {
|
||||
int call = log.record();
|
||||
if (failuresLeft.getAndDecrement() > 0) {
|
||||
throw new TransientException("attempt " + call + " failed");
|
||||
}
|
||||
return "ok after " + call + " invocation(s)";
|
||||
}
|
||||
|
||||
/** All defaults: any exception, maxRetries = 3, delay = 1000 ms, no back-off growth. */
|
||||
@Retryable
|
||||
public String defaults() {
|
||||
return attempt();
|
||||
}
|
||||
|
||||
/** Exponential: 100, 200, 400, 800 ms - capped at maxDelay. */
|
||||
@Retryable(maxRetries = 5, delay = 100, multiplier = 2, maxDelay = 500)
|
||||
public String exponential() {
|
||||
return attempt();
|
||||
}
|
||||
|
||||
/** delay 200 +/- jitter 100. */
|
||||
@Retryable(maxRetries = 6, delay = 200, jitter = 100)
|
||||
public String jittered() {
|
||||
return attempt();
|
||||
}
|
||||
|
||||
/** Only IllegalStateException is retryable; everything else fails on the first attempt. */
|
||||
@Retryable(includes = IllegalStateException.class, delay = 10)
|
||||
public String onlyIllegalState() {
|
||||
log.record();
|
||||
throw new IllegalArgumentException("not in includes");
|
||||
}
|
||||
|
||||
/**
|
||||
* includes = IOException, but what is thrown is an UncheckedIOException (a RuntimeException)
|
||||
* that WRAPS an IOException. Does the cause count?
|
||||
*/
|
||||
@Retryable(includes = IOException.class, maxRetries = 2, delay = 10)
|
||||
public String wrappedCause() {
|
||||
int call = log.record();
|
||||
throw new UncheckedIOException(new IOException("socket reset on attempt " + call));
|
||||
}
|
||||
|
||||
/**
|
||||
* A CompletableFuture that completes exceptionally. The method itself returns normally, so
|
||||
* from the interceptor's point of view there is nothing to retry.
|
||||
*/
|
||||
@Retryable(delay = 10)
|
||||
public CompletableFuture<String> future() {
|
||||
try {
|
||||
return CompletableFuture.completedFuture(attempt());
|
||||
}
|
||||
catch (RuntimeException ex) {
|
||||
return CompletableFuture.failedFuture(ex);
|
||||
}
|
||||
}
|
||||
|
||||
/** A Mono: retried by re-subscribing, so the deferred body runs again each time. */
|
||||
@Retryable(delay = 10)
|
||||
public Mono<String> mono() {
|
||||
return Mono.fromCallable(this::attempt);
|
||||
}
|
||||
|
||||
/** Each attempt takes 300 ms; the whole retry sequence has a 1 s budget. */
|
||||
@Retryable(maxRetries = 10, delay = 100, timeout = 1000)
|
||||
public String slowWithTimeout() {
|
||||
sleep(300);
|
||||
return attempt();
|
||||
}
|
||||
|
||||
/** One attempt that hangs for 1.5 s, with a 500 ms budget. Is the attempt interrupted? */
|
||||
@Retryable(maxRetries = 3, delay = 10, timeout = 500)
|
||||
public String hangingWithTimeout() {
|
||||
log.record();
|
||||
sleep(1500);
|
||||
throw new TransientException("slow failure");
|
||||
}
|
||||
|
||||
/** Self-invocation: calls the @Retryable method on {@code this}, bypassing the proxy. */
|
||||
public String selfInvocation() {
|
||||
return defaults();
|
||||
}
|
||||
|
||||
private static void sleep(long ms) {
|
||||
try {
|
||||
Thread.sleep(ms);
|
||||
}
|
||||
catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
package com.ankurm.resilience.retry;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
|
||||
import io.micrometer.core.instrument.MeterRegistry;
|
||||
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.resilience.retry.MethodRetryEvent;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* Spring's retry support publishes a {@link MethodRetryEvent} per failed attempt and records no
|
||||
* metrics of its own. This listener is the whole of the observability story unless you write
|
||||
* one - here it keeps the events for the demos and turns them into a Micrometer counter.
|
||||
*/
|
||||
@Component
|
||||
public class RetryEvents {
|
||||
|
||||
private final List<String> events = new CopyOnWriteArrayList<>();
|
||||
private final MeterRegistry registry;
|
||||
|
||||
public RetryEvents(MeterRegistry registry) {
|
||||
this.registry = registry;
|
||||
}
|
||||
|
||||
@EventListener
|
||||
void on(MethodRetryEvent event) {
|
||||
events.add(event.getMethod().getName() + " failed with " + event.getFailure().getClass().getSimpleName()
|
||||
+ (event.isRetryAborted() ? " -> retry aborted" : " -> will retry"));
|
||||
registry.counter("app.retry.failures",
|
||||
"method", event.getMethod().getName(),
|
||||
"aborted", String.valueOf(event.isRetryAborted())).increment();
|
||||
}
|
||||
|
||||
public List<String> drain() {
|
||||
List<String> copy = List.copyOf(events);
|
||||
events.clear();
|
||||
return copy;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
package com.ankurm.resilience.support;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* Records every real invocation of a guarded method: when it happened (milliseconds after the
|
||||
* first call) and on which thread. The demos report from this rather than from the retry
|
||||
* machinery's own view, so what is counted is what actually ran.
|
||||
*/
|
||||
@Component
|
||||
public class CallLog {
|
||||
|
||||
private final List<Long> times = new ArrayList<>();
|
||||
private final List<String> threads = new ArrayList<>();
|
||||
private final AtomicInteger inFlight = new AtomicInteger();
|
||||
private final AtomicInteger maxInFlight = new AtomicInteger();
|
||||
private long start;
|
||||
|
||||
public synchronized void reset() {
|
||||
times.clear();
|
||||
threads.clear();
|
||||
inFlight.set(0);
|
||||
maxInFlight.set(0);
|
||||
start = System.nanoTime();
|
||||
}
|
||||
|
||||
public synchronized int record() {
|
||||
times.add((System.nanoTime() - start) / 1_000_000);
|
||||
threads.add(Thread.currentThread().isVirtual() ? "virtual" : Thread.currentThread().getName());
|
||||
return times.size();
|
||||
}
|
||||
|
||||
public void enter() {
|
||||
int now = inFlight.incrementAndGet();
|
||||
maxInFlight.accumulateAndGet(now, Math::max);
|
||||
}
|
||||
|
||||
public void exit() {
|
||||
inFlight.decrementAndGet();
|
||||
}
|
||||
|
||||
public synchronized int count() {
|
||||
return times.size();
|
||||
}
|
||||
|
||||
public synchronized List<Long> times() {
|
||||
return List.copyOf(times);
|
||||
}
|
||||
|
||||
/** Gaps between consecutive invocations, in ms: the effective back-off. */
|
||||
public synchronized List<Long> gaps() {
|
||||
List<Long> gaps = new ArrayList<>();
|
||||
for (int i = 1; i < times.size(); i++) {
|
||||
gaps.add(times.get(i) - times.get(i - 1));
|
||||
}
|
||||
return gaps;
|
||||
}
|
||||
|
||||
public int maxInFlight() {
|
||||
return maxInFlight.get();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package com.ankurm.resilience.support;
|
||||
|
||||
/** What a flaky downstream throws: worth retrying. */
|
||||
public class TransientException extends RuntimeException {
|
||||
public TransientException(String message) {
|
||||
super(message);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
package com.ankurm.resilience.tx;
|
||||
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
|
||||
/**
|
||||
* The same retryable, transactional method - called from inside an existing transaction. Every
|
||||
* attempt joins the caller's transaction, the first failure marks it rollback-only, and the
|
||||
* retry that "succeeds" cannot un-mark it.
|
||||
*/
|
||||
@Service
|
||||
public class OrderFacade {
|
||||
|
||||
private final StockWriter writer;
|
||||
|
||||
public OrderFacade(StockWriter writer) {
|
||||
this.writer = writer;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public void placeOrder(String sku) {
|
||||
writer.record(sku);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
package com.ankurm.resilience.tx;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
|
||||
import com.ankurm.resilience.support.TransientException;
|
||||
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.resilience.annotation.Retryable;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
|
||||
/**
|
||||
* {@code @Retryable} and {@code @Transactional} on the same method. Each attempt inserts a row and
|
||||
* the first two then fail. What ends up in the table - and which transaction each attempt ran in -
|
||||
* shows which interceptor is outside the other. See docs/05-retry-and-transactions.md.
|
||||
*/
|
||||
@Service
|
||||
public class StockWriter {
|
||||
|
||||
private final JdbcTemplate jdbc;
|
||||
private final DataSource dataSource;
|
||||
private final AtomicInteger failuresLeft = new AtomicInteger();
|
||||
private final List<String> attempts = new ArrayList<>();
|
||||
|
||||
public StockWriter(JdbcTemplate jdbc, DataSource dataSource) {
|
||||
this.jdbc = jdbc;
|
||||
this.dataSource = dataSource;
|
||||
}
|
||||
|
||||
public void failNext(int n) {
|
||||
failuresLeft.set(n);
|
||||
attempts.clear();
|
||||
jdbc.update("delete from stock_movement");
|
||||
}
|
||||
|
||||
@Retryable(maxRetries = 3, delay = 10)
|
||||
@Transactional
|
||||
public void record(String sku) {
|
||||
Object holder = TransactionSynchronizationManager.getResource(dataSource);
|
||||
attempts.add("attempt " + (attempts.size() + 1)
|
||||
+ ": transaction active=" + TransactionSynchronizationManager.isActualTransactionActive()
|
||||
+ ", connection holder @" + Integer.toHexString(System.identityHashCode(holder)));
|
||||
jdbc.update("insert into stock_movement(sku) values (?)", sku);
|
||||
if (failuresLeft.getAndDecrement() > 0) {
|
||||
throw new TransientException("deadlock victim (simulated)");
|
||||
}
|
||||
}
|
||||
|
||||
public int rows() {
|
||||
Integer n = jdbc.queryForObject("select count(*) from stock_movement", Integer.class);
|
||||
return n == null ? 0 : n;
|
||||
}
|
||||
|
||||
public List<String> attempts() {
|
||||
return List.copyOf(attempts);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,252 @@
|
||||
package com.ankurm.resilience.web;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import com.ankurm.resilience.limit.ReportService;
|
||||
import com.ankurm.resilience.r4j.PaymentClient;
|
||||
import com.ankurm.resilience.retry.FlakyGateway;
|
||||
import com.ankurm.resilience.retry.RetryEvents;
|
||||
import com.ankurm.resilience.support.CallLog;
|
||||
import com.ankurm.resilience.tx.OrderFacade;
|
||||
import com.ankurm.resilience.tx.StockWriter;
|
||||
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
|
||||
import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
|
||||
|
||||
import org.springframework.aop.framework.Advised;
|
||||
import org.springframework.aop.support.AopUtils;
|
||||
import org.springframework.web.bind.annotation.GetMapping;
|
||||
import org.springframework.web.bind.annotation.PathVariable;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
/**
|
||||
* One endpoint per scenario. Each resets the call log, runs the scenario, and reports what really
|
||||
* happened: how many times the guarded method ran, the gaps between runs, and what the caller got
|
||||
* back. scripts/demo-*.sh curl these and commit the output. Delete before shipping anything real.
|
||||
*/
|
||||
@RestController
|
||||
public class DemoController {
|
||||
|
||||
private final FlakyGateway gateway;
|
||||
private final ReportService reports;
|
||||
private final StockWriter writer;
|
||||
private final OrderFacade facade;
|
||||
private final PaymentClient payments;
|
||||
private final CallLog log;
|
||||
private final RetryEvents events;
|
||||
private final CircuitBreakerRegistry breakers;
|
||||
|
||||
public DemoController(FlakyGateway gateway, ReportService reports, StockWriter writer, OrderFacade facade,
|
||||
PaymentClient payments, CallLog log, RetryEvents events, CircuitBreakerRegistry breakers) {
|
||||
this.gateway = gateway;
|
||||
this.reports = reports;
|
||||
this.writer = writer;
|
||||
this.facade = facade;
|
||||
this.payments = payments;
|
||||
this.log = log;
|
||||
this.events = events;
|
||||
this.breakers = breakers;
|
||||
}
|
||||
|
||||
@GetMapping("/demo/retry/{scenario}")
|
||||
public Map<String, Object> retry(@PathVariable String scenario) {
|
||||
log.reset();
|
||||
events.drain();
|
||||
return switch (scenario) {
|
||||
case "defaults" -> run(2, gateway::defaults);
|
||||
case "exhausted" -> run(99, gateway::defaults);
|
||||
case "exponential" -> run(99, gateway::exponential);
|
||||
case "jitter" -> run(99, gateway::jittered);
|
||||
case "includes" -> run(0, gateway::onlyIllegalState);
|
||||
case "cause" -> run(0, gateway::wrappedCause);
|
||||
case "future" -> run(99, () -> gateway.future().get(5, TimeUnit.SECONDS));
|
||||
case "mono" -> run(2, () -> gateway.mono().block());
|
||||
case "timeout" -> run(99, gateway::slowWithTimeout);
|
||||
case "hang" -> run(0, gateway::hangingWithTimeout);
|
||||
case "self-invocation" -> run(2, gateway::selfInvocation);
|
||||
default -> Map.of("error", "unknown scenario " + scenario);
|
||||
};
|
||||
}
|
||||
|
||||
@GetMapping("/demo/tx/{scenario}")
|
||||
public Map<String, Object> tx(@PathVariable String scenario) {
|
||||
events.drain();
|
||||
writer.failNext(2);
|
||||
Map<String, Object> out = new LinkedHashMap<>();
|
||||
try {
|
||||
if (scenario.equals("joined")) {
|
||||
facade.placeOrder("SKU-1");
|
||||
}
|
||||
else {
|
||||
writer.record("SKU-1");
|
||||
}
|
||||
out.put("caller got", "success");
|
||||
}
|
||||
catch (RuntimeException ex) {
|
||||
out.put("caller got", ex.getClass().getName() + ": " + ex.getMessage());
|
||||
}
|
||||
out.put("attempts", writer.attempts());
|
||||
out.put("rows in stock_movement", writer.rows());
|
||||
return out;
|
||||
}
|
||||
|
||||
@GetMapping("/demo/limit/{scenario}")
|
||||
public Map<String, Object> limit(@PathVariable String scenario) throws Exception {
|
||||
log.reset();
|
||||
Callable<String> task = switch (scenario) {
|
||||
case "block" -> reports::blocking;
|
||||
case "reject" -> reports::rejecting;
|
||||
case "r4j-bulkhead" -> payments::bulkheadReport;
|
||||
case "limit-and-retry" -> reports::limitedAndRetried;
|
||||
default -> reports::unlimited;
|
||||
};
|
||||
return concurrently(scenario.equals("limit-and-retry") ? 2 : 10, task);
|
||||
}
|
||||
|
||||
@GetMapping("/demo/r4j/{scenario}")
|
||||
public Map<String, Object> r4j(@PathVariable String scenario) throws Exception {
|
||||
log.reset();
|
||||
events.drain();
|
||||
payments.setFailing(true);
|
||||
return switch (scenario) {
|
||||
case "retry" -> run(0, payments::r4jRetry);
|
||||
case "breaker" -> breaker();
|
||||
case "combo" -> combo();
|
||||
case "timelimiter" -> run(0, () -> payments.slowAsync().get());
|
||||
default -> Map.of("error", "unknown scenario " + scenario);
|
||||
};
|
||||
}
|
||||
|
||||
/** The advisor chain on a bean's proxy, outermost first. */
|
||||
@GetMapping("/demo/proxy/{bean}")
|
||||
public List<String> proxy(@PathVariable String bean) {
|
||||
Object target = switch (bean) {
|
||||
case "payments" -> payments;
|
||||
case "stock-writer" -> writer;
|
||||
case "reports" -> reports;
|
||||
case "gateway" -> gateway;
|
||||
default -> throw new IllegalArgumentException(bean);
|
||||
};
|
||||
List<String> chain = new ArrayList<>();
|
||||
chain.add("proxy type: " + (AopUtils.isCglibProxy(target) ? "CGLIB" : AopUtils.isJdkDynamicProxy(target) ? "JDK" : "none"));
|
||||
if (AopUtils.isAopProxy(target) && target instanceof Advised advised) {
|
||||
for (var advisor : advised.getAdvisors()) {
|
||||
chain.add(advisor.getAdvice().getClass().getName()
|
||||
+ (advisor instanceof org.springframework.core.Ordered o ? " order=" + o.getOrder() : ""));
|
||||
}
|
||||
}
|
||||
return chain;
|
||||
}
|
||||
|
||||
private Map<String, Object> breaker() throws InterruptedException {
|
||||
CircuitBreaker cb = breakers.circuitBreaker("payments");
|
||||
cb.reset();
|
||||
List<String> outcomes = new ArrayList<>();
|
||||
for (int i = 1; i <= 8; i++) {
|
||||
outcomes.add(i + ": " + outcome(payments::charge) + " [state after: " + cb.getState() + "]");
|
||||
}
|
||||
int reachedWhileClosed = log.count();
|
||||
Thread.sleep(2100);
|
||||
payments.setFailing(false);
|
||||
outcomes.add("-- 2.1 s later, downstream recovered --");
|
||||
for (int i = 9; i <= 11; i++) {
|
||||
outcomes.add(i + ": " + outcome(payments::charge) + " [state after: " + cb.getState() + "]");
|
||||
}
|
||||
Map<String, Object> out = new LinkedHashMap<>();
|
||||
out.put("calls made", 11);
|
||||
out.put("calls that reached the downstream", log.count());
|
||||
out.put("of which before the breaker opened", reachedWhileClosed);
|
||||
out.put("outcomes", outcomes);
|
||||
return out;
|
||||
}
|
||||
|
||||
private Map<String, Object> combo() {
|
||||
CircuitBreaker cb = breakers.circuitBreaker("combo");
|
||||
cb.reset();
|
||||
Map<String, Object> out = new LinkedHashMap<>();
|
||||
for (int call = 1; call <= 2; call++) {
|
||||
int before = log.count();
|
||||
long start = System.nanoTime();
|
||||
String result = outcome(payments::chargeWithBoth);
|
||||
Map<String, Object> row = new LinkedHashMap<>();
|
||||
row.put("caller got", result);
|
||||
row.put("method body ran", log.count() - before);
|
||||
row.put("elapsed (ms)", (System.nanoTime() - start) / 1_000_000);
|
||||
row.put("breaker: buffered / failed / not permitted", cb.getMetrics().getNumberOfBufferedCalls() + " / "
|
||||
+ cb.getMetrics().getNumberOfFailedCalls() + " / " + cb.getMetrics().getNumberOfNotPermittedCalls());
|
||||
row.put("breaker state", cb.getState().toString());
|
||||
out.put("call " + call, row);
|
||||
}
|
||||
out.put("MethodRetryEvents", events.drain());
|
||||
return out;
|
||||
}
|
||||
|
||||
private Map<String, Object> run(int failures, Callable<?> action) {
|
||||
gateway.failNext(failures);
|
||||
long start = System.nanoTime();
|
||||
Object result;
|
||||
try {
|
||||
result = "returned: " + action.call();
|
||||
}
|
||||
catch (Exception ex) {
|
||||
Throwable t = ex instanceof ExecutionException && ex.getCause() != null ? ex.getCause() : ex;
|
||||
result = "threw: " + t.getClass().getName() + ": " + t.getMessage();
|
||||
}
|
||||
long elapsed = (System.nanoTime() - start) / 1_000_000;
|
||||
Map<String, Object> out = new LinkedHashMap<>();
|
||||
out.put("invocations", log.count());
|
||||
out.put("started at (ms)", log.times());
|
||||
out.put("gaps (ms)", log.gaps());
|
||||
out.put("elapsed (ms)", elapsed);
|
||||
out.put("caller", result);
|
||||
List<String> ev = events.drain();
|
||||
if (!ev.isEmpty()) {
|
||||
out.put("MethodRetryEvents", ev);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
private Map<String, Object> concurrently(int callers, Callable<String> task) throws Exception {
|
||||
long start = System.nanoTime();
|
||||
int ok = 0;
|
||||
Map<String, Integer> failures = new LinkedHashMap<>();
|
||||
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
|
||||
List<Future<String>> futures = new ArrayList<>();
|
||||
for (int i = 0; i < callers; i++) {
|
||||
futures.add(executor.submit(task));
|
||||
}
|
||||
for (Future<String> f : futures) {
|
||||
try {
|
||||
f.get();
|
||||
ok++;
|
||||
}
|
||||
catch (ExecutionException ex) {
|
||||
failures.merge(ex.getCause().getClass().getName(), 1, Integer::sum);
|
||||
}
|
||||
}
|
||||
}
|
||||
Map<String, Object> out = new LinkedHashMap<>();
|
||||
out.put("callers", callers);
|
||||
out.put("succeeded", ok);
|
||||
out.put("failed", failures);
|
||||
out.put("max concurrently inside the method", log.maxInFlight());
|
||||
out.put("elapsed (ms)", (System.nanoTime() - start) / 1_000_000);
|
||||
return out;
|
||||
}
|
||||
|
||||
private static String outcome(Callable<?> action) {
|
||||
try {
|
||||
return "returned " + action.call();
|
||||
}
|
||||
catch (Exception ex) {
|
||||
return ex.getClass().getSimpleName() + ": " + ex.getMessage();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
spring:
|
||||
application:
|
||||
name: resilience
|
||||
threads:
|
||||
virtual:
|
||||
enabled: true
|
||||
datasource:
|
||||
url: jdbc:h2:mem:resilience;DB_CLOSE_DELAY=-1
|
||||
|
||||
server:
|
||||
port: 8080
|
||||
|
||||
management:
|
||||
endpoints:
|
||||
web:
|
||||
exposure:
|
||||
include: health,metrics,circuitbreakers,circuitbreakerevents,retries,bulkheads,prometheus
|
||||
endpoint:
|
||||
health:
|
||||
show-details: always
|
||||
health:
|
||||
circuitbreakers:
|
||||
enabled: true
|
||||
|
||||
# Resilience4j instances used by PaymentClient. Every value is spelled out: the library defaults
|
||||
# (sliding window 100, minimum 100 calls, 60 s open) would need 100 calls before the breaker
|
||||
# could open at all - see docs/06-what-is-left-for-resilience4j.md.
|
||||
resilience4j:
|
||||
retry:
|
||||
instances:
|
||||
payments:
|
||||
max-attempts: 3
|
||||
wait-duration: 10ms
|
||||
circuitbreaker:
|
||||
instances:
|
||||
payments: &breaker
|
||||
sliding-window-size: 10
|
||||
minimum-number-of-calls: 5
|
||||
failure-rate-threshold: 50
|
||||
wait-duration-in-open-state: 2s
|
||||
permitted-number-of-calls-in-half-open-state: 2
|
||||
automatic-transition-from-open-to-half-open-enabled: false
|
||||
combo: *breaker
|
||||
bulkhead:
|
||||
instances:
|
||||
reports:
|
||||
max-concurrent-calls: 2
|
||||
max-wait-duration: 0
|
||||
timelimiter:
|
||||
instances:
|
||||
slow:
|
||||
timeout-duration: 500ms
|
||||
cancel-running-future: true
|
||||
@@ -0,0 +1,4 @@
|
||||
create table if not exists stock_movement (
|
||||
id bigint generated by default as identity primary key,
|
||||
sku varchar(64) not null
|
||||
);
|
||||
@@ -0,0 +1,34 @@
|
||||
package com.ankurm.resilience;
|
||||
|
||||
import java.util.Arrays;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.resilience.annotation.ConcurrencyLimit;
|
||||
import org.springframework.resilience.annotation.EnableResilientMethods;
|
||||
import org.springframework.resilience.annotation.Retryable;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/** Defaults the article quotes, read from the annotations rather than from the documentation. */
|
||||
class ApiSurfaceTest {
|
||||
|
||||
@Test
|
||||
void retryableDefaults() throws Exception {
|
||||
assertThat(Retryable.class.getMethod("maxRetries").getDefaultValue()).isEqualTo(3L);
|
||||
assertThat(Retryable.class.getMethod("delay").getDefaultValue()).isEqualTo(1000L);
|
||||
assertThat(Retryable.class.getMethod("multiplier").getDefaultValue()).isEqualTo(1.0d);
|
||||
assertThat(Retryable.class.getMethod("timeout").getDefaultValue()).isEqualTo(0L);
|
||||
}
|
||||
|
||||
@Test
|
||||
void concurrencyLimitHasTwoPolicies() {
|
||||
assertThat(Arrays.stream(ConcurrencyLimit.ThrottlePolicy.values()).map(Enum::name))
|
||||
.containsExactly("BLOCK", "REJECT");
|
||||
}
|
||||
|
||||
@Test
|
||||
void enableResilientMethodsOrderIsLowestMinusOne() throws Exception {
|
||||
assertThat(EnableResilientMethods.class.getMethod("order").getDefaultValue()).isEqualTo(Integer.MAX_VALUE - 1);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,104 @@
|
||||
package com.ankurm.resilience;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
|
||||
import com.ankurm.resilience.limit.ReportService;
|
||||
import com.ankurm.resilience.r4j.PaymentClient;
|
||||
import com.ankurm.resilience.support.CallLog;
|
||||
import com.ankurm.resilience.tx.OrderFacade;
|
||||
import com.ankurm.resilience.tx.StockWriter;
|
||||
import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
|
||||
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
|
||||
import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.resilience.InvocationRejectedException;
|
||||
import org.springframework.transaction.UnexpectedRollbackException;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
/** How the resilience interceptors nest with transactions, with each other, and with Resilience4j. */
|
||||
@SpringBootTest
|
||||
class CompositionContractTest {
|
||||
|
||||
@Autowired StockWriter writer;
|
||||
@Autowired OrderFacade facade;
|
||||
@Autowired ReportService reports;
|
||||
@Autowired PaymentClient payments;
|
||||
@Autowired CircuitBreakerRegistry breakers;
|
||||
@Autowired CallLog log;
|
||||
|
||||
@BeforeEach
|
||||
void reset() {
|
||||
log.reset();
|
||||
payments.setFailing(true);
|
||||
}
|
||||
|
||||
@Test
|
||||
void retryWrapsTheTransactionSoEachAttemptRollsBackOnItsOwn() {
|
||||
writer.failNext(2);
|
||||
writer.record("SKU-1");
|
||||
assertThat(writer.rows()).isEqualTo(1);
|
||||
assertThat(writer.attempts()).hasSize(3);
|
||||
}
|
||||
|
||||
@Test
|
||||
void insideAnOuterTransactionTheRetryCannotSaveIt() {
|
||||
writer.failNext(2);
|
||||
assertThatThrownBy(() -> facade.placeOrder("SKU-1")).isInstanceOf(UnexpectedRollbackException.class);
|
||||
assertThat(writer.rows()).isZero();
|
||||
}
|
||||
|
||||
@Test
|
||||
void blockPolicyQueuesAndRejectPolicyThrows() throws Exception {
|
||||
assertThat(runConcurrently(10, reports::blocking)).isEmpty();
|
||||
assertThat(log.maxInFlight()).isEqualTo(2);
|
||||
log.reset();
|
||||
assertThat(runConcurrently(10, reports::rejecting))
|
||||
.hasSize(8).allSatisfy(t -> assertThat(t).isInstanceOf(InvocationRejectedException.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void springRetryIsOutsideTheResilience4jBreaker() {
|
||||
CircuitBreaker cb = breakers.circuitBreaker("combo");
|
||||
cb.reset();
|
||||
assertThatThrownBy(payments::chargeWithBoth).isInstanceOf(RuntimeException.class);
|
||||
assertThat(log.count()).isEqualTo(4);
|
||||
assertThat(cb.getMetrics().getNumberOfBufferedCalls()).isEqualTo(4); // breaker saw every attempt
|
||||
|
||||
assertThatThrownBy(payments::chargeWithBoth).isInstanceOf(CallNotPermittedException.class);
|
||||
assertThat(cb.getMetrics().getNumberOfNotPermittedCalls()).isEqualTo(3); // retried an open circuit
|
||||
}
|
||||
|
||||
@Test
|
||||
void resilience4jMaxAttemptsCountsTheFirstCall() {
|
||||
assertThatThrownBy(payments::r4jRetry).isInstanceOf(RuntimeException.class);
|
||||
assertThat(log.count()).isEqualTo(3);
|
||||
}
|
||||
|
||||
private static List<Throwable> runConcurrently(int n, java.util.concurrent.Callable<String> task) throws Exception {
|
||||
List<Throwable> failures = new ArrayList<>();
|
||||
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
|
||||
List<Future<String>> futures = new ArrayList<>();
|
||||
for (int i = 0; i < n; i++) {
|
||||
futures.add(executor.submit(task));
|
||||
}
|
||||
for (Future<String> f : futures) {
|
||||
try {
|
||||
f.get();
|
||||
}
|
||||
catch (java.util.concurrent.ExecutionException ex) {
|
||||
failures.add(ex.getCause());
|
||||
}
|
||||
}
|
||||
}
|
||||
return failures;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,108 @@
|
||||
package com.ankurm.resilience;
|
||||
|
||||
import java.io.UncheckedIOException;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import com.ankurm.resilience.retry.FlakyGateway;
|
||||
import com.ankurm.resilience.support.CallLog;
|
||||
import com.ankurm.resilience.support.TransientException;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Nested;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
/**
|
||||
* Pins what Spring Framework 7.0.9's @Retryable actually does, counted by real invocations.
|
||||
* The one-second default delay makes a few of these slow on purpose.
|
||||
*/
|
||||
@SpringBootTest
|
||||
class RetryableContractTest {
|
||||
|
||||
@Autowired
|
||||
FlakyGateway gateway;
|
||||
|
||||
@Autowired
|
||||
CallLog log;
|
||||
|
||||
@BeforeEach
|
||||
void reset() {
|
||||
log.reset();
|
||||
}
|
||||
|
||||
@Test
|
||||
void maxRetriesThreeMeansFourInvocations() {
|
||||
gateway.failNext(99);
|
||||
assertThatThrownBy(gateway::defaults).isInstanceOf(TransientException.class)
|
||||
.hasMessage("attempt 4 failed"); // the LAST original exception, not a wrapper
|
||||
assertThat(log.count()).isEqualTo(4);
|
||||
assertThat(log.gaps()).allSatisfy(gap -> assertThat(gap).isBetween(950L, 1300L));
|
||||
}
|
||||
|
||||
@Test
|
||||
void exceptionNotInIncludesIsNotRetried() {
|
||||
assertThatThrownBy(gateway::onlyIllegalState).isInstanceOf(IllegalArgumentException.class);
|
||||
assertThat(log.count()).isEqualTo(1);
|
||||
}
|
||||
|
||||
@Test
|
||||
void includesMatchesAWrappedCause() {
|
||||
assertThatThrownBy(gateway::wrappedCause).isInstanceOf(UncheckedIOException.class);
|
||||
assertThat(log.count()).isEqualTo(3);
|
||||
}
|
||||
|
||||
@Test
|
||||
void aFailedCompletableFutureIsNotRetried() {
|
||||
gateway.failNext(99);
|
||||
assertThatThrownBy(() -> gateway.future().get(5, TimeUnit.SECONDS))
|
||||
.isInstanceOf(ExecutionException.class);
|
||||
assertThat(log.count()).isEqualTo(1);
|
||||
}
|
||||
|
||||
@Test
|
||||
void aMonoIsRetriedByResubscribing() {
|
||||
gateway.failNext(2);
|
||||
assertThat(gateway.mono().block()).isEqualTo("ok after 3 invocation(s)");
|
||||
assertThat(log.count()).isEqualTo(3);
|
||||
}
|
||||
|
||||
@Test
|
||||
void timeoutIsABudgetCheckedBetweenAttemptsNotAnInterrupt() {
|
||||
long start = System.nanoTime();
|
||||
assertThatThrownBy(gateway::hangingWithTimeout).isInstanceOf(TransientException.class);
|
||||
long elapsed = (System.nanoTime() - start) / 1_000_000;
|
||||
assertThat(log.count()).isEqualTo(1);
|
||||
assertThat(elapsed).isGreaterThanOrEqualTo(1500); // the 500 ms budget did not cut it short
|
||||
}
|
||||
|
||||
@Test
|
||||
void selfInvocationBypassesTheProxy() {
|
||||
gateway.failNext(1);
|
||||
assertThatThrownBy(gateway::selfInvocation).isInstanceOf(TransientException.class);
|
||||
assertThat(log.count()).isEqualTo(1);
|
||||
}
|
||||
|
||||
@Nested
|
||||
@SpringBootTest(properties = "demo.resilience.enabled=false")
|
||||
class WithoutEnableResilientMethods {
|
||||
|
||||
@Autowired
|
||||
FlakyGateway gateway;
|
||||
|
||||
@Autowired
|
||||
CallLog log;
|
||||
|
||||
@Test
|
||||
void theAnnotationIsInert() {
|
||||
log.reset();
|
||||
gateway.failNext(1);
|
||||
assertThatThrownBy(gateway::defaults).isInstanceOf(TransientException.class);
|
||||
assertThat(log.count()).isEqualTo(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user