Retry Mechanism for JDK HTTP Client

HttpInvocation.java

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.http.HttpResponse.BodyHandler;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.function.Predicate;

public class HttpInvocation<T> {

    private static final Logger LOG = LoggerFactory.getLogger(HttpInvocation.class);

    /**
     * Retry on all exceptions that inherits from IOException:
     * <ul>
     *     <li>{@link java.net.http.HttpTimeoutException}</li>
     *     <li>{@link java.net.http.HttpConnectTimeoutException}</li>
     *     <li>{@link java.nio.channels.ClosedChannelException}</li>
     *     <li>{@link sun.net.ConnectionResetException}</li>
     * </ul>
     */
    private static final Predicate<Throwable> DEFAULT_RETRY_ON_THROWABLE =
            ex -> ex instanceof IOException;

    /**
     * A default number of maximum retries on both types <b>on-response</b> and <b>on-throwable</b>
     */
    private static final int DEFAULT_MAX_ATTEMPTS = 5;

    /**
     * When a retry on-response exceeded then throw an exception by default.
     */
    private static final boolean DEFAULT_THROW_WHEN_RETRY_ON_RESPONSE_EXCEEDED = true;

    /**
     * By default it waits 5 seconds between two retries.
     */
    private static final Duration DEFAULT_RETRY_DELAY = Duration.ofSeconds(5);

    /**
     * If there is no {@link BodyHandler} specified, then discard entire incoming entity in a response.
     */
    private static final BodyHandler<Void> DEFAULT_BODY_HANDLER = HttpResponse.BodyHandlers.discarding();

    private final HttpClient client;
    private final HttpRequest request;
    private final BodyHandler<T> handler;
    private final AtomicInteger attempts = new AtomicInteger();
    private final Predicate<HttpResponse<?>> retryOnResponse;
    private final Predicate<Throwable> retryOnThrowable;
    private final int maxAttempts;
    private final boolean throwWhenRetryOnResponseExceeded;
    private final Executor delayedExecutor;

    private HttpInvocation(Builder<T> builder) {
        this.client = builder.client != null
                ? builder.client : HttpClient.newHttpClient();
        this.request = builder.request;
        this.handler = builder.bodyHandler;
        this.maxAttempts = builder.maxAttempts != null
                ? builder.maxAttempts : DEFAULT_MAX_ATTEMPTS;
        this.retryOnResponse = builder.retryOnResponse != null
                ? builder.retryOnResponse : DEFAULT_RETRY_ON_RESPONSE;
        this.retryOnThrowable = builder.retryOnThrowable != null
                ? builder.retryOnThrowable : DEFAULT_RETRY_ON_THROWABLE;
        this.throwWhenRetryOnResponseExceeded = builder.throwWhenRetryOnResponseExceeded != null
                ? builder.throwWhenRetryOnResponseExceeded : DEFAULT_THROW_WHEN_RETRY_ON_RESPONSE_EXCEEDED;

        Duration delay = builder.retryDelay != null
                ? builder.retryDelay
                : DEFAULT_RETRY_DELAY;
        this.delayedExecutor = CompletableFuture.delayedExecutor(
                delay.toMillis(), TimeUnit.MILLISECONDS);
    }
   
    /**
     * Retry on all Server Response (status code >= 500).
     *
     * @return default implementation of retry-on-response based on a status code.
     */
    private static <T> Predicate<HttpResponse<T>> defaultRetryOnResponse() {
        return resp -> resp.statusCode() >= 500;
    }

    /**
     * Invokes a configured {@link HttpInvocation} using {@link Builder} and
     * handle exceptions, incorrect responses and retries with a configured
     * delay.
     *
     * @return a completable future with a completed response or failed in
     * case of any exception.
     */
    public CompletableFuture<HttpResponse<T>> invoke() {
        attempts.incrementAndGet();
        return client.sendAsync(request, handler)
                .thenApply(resp -> {
                    if (retryOnResponse.test(resp)) {
                        return attemptRetry(resp, null);
                    } else {
                        return CompletableFuture.completedFuture(resp);
                    }
                })
                .exceptionally(ex -> {
                    // All internal exceptions are wrapped by `CompletionException`
                    if (retryOnThrowable.test(ex.getCause())) {
                        return attemptRetry(null, ex);
                    } else {
                        return CompletableFuture.failedFuture(ex);
                    }
                })
                .thenCompose(Function.identity());
    }

