From 54d4393a6b20587f5a258c3148d393f0d1429807 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Wed, 17 Jul 2024 03:19:58 +0000 Subject: [PATCH] Fix for AwsSdk2Transport error handling (#1068) (#1072) Signed-off-by: Wesley Workman (cherry picked from commit 5ec233d2547fce86f9b290aed8ec5404ee7f0f97) Signed-off-by: github-actions[bot] --- CHANGELOG.md | 1 + .../transport/aws/AwsSdk2Transport.java | 122 +++++++++++------- .../integTest/aws/AwsSdk2GetRequestIT.java | 57 ++++++++ .../integTest/aws/AwsSdk2SearchIT.java | 17 --- .../aws/AwsSdk2TransportTestCase.java | 18 +++ 5 files changed, 151 insertions(+), 64 deletions(-) create mode 100644 java-client/src/test/java/org/opensearch/client/opensearch/integTest/aws/AwsSdk2GetRequestIT.java diff --git a/CHANGELOG.md b/CHANGELOG.md index b0782ad0c4..abcc1428d6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) ### Fixed - Deserialize aggregation containing a parent aggregation ([#706](https://github.com/opensearch-project/opensearch-java/pull/706)) +- Deserialize error response when using AwsSdk2Transport ([#1068](https://github.com/opensearch-project/opensearch-java/pull/1068)) ### Security diff --git a/java-client/src/main/java/org/opensearch/client/transport/aws/AwsSdk2Transport.java b/java-client/src/main/java/org/opensearch/client/transport/aws/AwsSdk2Transport.java index ad6718ec20..310d936e4a 100644 --- a/java-client/src/main/java/org/opensearch/client/transport/aws/AwsSdk2Transport.java +++ b/java-client/src/main/java/org/opensearch/client/transport/aws/AwsSdk2Transport.java @@ -48,6 +48,7 @@ import org.opensearch.client.transport.TransportOptions; import org.opensearch.client.transport.endpoints.BooleanEndpoint; import org.opensearch.client.transport.endpoints.BooleanResponse; +import org.opensearch.client.util.MissingRequiredPropertyException; import org.opensearch.client.util.OpenSearchRequestBodyBuffer; import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider; @@ -64,6 +65,7 @@ import software.amazon.awssdk.http.async.AsyncExecuteRequest; import software.amazon.awssdk.http.async.SdkAsyncHttpClient; import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.utils.IoUtils; import software.amazon.awssdk.utils.SdkAutoCloseable; /** @@ -534,10 +536,17 @@ private ResponseT parseResponse( if (errorDeserializer == null || bodyStream == null) { throw new TransportException("Request failed with status code '" + statusCode + "'"); } + + // We may have to reset if there is a parse deserialization exception + bodyStream = toByteArrayInputStream(bodyStream); + try { try (JsonParser parser = mapper.jsonProvider().createParser(bodyStream)) { ErrorT error = errorDeserializer.deserialize(parser, mapper); throw new OpenSearchException((ErrorResponse) error); + } catch (MissingRequiredPropertyException errorEx) { + bodyStream.reset(); + return decodeResponse(uri, method, protocol, httpResponse, bodyStream, endpoint, mapper); } } catch (OpenSearchException e) { throw e; @@ -551,57 +560,68 @@ private ResponseT parseResponse( } } } else { - if (endpoint instanceof BooleanEndpoint) { - BooleanEndpoint bep = (BooleanEndpoint) endpoint; - @SuppressWarnings("unchecked") - ResponseT response = (ResponseT) new BooleanResponse(bep.getResult(statusCode)); - return response; - } else if (endpoint instanceof JsonEndpoint) { - JsonEndpoint jsonEndpoint = (JsonEndpoint) endpoint; - // Successful response - ResponseT response = null; - JsonpDeserializer responseParser = jsonEndpoint.responseDeserializer(); - if (responseParser != null) { - // Expecting a body - if (bodyStream == null) { - throw new TransportException("Expecting a response body, but none was sent"); - } - try (JsonParser parser = mapper.jsonProvider().createParser(bodyStream)) { - try { - response = responseParser.deserialize(parser, mapper); - } catch (NullPointerException e) { - response = responseParser.deserialize(parser, mapper); - } - } - ; - } - return response; - } else if (endpoint instanceof GenericEndpoint) { - @SuppressWarnings("unchecked") - final GenericEndpoint rawEndpoint = (GenericEndpoint) endpoint; + return decodeResponse(uri, method, protocol, httpResponse, bodyStream, endpoint, mapper); + } + } - String contentType = null; - if (bodyStream != null) { - contentType = httpResponse.firstMatchingHeader(Header.CONTENT_TYPE).orElse(null); + private ResponseT decodeResponse( + URI uri, + @Nonnull SdkHttpMethod method, + String protocol, + @Nonnull SdkHttpResponse httpResponse, + @CheckForNull InputStream bodyStream, + @Nonnull Endpoint endpoint, + JsonpMapper mapper + ) throws IOException { + if (endpoint instanceof BooleanEndpoint) { + BooleanEndpoint bep = (BooleanEndpoint) endpoint; + @SuppressWarnings("unchecked") + ResponseT response = (ResponseT) new BooleanResponse(bep.getResult(httpResponse.statusCode())); + return response; + } else if (endpoint instanceof JsonEndpoint) { + JsonEndpoint jsonEndpoint = (JsonEndpoint) endpoint; + // Successful response + ResponseT response = null; + JsonpDeserializer responseParser = jsonEndpoint.responseDeserializer(); + if (responseParser != null) { + // Expecting a body + if (bodyStream == null) { + throw new TransportException("Expecting a response body, but none was sent"); + } + try (JsonParser parser = mapper.jsonProvider().createParser(bodyStream)) { + try { + response = responseParser.deserialize(parser, mapper); + } catch (NullPointerException e) { + response = responseParser.deserialize(parser, mapper); + } } + } + return response; + } else if (endpoint instanceof GenericEndpoint) { + @SuppressWarnings("unchecked") + final GenericEndpoint rawEndpoint = (GenericEndpoint) endpoint; - return rawEndpoint.responseDeserializer( - uri.toString(), - method.name(), - protocol, - httpResponse.statusCode(), - httpResponse.statusText().orElse(null), - httpResponse.headers() - .entrySet() - .stream() - .map(h -> new AbstractMap.SimpleEntry(h.getKey(), Objects.toString(h.getValue()))) - .collect(Collectors.toList()), - contentType, - bodyStream - ); - } else { - throw new TransportException("Unhandled endpoint type: '" + endpoint.getClass().getName() + "'"); + String contentType = null; + if (bodyStream != null) { + contentType = httpResponse.firstMatchingHeader(Header.CONTENT_TYPE).orElse(null); } + + return rawEndpoint.responseDeserializer( + uri.toString(), + method.name(), + protocol, + httpResponse.statusCode(), + httpResponse.statusText().orElse(null), + httpResponse.headers() + .entrySet() + .stream() + .map(h -> new AbstractMap.SimpleEntry(h.getKey(), Objects.toString(h.getValue()))) + .collect(Collectors.toList()), + contentType, + bodyStream + ); + } else { + throw new TransportException("Unhandled endpoint type: '" + endpoint.getClass().getName() + "'"); } } @@ -617,6 +637,14 @@ private static Optional or(Optional opt, Supplier doc1Response = client.get(doc1Request, SimplePojo.class); + Assert.assertTrue(doc1Response.found()); + + GetRequest doc2Request = new GetRequest.Builder().index(TEST_INDEX).id("does-not-exist").build(); + GetResponse doc2Response = client.get(doc2Request, SimplePojo.class); + Assert.assertFalse(doc2Response.found()); + } + + @Test + public void testAsyncGetRequest() throws Exception { + resetTestIndex(false); + final OpenSearchAsyncClient client = getAsyncClient(false, null, null); + + SimplePojo doc1 = new SimplePojo("Document 1", "The text of document 1"); + addDoc(client, "id1", doc1).join(); + + Thread.sleep(1000); + + GetRequest doc1Request = new GetRequest.Builder().index(TEST_INDEX).id("id1").build(); + CompletableFuture> doc1ResponseFuture = client.get(doc1Request, SimplePojo.class); + + GetRequest doc2Request = new GetRequest.Builder().index(TEST_INDEX).id("does-not-exist").build(); + CompletableFuture> doc2ResponseFuture = client.get(doc2Request, SimplePojo.class); + + Assert.assertTrue(doc1ResponseFuture.join().found()); + Assert.assertFalse(doc2ResponseFuture.join().found()); + } +} diff --git a/java-client/src/test/java/org/opensearch/client/opensearch/integTest/aws/AwsSdk2SearchIT.java b/java-client/src/test/java/org/opensearch/client/opensearch/integTest/aws/AwsSdk2SearchIT.java index 7ec5b1f654..e563600fed 100644 --- a/java-client/src/test/java/org/opensearch/client/opensearch/integTest/aws/AwsSdk2SearchIT.java +++ b/java-client/src/test/java/org/opensearch/client/opensearch/integTest/aws/AwsSdk2SearchIT.java @@ -16,7 +16,6 @@ import org.opensearch.client.opensearch.OpenSearchAsyncClient; import org.opensearch.client.opensearch.OpenSearchClient; import org.opensearch.client.opensearch._types.OpenSearchException; -import org.opensearch.client.opensearch.core.IndexRequest; import org.opensearch.client.opensearch.core.IndexResponse; import org.opensearch.client.opensearch.core.SearchResponse; import org.opensearch.client.opensearch.indices.CreateIndexRequest; @@ -105,22 +104,6 @@ void testClientAsync(boolean async) throws Exception { Assert.assertEquals(doc1, response.hits().hits().get(0).source()); } - private void addDoc(OpenSearchClient client, String id, SimplePojo doc) throws Exception { - IndexRequest.Builder req = new IndexRequest.Builder().index(TEST_INDEX).document(doc).id(id); - client.index(req.build()); - } - - private CompletableFuture addDoc(OpenSearchAsyncClient client, String id, SimplePojo doc) { - IndexRequest.Builder req = new IndexRequest.Builder().index(TEST_INDEX).document(doc).id(id); - try { - return client.index(req.build()); - } catch (Exception e) { - final CompletableFuture failed = new CompletableFuture<>(); - failed.completeExceptionally(e); - return failed; - } - } - @Test public void testDoubleWrappedException() throws Exception { // ensure the test index exists diff --git a/java-client/src/test/java/org/opensearch/client/opensearch/integTest/aws/AwsSdk2TransportTestCase.java b/java-client/src/test/java/org/opensearch/client/opensearch/integTest/aws/AwsSdk2TransportTestCase.java index 79c4985b81..a7add62867 100644 --- a/java-client/src/test/java/org/opensearch/client/opensearch/integTest/aws/AwsSdk2TransportTestCase.java +++ b/java-client/src/test/java/org/opensearch/client/opensearch/integTest/aws/AwsSdk2TransportTestCase.java @@ -26,6 +26,8 @@ import org.opensearch.client.opensearch._types.SortOptions; import org.opensearch.client.opensearch._types.SortOrder; import org.opensearch.client.opensearch._types.query_dsl.Query; +import org.opensearch.client.opensearch.core.IndexRequest; +import org.opensearch.client.opensearch.core.IndexResponse; import org.opensearch.client.opensearch.core.SearchRequest; import org.opensearch.client.opensearch.core.SearchResponse; import org.opensearch.client.opensearch.indices.CreateIndexRequest; @@ -215,6 +217,22 @@ public void resetTestIndex(boolean async) throws Exception { client.create(req.build()); } + protected void addDoc(OpenSearchClient client, String id, SimplePojo doc) throws Exception { + IndexRequest.Builder req = new IndexRequest.Builder().index(TEST_INDEX).document(doc).id(id); + client.index(req.build()); + } + + protected CompletableFuture addDoc(OpenSearchAsyncClient client, String id, SimplePojo doc) { + IndexRequest.Builder req = new IndexRequest.Builder().index(TEST_INDEX).document(doc).id(id); + try { + return client.index(req.build()); + } catch (Exception e) { + final CompletableFuture failed = new CompletableFuture<>(); + failed.completeExceptionally(e); + return failed; + } + } + protected SearchResponse query(OpenSearchClient client, String title, String text) throws Exception { final Query query = Query.of(qb -> { if (title != null) {