spring reactive retry with exponential backoff conditionally

Question

Using spring reactive WebClient, I consume an API and in case of response with 500 status I need to retry with exponential backoff. But in Mono class, I don't see any retryBackoff with Predicate as input parameter.

This is the kind of function I search for:

public final Mono<T> retryBackoff(Predicate<? super Throwable> retryMatcher, long numRetries, Duration firstBackoff)

Right now my implementation is as following (I don't have retry with backOff mechanism):

client.sendRequest()
    .retry(e -> ((RestClientException) e).getStatus() == 500)
    .subscribe();

score 4 · Answer 1 · answered May 22 '19 at 10:57

You might want to have a look at the reactor-extra module in the reactor-addons project. In Maven you can do:

<dependency>
    <groupId>io.projectreactor.addons</groupId>
    <artifactId>reactor-extra</artifactId>
    <version>3.2.3.RELEASE</version>
</dependency>

And then use it like this:

client.post()
    .syncBody("test")
    .retrieve()
    .bodyToMono(String.class)
    .retryWhen(Retry.onlyIf(ctx -> ctx.exception() instanceof RestClientException)
                    .exponentialBackoff(firstBackoff, maxBackoff)
                    .retryMax(maxRetries))

score 4 · Answer 2 · edited Apr 26 '22 at 09:10

Retry.onlyIf is now deprecated/removed.

If anyone is interested in the up-to-date solution:

client.post()
      .syncBody("test")
      .retrieve()
      .bodyToMono(String.class)
      .retryWhen(Retry.backoff(maxRetries, minBackoff).filter(ctx -> {
          return ctx.exception() instanceof RestClientException && ctx.exception().statusCode == 500; 
      }))

It's worth mentioning that retryWhen wraps the source exception into the RetryExhaustedException. If you want to 'restore' the source exception you can use the reactor.core.Exceptions util:

.onErrorResume(throwable -> {
    if (Exceptions.isRetryExhausted(throwable)) {
        throwable = throwable.getCause();
    }
    return Mono.error(throwable);
})

score 0 · Answer 3 · answered May 10 '19 at 15:55

0

I'm not sure, what spring version you are using, in 2.1.4 I have this:

client.post()
    .syncBody("test")
    .retrieve()
    .bodyToMono(String.class)
    .retryBackoff(numretries, firstBackoff, maxBackoff, jitterFactor);

... so that's exactly what you want, right?

answered May 10 '19 at 15:55

Frischling

2,100
14
34

He want's to retry only on error 500, not 400, not 501...only 500. – programaths Aug 28 '19 at 06:29
1

`retryBackoff` was removed in Reactor 3.4. – Inego Mar 04 '21 at 03:52

score 0 · Answer 4 · answered Mar 23 '20 at 10:40

I'm currently trying it with Kotlin Coroutines + Spring WebFlux:

It seems the following is not working:

suspend fun ClientResponse.asResponse(): ServerResponse =
    status(statusCode())
        .headers { headerConsumer -> headerConsumer.addAll(headers().asHttpHeaders()) }
        .body(bodyToMono(DataBuffer::class.java), DataBuffer::class.java)
        .retryWhen { 
            Retry.onlyIf { ctx: RetryContext<Throwable> -> (ctx.exception() as? WebClientResponseException)?.statusCode in retryableErrorCodes }
                .exponentialBackoff(ofSeconds(1), ofSeconds(5))
                .retryMax(3)
                .doOnRetry { log.error("Retry for {}", it.exception()) }
        )
        .awaitSingle()

score 0 · Answer 5 · answered Nov 18 '22 at 17:32

AtomicInteger errorCount = new AtomicInteger();
Flux<String> flux =
Flux.<String>error(new IllegalStateException("boom"))
        .doOnError(e -> { 
            errorCount.incrementAndGet();
            System.out.println(e + " at " + LocalTime.now());
        })
        .retryWhen(Retry
                .backoff(3, Duration.ofMillis(100)).jitter(0d) 
                .doAfterRetry(rs -> System.out.println("retried at " + LocalTime.now() + ", attempt " + rs.totalRetries())) 
                .onRetryExhaustedThrow((spec, rs) -> rs.failure()) 
        );

We will log the time of errors emitted by the source and count them.

We configure an exponential backoff retry with at most 3 attempts and no jitter.

We also log the time at which the retry happens, and the retry attempt number (starting from 0).

By default, an Exceptions.retryExhausted exception would be thrown, with the last failure() as a cause. Here we customize that to directly emit the cause as onError.

spring reactive retry with exponential backoff conditionally

5 Answers5