[FEATURE] Raise errors for HTTP error codes in the generic client

Signed-off-by: Andriy Redko <[email protected]>

guides/generic.md

@@ -13,23 +13,29 @@ There are rare circumstances when the typed OpenSearch client APIs are too const
 The following sample code gets the `OpenSearchGenericClient` from the `OpenSearchClient` instance.
 
 ```java
-final OpenSearchGenericClient generic = javaClient().generic();
+final OpenSearchGenericClient generic = javaClient().generic(ClientOptions.DEFAULT);
+```
+
+The generic client with default options (`ClientOptions.DEFAULT`) returns the responses as those were received from the server. The generic client could be instructed to raise an `OpenSearchClientException` exception instead if the HTTP status code is not indicating the successful response, for example:
+
+```java
+final OpenSearchGenericClient generic = javaClient().generic(ClientOptions.throwOnHttpErrors());
 ```
 
 ## Sending Simple Request
 The following sample code sends a simple request that does not require any payload to be provided (typically, `GET` requests).
 
 ```java
 // compare with what the low level client outputs
-try (Response response = javaClient().generic().execute(Requests.builder().endpoint("/").method("GET").build())) {
+try (Response response = javaClient().generic(ClientOptions.DEFAULT).execute(Requests.builder().endpoint("/").method("GET").build())) {
    // ...
 }
 ```
 
 Please notice that the `Response` instance should be closed explicitly in order to free up any allocated resource (like response input streams or buffers), the [`try-with-resource`](https://docs.oracle.com/javase/tutorial/essential/exceptions/tryResourceClose.html) pattern is encouraged.
 
 ```java
-try (Response response = javaClient().generic().execute(...)) {
+try (Response response = javaClient().generic(ClientOptions.DEFAULT).execute(...)) {
    // ...
 }
 ```
@@ -38,7 +44,7 @@ The generic client never interprets status codes and provides the direct access
 
 ```java
 // compare with what the low level client outputs
-try (Response response = javaClient().generic().execute(...)) {
+try (Response response = javaClient().generic(ClientOptions.DEFAULT).execute(...)) {
    if (response.getStatus() != 200) {
       // Request was not successful
    }
@@ -49,7 +55,7 @@ try (Response response = javaClient().generic().execute(...)) {
 The following sample code a simple request with JSON body.
 
 ```java
- try (Response response = javaClient().generic()
+ try (Response response = javaClient().generic(ClientOptions.DEFAULT)
         .execute(
             Requests.builder()
                 .endpoint("/" + index + "/_search")
@@ -83,7 +89,7 @@ final CreateIndexRequest request = CreateIndexRequest.of(
     .settings(settings -> settings.sort(s -> s.field("name").order(SegmentSortOrder.Asc)))
 );
 
-try (Response response = javaClient().generic()
+try (Response response = javaClient().generic(ClientOptions.DEFAULT)
       .execute(
           Requests.builder()
             .endpoint("/" + index).method("PUT")
@@ -100,7 +106,7 @@ try (Response response = javaClient().generic()
 Dealing with strings or POJOs could be daunting sometimes, using structured JSON APIs is a middle ground of both approaches, as per following sample code that uses (`jakarta.json.Json`)[https://jakarta.ee/specifications/jsonp].
 
 ```java
-try (Response response = javaClient().generic()
+try (Response response = javaClient().generic(ClientOptions.DEFAULT)
       .execute(
         Requests.builder()
           .endpoint("/" + index)

java-client/src/main/java/org/opensearch/client/opensearch/OpenSearchClient.java

@@ -156,8 +156,8 @@ public OpenSearchClient withTransportOptions(@Nullable TransportOptions transpor
     }
 
     // ----- Child clients
-    public OpenSearchGenericClient generic() {
-        return new OpenSearchGenericClient(this.transport, this.transportOptions);
+    public OpenSearchGenericClient generic(OpenSearchGenericClient.ClientOptions clientOptions) {
+        return new OpenSearchGenericClient(this.transport, this.transportOptions, clientOptions);
     }
 
     public OpenSearchCatClient cat() {

java-client/src/main/java/org/opensearch/client/opensearch/generic/Body.java

@@ -67,6 +67,14 @@ public interface Body extends AutoCloseable {
      * @return body as {@link String}
      */
     default String bodyAsString() {
+        return new String(bodyAsBytes(), StandardCharsets.UTF_8);
+    }
+
+    /**
+     * Gets the body as {@link byte[]}
+     * @return body as {@link byte[]}
+     */
+    default byte[] bodyAsBytes() {
         try (final ByteArrayOutputStream out = new ByteArrayOutputStream()) {
             try (final InputStream in = body()) {
                 final byte[] buffer = new byte[DEFAULT_BUFFER_SIZE];
@@ -77,7 +85,7 @@ default String bodyAsString() {
             }
 
             out.flush();
-            return new String(out.toByteArray(), StandardCharsets.UTF_8);
+            return out.toByteArray();
         } catch (final IOException ex) {
             throw new UncheckedIOException(ex);
         }

java-client/src/main/java/org/opensearch/client/opensearch/generic/GenericResponse.java

@@ -27,6 +27,10 @@ final class GenericResponse implements Response {
     private final Collection<Map.Entry<String, String>> headers;
     private final Body body;
 
+    GenericResponse(String uri, String protocol, String method, int status, String reason, Collection<Map.Entry<String, String>> headers) {
+        this(uri, protocol, method, status, reason, headers, null);
+    }
+
     GenericResponse(
         String uri,
         String protocol,

java-client/src/main/java/org/opensearch/client/opensearch/generic/OpenSearchClientException.java

@@ -0,0 +1,41 @@
+/*
+ * 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.opensearch.generic;
+
+/**
+ * Exception thrown by API client methods when OpenSearch could not accept or
+ * process a request.
+ * <p>
+ * The {@link #response()} contains the the raw response as returned by the API
+ * endpoint that was called.
+ */
+public class OpenSearchClientException extends RuntimeException {
+
+    private final Response response;
+
+    public OpenSearchClientException(Response response) {
+        super("Request failed: [" + response.getStatus() + "] " + response.getReason());
+        this.response = response;
+    }
+
+    /**
+     * The error response sent by OpenSearch
+     */
+    public Response response() {
+        return this.response;
+    }
+
+    /**
+     * Status code returned by OpenSearch. Shortcut for
+     * {@code response().status()}.
+     */
+    public int status() {
+        return this.response.getStatus();
+    }
+}

java-client/src/main/java/org/opensearch/client/opensearch/generic/OpenSearchGenericClient.java

@@ -10,10 +10,12 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.UncheckedIOException;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 import org.opensearch.client.ApiClient;
@@ -24,14 +26,37 @@
  * Client for the generic HTTP requests.
  */
 public class OpenSearchGenericClient extends ApiClient<OpenSearchTransport, OpenSearchGenericClient> {
+    /**
+     * Generic client options
+     */
+    public static final class ClientOptions {
+        public static final ClientOptions DEFAULT = new ClientOptions();
+
+        private final Predicate<Integer> error;
+
+        private ClientOptions() {
+            this(statusCode -> false);
+        }
+
+        private ClientOptions(final Predicate<Integer> error) {
+            this.error = error;
+        }
+
+        public static ClientOptions throwOnHttpErrors() {
+            return new ClientOptions(statusCode -> statusCode >= 400);
+        }
+    }
+
     /**
      * Generic endpoint instance
      */
     private static final class GenericEndpoint implements org.opensearch.client.transport.GenericEndpoint<Request, Response> {
         private final Request request;
+        private final Predicate<Integer> error;
 
-        public GenericEndpoint(Request request) {
+        public GenericEndpoint(Request request, Predicate<Integer> error) {
             this.request = request;
+            this.error = error;
         }
 
         @Override
@@ -67,24 +92,62 @@ public GenericResponse responseDeserializer(
             int status,
             String reason,
             List<Entry<String, String>> headers,
-            String contentType,
-            InputStream body
+            @Nullable String contentType,
+            @Nullable InputStream body
         ) {
-            return new GenericResponse(uri, protocol, method, status, reason, headers, Body.from(body, contentType));
+            if (isError(status)) {
+                // Fully consume the response body since the it will be propagated as an exception with possible no chance to be closed
+                try (Body b = Body.from(body, contentType)) {
+                    if (b != null) {
+                        return new GenericResponse(
+                            uri,
+                            protocol,
+                            method,
+                            status,
+                            reason,
+                            headers,
+                            Body.from(b.bodyAsBytes(), b.contentType())
+                        );
+                    } else {
+                        return new GenericResponse(uri, protocol, method, status, reason, headers);
+                    }
+                } catch (final IOException ex) {
+                    throw new UncheckedIOException(ex);
+                }
+            } else {
+                return new GenericResponse(uri, protocol, method, status, reason, headers, Body.from(body, contentType));
+            }
+        }
+
+        @Override
+        public boolean isError(int statusCode) {
+            return error.test(statusCode);
+        }
+
+        @Override
+        public <T extends RuntimeException> T exceptionConverter(Response error) {
+            throw new OpenSearchClientException(error);
         }
     }
 
+    private final ClientOptions clientOptions;
+
     public OpenSearchGenericClient(OpenSearchTransport transport) {
-        super(transport, null);
+        this(transport, null, ClientOptions.DEFAULT);
     }
 
-    public OpenSearchGenericClient(OpenSearchTransport transport, @Nullable TransportOptions transportOptions) {
+    public OpenSearchGenericClient(
+        OpenSearchTransport transport,
+        @Nullable TransportOptions transportOptions,
+        ClientOptions clientOptions
+    ) {
         super(transport, transportOptions);
+        this.clientOptions = clientOptions;
     }
 
     @Override
     public OpenSearchGenericClient withTransportOptions(@Nullable TransportOptions transportOptions) {
-        return new OpenSearchGenericClient(this.transport, transportOptions);
+        return new OpenSearchGenericClient(this.transport, transportOptions, this.clientOptions);
     }
 
     /**
@@ -94,7 +157,7 @@ public OpenSearchGenericClient withTransportOptions(@Nullable TransportOptions t
      * @throws IOException I/O exception
      */
     public Response execute(Request request) throws IOException {
-        return transport.performRequest(request, new GenericEndpoint(request), this.transportOptions);
+        return transport.performRequest(request, new GenericEndpoint(request, clientOptions.error), this.transportOptions);
     }
 
     /**
@@ -103,6 +166,6 @@ public Response execute(Request request) throws IOException {
      * @return generic HTTP response future
      */
     public CompletableFuture<Response> executeAsync(Request request) {
-        return transport.performRequestAsync(request, new GenericEndpoint(request), this.transportOptions);
+        return transport.performRequestAsync(request, new GenericEndpoint(request, clientOptions.error), this.transportOptions);
     }
 }

java-client/src/main/java/org/opensearch/client/transport/Endpoint.java

@@ -37,6 +37,8 @@
 import javax.annotation.Nullable;
 import org.opensearch.client.json.JsonpDeserializer;
 import org.opensearch.client.json.NdJsonpSerializable;
+import org.opensearch.client.opensearch._types.ErrorResponse;
+import org.opensearch.client.opensearch._types.OpenSearchException;
 
 /**
  * An endpoint links requests and responses to HTTP protocol encoding. It also defines the error response
@@ -90,4 +92,13 @@ default Map<String, String> headers(RequestT request) {
     @Nullable
     JsonpDeserializer<ErrorT> errorDeserializer(int statusCode);
 
+    /**
+     * Converts error response to exception instance of type {@code T}
+     * @param <T> exception type
+     * @param error error response
+     * @return exception instance
+     */
+    default <T extends RuntimeException> T exceptionConverter(ErrorT error) {
+        throw new OpenSearchException((ErrorResponse) error);
+    }
 }