    /**
     * It tries to invoke the request again if there is any remaining attempt, or handle the situation
     * when a threshold of maximum attempts was exceeded.
     *
     * @param response  a failed response or <b>NULL</b>.
     * @param throwable a thrown exception or <b>NULL</b>.
     * @return a new completable future with a next attempt, or a failed response/exception in a case
     * of exceeded attempts.
     */
    private CompletableFuture<HttpResponse<T>> attemptRetry(HttpResponse<T> response, Throwable throwable) {
        if (attemptsRemains()) {
            LOG.warn("Retrying: attempt={} path={}", attempts.get() + 1, request.uri());
            return CompletableFuture.supplyAsync(this::invoke, delayedExecutor)
                    .thenCompose(Function.identity());
        } else {
            return handleRetryExceeded(response, throwable);
        }
    }

    /**
     * Defines the handler for an exceeded retry attempts. If the last attempt failed because of
     * an exception then throw it immediately. However, if the attempt failed on a regular response and
     * status code, them there are two possible behaviors based on the property {@link #throwWhenRetryOnResponseExceeded}.
     * <ul>
     *     <li><b>TRUE</b> when {@link #maxAttempts} is exceeded then an exception is thrown</li>
     *     <li><b>FALSE</b> when {@link #maxAttempts} is exceeded then the latest {@link HttpResponse}
     *     is returned</li>
     * </ul>
     *
     * @param response the very latest response object
     * @return a new completable future with a completed or failed state
     * depending on {@link #throwWhenRetryOnResponseExceeded}
     */
    private CompletableFuture<HttpResponse<T>> handleRetryExceeded(
            HttpResponse<T> response, Throwable throwable) {

        if (throwable != null || throwWhenRetryOnResponseExceeded) {
            Throwable ex = throwable == null
                    ? new RuntimeException("Retries exceeded: status-code=" + response.statusCode())
                    : throwable;

            return CompletableFuture.failedFuture(ex);
        } else {
            return CompletableFuture.completedFuture(response);
        }
    }

    /**
     * Returns <b>TRUE</b> if the number of retries has not exceeded the predefined
     * {@link #maxAttempts} value.
     *
     * @return <b>TRUE</b> if some attempts still remaining.
     */
    private boolean attemptsRemains() {
        return attempts.get() < maxAttempts;
    }

    /**
     * Creates a builder without an explicit {@link BodyHandler} which means that the default
     * {@link #DEFAULT_BODY_HANDLER} (discarding) with a return type {@link Void}.
     *
     * @param request an http request to invoke.
     * @return a builder with predefined <b>request</b> and a body-handler {@link #DEFAULT_BODY_HANDLER}.
     */
    public static Builder<Void> builder(HttpRequest request) {
        return new Builder<>(request, DEFAULT_BODY_HANDLER);
    }

    /**
     * Creates a builder along with a {@link BodyHandler} that determines the return type
     * defined by a generic <b>T</b>.
     *
     * @param request     an http request to invoke.
     * @param bodyHandler a handler to process an incoming entity in a response.
     * @param <T>         a type of a body of incoming entity.
     * @return a builder with predefined <b>request</b> and <b>bodyHandler</b>.
     */
    public static <T> Builder<T> builder(HttpRequest request, BodyHandler<T> bodyHandler) {
        return new Builder<>(request, bodyHandler);
    }

    public static final class Builder<T> {
        private final HttpRequest request;
        private final BodyHandler<T> bodyHandler;
        private HttpClient client;
        private Integer maxAttempts;
        private Duration retryDelay;
        private Predicate<HttpResponse<?>> retryOnResponse;
        private Predicate<Throwable> retryOnThrowable;
        private Boolean throwWhenRetryOnResponseExceeded;

        public Builder(HttpRequest request, BodyHandler<T> bodyHandler) {
            this.request = request;
            this.bodyHandler = bodyHandler;
        }

