CODEWITHSUNDEEP
kotlinspring-webfluxproject-reactorretry-logic
Peter
Peter

Reputation: 333

Custom retry in Reactor

I was trying to implement a retry logic in Kotlin and Reactor based on the Reactor extra package's features. What I'm trying to do is pass a list of durations, and on each context.iteration I'm getting the (iteration-1)th element of the list. It works partly, I'm always getting an IndexOutOfBoundsException on the last iteration, which is more than I wanted, although I've provided a max number of retries - the size of the list. The retries are running though, in the given duration and "correct" number of times (surely because IndexOutOfBoundsException prevents more), only this exception (and it's root cause) bothers me.

This is my custom BackOff interface:

interface MyCustomBackoff : Backoff {
    companion object {
        fun getBackoffDelay(backoffList: List<Duration>): (IterationContext<*>) -> BackoffDelay {
            return { context -> BackoffDelay(backoffList[(context.iteration() - 1).toInt()]) }
        }
    }
}

And my Kotlin extension is:

fun <T> Mono<T>.retryCustomBackoffs(backoffList: List<Duration>, doOnRetry: ((RetryContext<T>) -> Unit)? = null): Mono<T> {
    val retry = Retry.any<T>().retryMax(backoffList.size.toLong()).backoff(MyCustomBackoff.getBackoffDelay(backoffList))

    return if (doOnRetry == null) {
        this.retryWhen(retry)
    }
    else {
        this.retryWhen(retry.doOnRetry(doOnRetry))
    }
}

What am I missing here?

Upvotes: 0

Views: 1214

Answers (1)

Alexander Pankin
Alexander Pankin

Reputation: 3955

If you look at reactor.retry.AbstractRetry#calculateBackoff, you could find that there is special BackoffDelay named RETRY_EXHAUSTED. And it returns when retryContext.iteration() > maxIterations (not >=) after backoff.apply(retryContext)

if (retryContext.iteration() > maxIterations || Instant.now(clock).plus(jitteredBackoff).isAfter(timeoutInstant))
    return RETRY_EXHAUSTED;

So, if you have 2 custom backoff delays in the list, there will be 3 backoff delays generated by calculateBackoff.

You could change your MyCustomBackoff like so (excuse me for Java, I'm not familiar with Kotlin):

public interface MyCustomBackoff extends Backoff {
    static Backoff getBackoffDelay(List<Duration> backoffList) {
        return context -> context.iteration() <= backoffList.size() ?
                new BackoffDelay(backoffList.get(Long.valueOf(context.iteration() - 1).intValue())) :
                new BackoffDelay(Duration.ZERO);
    }
}

Upvotes: 1

Related Questions