Transport level retry

java-client/src/main/java/co/elastic/clients/transport/BackoffPolicy.java

@@ -55,7 +55,7 @@ public static BackoffPolicy noBackoff() {
     }
 
     /**
-     * Creates an new constant backoff policy with the provided configuration.
+     * Creates a new constant backoff policy with the provided configuration.
      *
      * @param delay              The delay defines how long to wait between retry attempts. Must not be null.
      *                           Must be &lt;= <code>Integer.MAX_VALUE</code> ms.
@@ -68,7 +68,7 @@ public static BackoffPolicy constantBackoff(Long delay, int maxNumberOfRetries)
     }
 
     /**
-     * Creates an new exponential backoff policy with a default configuration of 50 ms initial wait period and 8 retries taking
+     * Creates a new exponential backoff policy with a default configuration of 50 ms initial wait period and 8 retries taking
      * roughly 5.1 seconds in total.
      *
      * @return A backoff policy with an exponential increase in wait time for retries. The returned instance is thread safe but each
@@ -79,7 +79,7 @@ public static BackoffPolicy exponentialBackoff() {
     }
 
     /**
-     * Creates an new exponential backoff policy with the provided configuration.
+     * Creates a new exponential backoff policy with the provided configuration.
      *
      * @param initialDelay       The initial delay defines how long to wait for the first retry attempt. Must not be null.
      *                           Must be &lt;= <code>Integer.MAX_VALUE</code> ms.

java-client/src/main/java/co/elastic/clients/transport/DefaultTransportOptions.java

@@ -38,6 +38,7 @@ public class DefaultTransportOptions implements TransportOptions {
     private final Map<String, String> parameters;
     private final Function<List<String>, Boolean> onWarnings;
     private boolean keepResponseBodyOnException;
+    private RetryConfig retryConfig;
 
     public static final DefaultTransportOptions EMPTY = new DefaultTransportOptions();
 
@@ -51,8 +52,19 @@ public DefaultTransportOptions(
         @Nullable Function<List<String>, Boolean> onWarnings,
         boolean keepResponseBodyOnException
     ) {
-        this(headers,parameters,onWarnings);
+        this(headers, parameters, onWarnings, keepResponseBodyOnException, RetryConfig.disabled());
+    }
+
+    public DefaultTransportOptions(
+        @Nullable HeaderMap headers,
+        @Nullable Map<String, String> parameters,
+        @Nullable Function<List<String>, Boolean> onWarnings,
+        boolean keepResponseBodyOnException,
+        @Nullable RetryConfig retryConfig
+    ) {
+        this(headers, parameters, onWarnings);
         this.keepResponseBodyOnException = keepResponseBodyOnException;
+        this.retryConfig = retryConfig == null ? RetryConfig.disabled() : retryConfig;
     }
 
     public DefaultTransportOptions(
@@ -65,10 +77,11 @@ public DefaultTransportOptions(
             Collections.emptyMap() : Collections.unmodifiableMap(parameters);
         this.onWarnings = onWarnings;
         this.keepResponseBodyOnException = false;
+        this.retryConfig = RetryConfig.disabled();
     }
 
     protected DefaultTransportOptions(AbstractBuilder<?> builder) {
-        this(builder.headers, builder.parameters, builder.onWarnings, builder.keepResponseBodyOnException);
+        this(builder.headers, builder.parameters, builder.onWarnings, builder.keepResponseBodyOnException, builder.retryConfig);
     }
 
     public static DefaultTransportOptions of(@Nullable TransportOptions options) {
@@ -111,6 +124,11 @@ public boolean keepResponseBodyOnException() {
         return keepResponseBodyOnException;
     }
 
+    @Override
+    public RetryConfig retryConfig() {
+        return retryConfig;
+    }
+
     @Override
     public Builder toBuilder() {
         return new Builder(this);
@@ -135,6 +153,7 @@ public abstract static class AbstractBuilder<BuilderT extends AbstractBuilder<Bu
         private Map<String, String> parameters;
         private Function<List<String>, Boolean> onWarnings;
         private boolean keepResponseBodyOnException;
+        private RetryConfig retryConfig = RetryConfig.disabled();
 
         public AbstractBuilder() {
         }
@@ -144,6 +163,7 @@ public AbstractBuilder(DefaultTransportOptions options) {
             this.parameters = copyOrNull(options.parameters);
             this.onWarnings = options.onWarnings;
             this.keepResponseBodyOnException = options.keepResponseBodyOnException;
+            this.retryConfig = options.retryConfig == null ? RetryConfig.disabled() : options.retryConfig;
         }
 
         protected abstract BuilderT self();
@@ -154,6 +174,12 @@ public BuilderT keepResponseBodyOnException(boolean value) {
             return self();
         }
 
+        @Override
+        public BuilderT retryConfig(RetryConfig config) {
+            this.retryConfig = config == null ? RetryConfig.disabled() : config;
+            return self();
+        }
+
         @Override
         public BuilderT addHeader(String name, String value) {
             if (name.equalsIgnoreCase(HeaderMap.CLIENT_META)) {

java-client/src/main/java/co/elastic/clients/transport/ElasticsearchTransportBase.java

@@ -32,6 +32,7 @@
 import co.elastic.clients.transport.endpoints.TextEndpoint;
 import co.elastic.clients.transport.http.HeaderMap;
 import co.elastic.clients.transport.http.RepeatableBodyResponse;
+import co.elastic.clients.transport.http.RetryingHttpClient;
 import co.elastic.clients.transport.http.TransportHttpClient;
 import co.elastic.clients.transport.instrumentation.Instrumentation;
 import co.elastic.clients.transport.instrumentation.NoopInstrumentation;
@@ -80,6 +81,10 @@ public abstract class ElasticsearchTransportBase implements ElasticsearchTranspo
     }
 
     protected final TransportHttpClient httpClient;
+    // Single retry wrapper, shared by every request that uses retries (whether configured on the client or per
+    // request) and reading the effective RetryConfig from each call's TransportOptions.
+    private volatile RetryingHttpClient retryingHttpClient;
+    private boolean closed;
     protected final Instrumentation instrumentation;
     protected final JsonpMapper mapper;
     protected final TransportOptions transportOptions;
@@ -120,7 +125,18 @@ protected ElasticsearchTransportBase cloneWith(
 
     @Override
     public void close() throws IOException {
-        httpClient.close();
+        // The retry wrapper, if it was ever created, owns the delegate and the retry scheduler, so closing it
+        // also closes httpClient and shuts down the scheduler. Otherwise close httpClient directly.
+        RetryingHttpClient retrying;
+        synchronized (this) {
+            closed = true;
+            retrying = retryingHttpClient;
+        }
+        if (retrying != null) {
+            retrying.close();
+        } else {
+            httpClient.close();
+        }
     }
 
     @Override
@@ -137,6 +153,37 @@ public TransportHttpClient httpClient() {
         return httpClient;
     }
 
+    // Returns the shared retry wrapper when a request uses retries, the bare http client otherwise.
+    // Clients that never retry will never walk the retry path.
+    private TransportHttpClient httpClientFor(TransportOptions options) {
+        RetryConfig retryConfig = options.retryConfig();
+        if (retryConfig == null || !retryConfig.isEnabled()) {
+            return httpClient;
+        }
+        // If the user supplied a client that already retries, don't wrap it a second time.
+        if (httpClient instanceof RetryingHttpClient) {
+            return httpClient;
+        }
+        return retryingHttpClient();
+    }
+
+    private RetryingHttpClient retryingHttpClient() {
+        RetryingHttpClient client = retryingHttpClient;
+        if (client == null) {
+            synchronized (this) {
+                if (closed) {
+                    throw new IllegalStateException("Transport has been closed");
+                }
+                client = retryingHttpClient;
+                if (client == null) {
+                    client = new RetryingHttpClient(httpClient);
+                    retryingHttpClient = client;
+                }
+            }
+        }
+        return client;
+    }
+
     @Override
     public final <RequestT, ResponseT, ErrorT> ResponseT performRequest(
         RequestT request,
@@ -151,7 +198,7 @@ public final <RequestT, ResponseT, ErrorT> ResponseT performRequest(
                 TransportHttpClient.Request req = prepareTransportRequest(request, endpoint);
                 ctx.beforeSendingHttpRequest(req, options);
 
-                TransportHttpClient.Response resp = httpClient.performRequest(endpoint.id(), null, req, opts);
+                TransportHttpClient.Response resp = httpClientFor(opts).performRequest(endpoint.id(), null, req, opts);
                 ctx.afterReceivingHttpResponse(resp);
 
                 ResponseT apiResponse = getApiResponse(resp, endpoint);
@@ -183,9 +230,7 @@ public final <RequestT, ResponseT, ErrorT> CompletableFuture<ResponseT> performR
             TransportHttpClient.Request clientReq = prepareTransportRequest(request, endpoint);
             ctx.beforeSendingHttpRequest(clientReq, options);
 
-            clientFuture = httpClient.performRequestAsync(
-                endpoint.id(), null, clientReq, opts
-            );
+            clientFuture = httpClientFor(opts).performRequestAsync(endpoint.id(), null, clientReq, opts);
         } catch (Exception e) {
             // Terminate early
             ctx.recordException(e);

java-client/src/main/java/co/elastic/clients/transport/ElasticsearchTransportConfig.java

@@ -30,6 +30,7 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.List;
+import java.util.function.Consumer;
 import java.util.function.Function;
 
 public abstract class ElasticsearchTransportConfig {
@@ -326,6 +327,20 @@ public BuilderT transportOptions(Function<TransportOptions.Builder, TransportOpt
             return self();
         }
 
+        /**
+         * Configure transport-level retries.
+         */
+        public BuilderT retryConfig(RetryConfig retryConfig) {
+            return transportOptions(o -> o.retryConfig(retryConfig));
+        }
+
+        /**
+         * Configure transport-level retries with a builder lambda.
+         */
+        public BuilderT retryConfig(Consumer<RetryConfig.Builder> fn) {
+            return transportOptions(o -> o.retryConfig(fn));
+        }
+
         protected void checkNull(Object value, String name, String other) {
             if (value != null) {
                 throw new IllegalArgumentException("Cannot set both " + other + " and " + name + ".");