        public Builder<T> withHttpClient(HttpClient client) {
            this.client = client;
            return this;
        }

        public Builder<T> withMaxAttempts(int maxAttempts) {
            this.maxAttempts = maxAttempts;
            return this;
        }

        public Builder<T> withRetryDelay(Duration retryDelay) {
            this.retryDelay = retryDelay;
            return this;
        }

        public Builder<T> withRetryOnResponse(Predicate<HttpResponse<?>> retryOnResponse) {
            this.retryOnResponse = retryOnResponse;
            return this;
        }

        public Builder<T> withRetryOnThrowable(Predicate<Throwable> retryOnThrowable) {
            this.retryOnThrowable = retryOnThrowable;
            return this;
        }

        public Builder<T> withThrowWhenRetryOnResponseExceeded(boolean throwWhenRetryOnResponseExceeded) {
            this.throwWhenRetryOnResponseExceeded = throwWhenRetryOnResponseExceeded;
            return this;
        }

        public HttpInvocation<T> build() {
            return new HttpInvocation<>(this);
        }
    }
}

tests.java

import ch.qos.logback.classic.spi.ILoggingEvent;
import ... <different-gist> AssertAppender;
import ...MockServerContainer;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.function.Executable;
import org.mockserver.client.MockServerClient;
import org.mockserver.matchers.Times;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;

import java.net.ConnectException;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockserver.model.HttpRequest.request;
import static org.mockserver.model.HttpResponse.response;

@Testcontainers
class HttpInvocationTest {

    private static final String PATH = "/rest/api/latest";

    @Container
    private static final MockServerContainer CONTAINER = new MockServerContainer();

    private static HttpRequest REQUEST;

    @BeforeAll
    public static void setup() {
        REQUEST = HttpRequest.newBuilder()
                .uri(CONTAINER.url().resolve(PATH))
                .GET()
                .build();
    }

    @Test
    public void singleSuccessInvocationWithoutBody() throws Exception {
        MockServerClient mockClient = CONTAINER.cleanClient();

        mockClient
                .when(request()
                        .withMethod("GET")
                        .withPath(PATH)
                )
                .respond(response()
                        .withStatusCode(200)
                );

        HttpResponse<Void> response = HttpInvocation.builder(REQUEST)
                .build()
                .invoke()
                .get(1, TimeUnit.SECONDS);

        assertEquals(200, response.statusCode());
    }

    @Test
    public void singleSuccessInvocationWithBody() throws Exception {
        MockServerClient mockClient = CONTAINER.cleanClient();

        mockClient
                .when(request()
                        .withMethod("GET")
                        .withPath(PATH)
                )
                .respond(response()
                        .withStatusCode(200)
                        .withBody("Body")
                );

        HttpResponse<String> response =
                HttpInvocation.builder(REQUEST, HttpResponse.BodyHandlers.ofString())
                        .build()
                        .invoke()
                        .get(1, TimeUnit.SECONDS);

        assertEquals(200, response.statusCode());
        assertEquals("Body", response.body());
    }

    @Test
    public void successfulWithRetryOnResponseWithBody() throws Exception {
        MockServerClient mockClient = CONTAINER.cleanClient();

        mockClient
                .when(request()
                        .withMethod("GET")
                        .withPath(PATH)
                )
                .respond(response()
                        .withStatusCode(200)
                        .withBody("Body")
                );

        HttpResponse<String> response =
                HttpInvocation.builder(REQUEST, HttpResponse.BodyHandlers.ofString())
                        .withRetryOnResponse(resp -> !resp.body().equals("Weird Body"))
                        .build()
                        .invoke()
                        .get(1, TimeUnit.SECONDS);

        assertEquals(200, response.statusCode());
        assertEquals("Body", response.body());
    }

