[pull] main from opensearch-project:main

CHANGELOG.md

@@ -24,6 +24,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Add document lifecycle guide and runnable sample ([#2017](https://github.com/opensearch-project/opensearch-java/pull/2017))
 - Add transparent gRPC transport with HybridTransport (bulk over gRPC, REST fallback), translation layer, TLS, basic auth, AWS SigV4, and JWT support ([#2062](https://github.com/opensearch-project/opensearch-java/pull/2062))
 - Add search over gRPC with match_all query support, SearchRequestConverter, SearchResponseConverter, and _source deserialization ([#2071](https://github.com/opensearch-project/opensearch-java/pull/2071))
+- Add `setAutomaticRetriesDisabled` to `ApacheHttpClient5TransportBuilder` to allow enabling automatic retries ([#2086](https://github.com/opensearch-project/opensearch-java/pull/2086))
 
 ### Fixed
 - Fix `unitTest` task not running the tests in the `test` source set ([#2074](https://github.com/opensearch-project/opensearch-java/pull/2074))

java-client/src/main/java/org/opensearch/client/transport/httpclient5/ApacheHttpClient5TransportBuilder.java

@@ -8,7 +8,7 @@
 
 package org.opensearch.client.transport.httpclient5;
 
 import java.security.AccessController;
 import java.security.NoSuchAlgorithmException;
 import java.security.PrivilegedAction;
 import java.util.Arrays;
@@ -72,6 +72,7 @@
     private HttpClientConfigCallback httpClientConfigCallback;
     private RequestConfigCallback requestConfigCallback;
     private ConnectionConfigCallback connectionConfigCallback;
+    private boolean automaticRetriesDisabled = true;
     private String pathPrefix;
     private NodeSelector nodeSelector = NodeSelector.ANY;
     private boolean strictDeprecationMode = false;
@@ -163,6 +164,19 @@
         return this;
     }
 
+    /**
+     * Whether the http client should automatically retry requests. Automatic retries are disabled
+     * by default, on par with the previous 4.x http client.
+     *
+     * When enabled, the http client's default retry strategy is used, and a custom
+     * {@link org.apache.hc.client5.http.HttpRequestRetryStrategy} can be set through
+     * {@link HttpClientConfigCallback}.
+     */
+    public ApacheHttpClient5TransportBuilder setAutomaticRetriesDisabled(boolean automaticRetriesDisabled) {
+        this.automaticRetriesDisabled = automaticRetriesDisabled;
+        return this;
+    }
+
     /**
      * Sets the path's prefix for every request used by the http client.
      * <p>
@@ -282,7 +296,7 @@
         if (failureListener == null) {
             failureListener = new ApacheHttpClient5Transport.FailureListener();
         }
         CloseableHttpAsyncClient httpClient = AccessController.doPrivileged(
             (PrivilegedAction<CloseableHttpAsyncClient>) this::createHttpClient
         );
 
@@ -378,14 +392,17 @@
             HttpAsyncClientBuilder httpClientBuilder = HttpAsyncClientBuilder.create()
                 .setDefaultRequestConfig(requestConfigBuilder.build())
                 .setConnectionManager(connectionManager)
-                .setTargetAuthenticationStrategy(DefaultAuthenticationStrategy.INSTANCE)
-                .disableAutomaticRetries();
+                .setTargetAuthenticationStrategy(DefaultAuthenticationStrategy.INSTANCE);
+            if (automaticRetriesDisabled) {
+                // Keep behavior on par with the 4.x http client, which had no automatic retries
+                httpClientBuilder = httpClientBuilder.disableAutomaticRetries();
+            }
             if (httpClientConfigCallback != null) {
                 httpClientBuilder = httpClientConfigCallback.customizeHttpClient(httpClientBuilder);
             }
 
             final HttpAsyncClientBuilder finalBuilder = httpClientBuilder;
             return AccessController.doPrivileged((PrivilegedAction<CloseableHttpAsyncClient>) finalBuilder::build);
         } catch (NoSuchAlgorithmException e) {
             throw new IllegalStateException("could not create the default ssl context", e);
         }

java-client/src/test/java/org/opensearch/client/transport/httpclient5/AutomaticRetriesTest.java

@@ -0,0 +1,160 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.client.transport.httpclient5;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hc.client5.http.impl.DefaultHttpRequestRetryStrategy;
+import org.apache.hc.core5.http.HttpHost;
+import org.apache.hc.core5.util.TimeValue;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.opensearch.client.opensearch.OpenSearchClient;
+import org.opensearch.client.opensearch.generic.Requests;
+import org.opensearch.client.opensearch.generic.Response;
+
+/**
+ * Tests that automatic retries can be enabled through
+ * {@link ApacheHttpClient5TransportBuilder#setAutomaticRetriesDisabled}, and that requests are
+ * not retried by default.
+ */
+public class AutomaticRetriesTest {
+    private ServerSocket serverSocket;
+    private ExecutorService serverExecutor;
+    private volatile boolean serverRunning;
+    private final AtomicInteger requestCount = new AtomicInteger();
+
+    @Before
+    public void startRawSocketServer() throws IOException {
+        serverSocket = new ServerSocket(0);
+        serverExecutor = Executors.newCachedThreadPool();
+        serverRunning = true;
+
+        serverExecutor.submit(() -> {
+            while (serverRunning) {
+                try {
+                    Socket clientSocket = serverSocket.accept();
+                    serverExecutor.submit(() -> handleClientRequest(clientSocket));
+                } catch (IOException e) {
+                    if (serverRunning) {
+                        e.printStackTrace();
+                    }
+                }
+            }
+        });
+    }
+
+    @After
+    public void stopRawSocketServer() throws IOException {
+        serverRunning = false;
+        if (serverSocket != null && !serverSocket.isClosed()) {
+            serverSocket.close();
+        }
+        serverExecutor.shutdownNow();
+    }
+
+    private void handleClientRequest(Socket clientSocket) {
+        try (InputStream in = clientSocket.getInputStream(); OutputStream out = clientSocket.getOutputStream()) {
+            readRequestHead(in);
+
+            String response;
+            if (requestCount.incrementAndGet() == 1) {
+                response = "HTTP/1.1 503 Service Unavailable\r\n" + "Content-Length: 0\r\n" + "Connection: close\r\n" + "\r\n";
+            } else {
+                String body = "{\"acknowledged\":true}";
+                response = "HTTP/1.1 200 OK\r\n"
+                    + "Content-Type: application/json\r\n"
+                    + "Content-Length: "
+                    + body.length()
+                    + "\r\n"
+                    + "Connection: close\r\n"
+                    + "\r\n"
+                    + body;
+            }
+            out.write(response.getBytes(StandardCharsets.UTF_8));
+            out.flush();
+        } catch (IOException e) {
+            System.err.println("Server: Error handling client connection: " + e.getMessage());
+        } finally {
+            try {
+                clientSocket.close();
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+    // Reads the request line and headers, up to the terminating blank line
+    private static void readRequestHead(InputStream in) throws IOException {
+        int c;
+        int newLines = 0;
+        while (newLines < 2 && (c = in.read()) != -1) {
+            if (c == '\n') {
+                newLines++;
+            } else if (c != '\r') {
+                newLines = 0;
+            }
+        }
+    }
+
+    @Test
+    public void testDefaultRetryStrategyRetriesFailedRequest() throws IOException {
+        OpenSearchClient client = createClient(builder().setAutomaticRetriesDisabled(false));
+
+        Response response = client.generic().execute(Requests.builder().method("GET").endpoint("/").build());
+
+        assertEquals(200, response.getStatus());
+        assertEquals(2, requestCount.get());
+    }
+
+    @Test
+    public void testCustomRetryStrategySetThroughCallback() throws IOException {
+        OpenSearchClient client = createClient(
+            builder().setAutomaticRetriesDisabled(false)
+                .setHttpClientConfigCallback(
+                    httpClientBuilder -> httpClientBuilder.setRetryStrategy(
+                        new DefaultHttpRequestRetryStrategy(3, TimeValue.ofMilliseconds(50))
+                    )
+                )
+        );
+
+        Response response = client.generic().execute(Requests.builder().method("GET").endpoint("/").build());
+
+        assertEquals(200, response.getStatus());
+        assertEquals(2, requestCount.get());
+    }
+
+    @Test
+    public void testNoRetriesByDefault() {
+        OpenSearchClient client = createClient(builder());
+
+        assertThrows(ResponseException.class, () -> client.generic().execute(Requests.builder().method("GET").endpoint("/").build()));
+
+        assertEquals(1, requestCount.get());
+    }
+
+    private ApacheHttpClient5TransportBuilder builder() {
+        return ApacheHttpClient5TransportBuilder.builder(new HttpHost("http", "localhost", serverSocket.getLocalPort()));
+    }
+
+    private static OpenSearchClient createClient(ApacheHttpClient5TransportBuilder builder) {
+        return new OpenSearchClient(builder.build());
+    }
+}