    @Test
    public void successfulRetry() throws Exception {
        MockServerClient mockClient = CONTAINER.cleanClient();

        mockClient
                .when(request()
                                .withMethod("GET")
                                .withPath(PATH),
                        Times.exactly(2)
                )
                .respond(response()
                        .withStatusCode(500)
                );

        mockClient
                .when(request()
                        .withMethod("GET")
                        .withPath(PATH)
                )
                .respond(response()
                        .withStatusCode(200)
                        .withBody("Body")
                );

        Predicate<ILoggingEvent> assertion = event ->
                event.toString().startsWith("[WARN] Retrying: attempt=3");

        try (var appender = new AssertAppender(HttpInvocation.class, assertion)) {
            appender.start();

            HttpResponse<String> response =
                    HttpInvocation.builder(REQUEST, HttpResponse.BodyHandlers.ofString())
                            .withMaxAttempts(3)
                            .withRetryDelay(Duration.ofMillis(100))
                            .build()
                            .invoke()
                            .get(1, TimeUnit.SECONDS);

            appender.waitForAssertion(Duration.ofSeconds(1));

            assertEquals(200, response.statusCode());
            assertEquals("Body", response.body());
        }
    }

    @Test
    public void attemptsExceededOnResponseThrowException() {
        MockServerClient mockClient = CONTAINER.cleanClient();

        mockClient
                .when(request()
                        .withMethod("GET")
                        .withPath(PATH)
                )
                .respond(response()
                        .withStatusCode(500)
                );

        Executable executable =
                () -> HttpInvocation.builder(REQUEST, HttpResponse.BodyHandlers.ofString())
                        .withMaxAttempts(3)
                        .withRetryDelay(Duration.ofMillis(100))
                        .withThrowWhenRetryOnResponseExceeded(true)
                        .build()
                        .invoke()
                        .get(1, TimeUnit.SECONDS);

        ExecutionException ex = assertThrows(ExecutionException.class, executable);
        assertEquals("java.lang.RuntimeException: Retries exceeded: status-code=500", ex.getMessage());
    }

    @Test
    public void attemptsExceededOnResponseReturnResponse() throws Exception {
        MockServerClient mockClient = CONTAINER.cleanClient();

        mockClient
                .when(request()
                        .withMethod("GET")
                        .withPath(PATH)
                )
                .respond(response()
                        .withStatusCode(500)
                );

        HttpResponse<String> response = HttpInvocation.builder(REQUEST, HttpResponse.BodyHandlers.ofString())
                .withMaxAttempts(3)
                .withRetryDelay(Duration.ofMillis(100))
                .withThrowWhenRetryOnResponseExceeded(false)
                .build()
                .invoke()
                .get(1, TimeUnit.SECONDS);

        assertEquals(500, response.statusCode());
    }

    @Test
    public void attemptsExceededOnIOException() throws Exception {
        HttpRequest request = HttpRequest.newBuilder(URI.create("http://non-existing"))
                .build();

        Predicate<ILoggingEvent> assertion = event ->
                event.toString().startsWith("[WARN] Retrying: attempt=3");

        try (var appender = new AssertAppender(HttpInvocation.class, assertion)) {
            appender.start();

            HttpClient httpClient = HttpClient.newBuilder()
                    .connectTimeout(Duration.ofMillis(10))
                    .build();

            Executable executable =
                    () -> HttpInvocation.builder(request, HttpResponse.BodyHandlers.ofString())
                            .withHttpClient(httpClient)
                            .withMaxAttempts(3)
                            .withRetryDelay(Duration.ofMillis(100))
                            .withThrowWhenRetryOnResponseExceeded(false)
                            .build()
                            .invoke()
                            .get(1, TimeUnit.SECONDS);

            ExecutionException ex = assertThrows(ExecutionException.class, executable);

            appender.waitForAssertion(Duration.ofSeconds(1));

            assertEquals(ConnectException.class, ex.getCause().getClass());
        }
    }
}

Is this public domain?

No!! you have to follow me on Twitter 😀 @p_bouda :)) .. of course it is! use it and make it better :)

…

On Wed, 25 May 2022, 19:03 scottgorman, ***@***.***> wrote: ***@***.**** commented on this gist. ------------------------------ Is this public domain? — Reply to this email directly, view it on GitHub <https://gist.github.com/92647b243eac71b089eb4fb2cfa90bf2#gistcomment-4178914>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAOTIOQVTF6FU6IN2M54A4TVLZMPZANCNFSM5W5ZKZ3A> . You are receiving this because you authored the thread.Message ID: ***@***.***>