Used retry hint from CRT before converting HTTPException to IOException. (#3575)

zoewangg · JonathanHenson · web-flow · commit 46e49ba4bf94 · 2022-11-30T15:51:10.000-08:00
* Used retry hint from CRT before converting HTTPException to IOException. (#3540) * Used retry hint from CRT before converting HTTPException to IOException. * Add branch for handling http exceptions in CrtRequestExecutor. Updated test to use a mock http request that doesn't require a builder. * Fix checkstyle errors and test * Add more tests and fix existing tests Co-authored-by: Jonathan M. Henson <henso@amazon.com>
diff --git a/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/CrtRequestExecutor.java b/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/CrtRequestExecutor.java
@@ -28,6 +28,7 @@
 import software.amazon.awssdk.crt.CrtRuntimeException;
 import software.amazon.awssdk.crt.http.HttpClientConnection;
 import software.amazon.awssdk.crt.http.HttpClientConnectionManager;
+import software.amazon.awssdk.crt.http.HttpException;
 import software.amazon.awssdk.crt.http.HttpManagerMetrics;
 import software.amazon.awssdk.crt.http.HttpRequest;
 import software.amazon.awssdk.crt.http.HttpStreamResponseHandler;
@@ -84,18 +85,7 @@ public CompletableFuture<Void> execute(CrtRequestContext executionContext) {
                 metricCollector.reportMetric(CONCURRENCY_ACQUIRE_DURATION, acquireTimeTaken);
             }
 
-            HttpRequest crtRequest = CrtRequestAdapter.toCrtRequest(executionContext);
-            HttpStreamResponseHandler crtResponseHandler =
-                CrtResponseAdapter.toCrtResponseHandler(crtConn, requestFuture, asyncRequest.responseHandler());
-
-            // Submit the request on the connection
-            try {
-                crtConn.makeRequest(crtRequest, crtResponseHandler).activate();
-            } catch (IllegalStateException | CrtRuntimeException e) {
-                reportFailure(new IOException("An exception occurred when making the request", e),
-                              requestFuture,
-                              asyncRequest.responseHandler());
-            }
+            executeRequest(executionContext, requestFuture, crtConn, asyncRequest);
         });
 
         requestFuture.whenComplete((obj, err) -> {
@@ -113,6 +103,35 @@ public CompletableFuture<Void> execute(CrtRequestContext executionContext) {
         return requestFuture;
     }
 
+    private void executeRequest(CrtRequestContext executionContext,
+                                CompletableFuture<Void> requestFuture,
+                                HttpClientConnection crtConn,
+                                AsyncExecuteRequest asyncRequest) {
+        HttpRequest crtRequest = CrtRequestAdapter.toCrtRequest(executionContext);
+        HttpStreamResponseHandler crtResponseHandler =
+            CrtResponseAdapter.toCrtResponseHandler(crtConn, requestFuture, asyncRequest.responseHandler());
+
+        // Submit the request on the connection
+        try {
+            crtConn.makeRequest(crtRequest, crtResponseHandler).activate();
+        } catch (HttpException e) {
+            Throwable toThrow = e;
+            if (HttpClientConnection.isErrorRetryable(e)) {
+                // IOExceptions get retried, and if the CRT says this error is retryable,
+                // it's semantically an IOException anyway.
+                toThrow = new IOException(e);
+            }
+            reportFailure(toThrow,
+                          requestFuture,
+                          asyncRequest.responseHandler());
+        } catch (IllegalStateException | CrtRuntimeException e) {
+            // CRT throws IllegalStateException if the connection is closed
+            reportFailure(new IOException("An exception occurred when making the request", e),
+                          requestFuture,
+                          asyncRequest.responseHandler());
+        }
+    }
+
     /**
      * Create the execution future and set up the cancellation logic.
      * @return The created execution future.
@@ -137,7 +156,7 @@ private CompletableFuture<Void> createExecutionFuture(AsyncExecuteRequest reques
     /**
      * Notify the provided response handler and future of the failure.
      */
-    private void reportFailure(IOException cause,
+    private void reportFailure(Throwable cause,
                                CompletableFuture<Void> executeFuture,
                                SdkAsyncHttpResponseHandler responseHandler) {
         try {
diff --git a/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/response/CrtResponseAdapter.java b/http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/response/CrtResponseAdapter.java
@@ -129,9 +129,16 @@ private void onSuccessfulResponseComplete(HttpStream stream) {
     private void onFailedResponseComplete(HttpStream stream, HttpException error) {
         log.debug(() -> "HTTP response encountered an error.", error);
 
-        IOException wrappedError = new IOException(error);
-        responsePublisher.error(wrappedError);
-        failResponseHandlerAndFuture(stream, wrappedError);
+        Throwable toThrow = error;
+
+        if (HttpClientConnection.isErrorRetryable(error)) {
+            // IOExceptions get retried, and if the CRT says this error is retryable,
+            // it's semantically an IOException anyway.
+            toThrow = new IOException(error);
+        }
+
+        responsePublisher.error(toThrow);
+        failResponseHandlerAndFuture(stream, toThrow);
     }
 
     private void failResponseHandlerAndFuture(HttpStream stream, Throwable error) {
diff --git a/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/AwsCrtHttpClientSpiVerificationTest.java b/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/AwsCrtHttpClientSpiVerificationTest.java
@@ -35,6 +35,11 @@
 import java.net.URI;
 import java.nio.ByteBuffer;
 import java.time.Duration;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
 import java.util.Random;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -48,8 +53,6 @@
 import org.reactivestreams.Subscription;
 import software.amazon.awssdk.crt.CrtResource;
 import software.amazon.awssdk.crt.http.HttpException;
-import software.amazon.awssdk.crt.io.EventLoopGroup;
-import software.amazon.awssdk.crt.io.HostResolver;
 import software.amazon.awssdk.http.RecordingResponseHandler;
 import software.amazon.awssdk.http.SdkHttpMethod;
 import software.amazon.awssdk.http.SdkHttpRequest;
@@ -72,8 +75,6 @@ public class AwsCrtHttpClientSpiVerificationTest {
 
     @BeforeClass
     public static void setup() throws Exception {
-        CrtResource.waitForNoResources();
-
         client = AwsCrtAsyncHttpClient.builder()
                                       .connectionHealthChecksConfiguration(b -> b.minThroughputInBytesPerSecond(4068L)
                                                                                  .allowableThroughputFailureInterval(Duration.ofSeconds(3)))
@@ -83,8 +84,6 @@ public static void setup() throws Exception {
     @AfterClass
     public static void tearDown() {
         client.close();
-        EventLoopGroup.closeStaticDefault();
-        HostResolver.closeStaticDefault();
         CrtResource.waitForNoResources();
     }
 
@@ -132,6 +131,28 @@ public void requestFailed_connectionTimeout_shouldWrapException() {
         }
     }
 
+    @Test
+    public void requestFailed_notRetryable_shouldNotWrapException() {
+        try (SdkAsyncHttpClient client = AwsCrtAsyncHttpClient.builder().build()) {
+            URI uri = URI.create("http://localhost:" + mockServer.port());
+            // make it invalid by doing a non-zero content length with no request body...
+            Map<String, List<String>> headers = new HashMap<>();
+            headers.put("host", Collections.singletonList(uri.getHost()));
+
+            List<String> contentLengthValues = new LinkedList<>();
+            contentLengthValues.add("1");
+            headers.put("content-length", contentLengthValues);
+
+            SdkHttpRequest request = createRequest(uri).toBuilder().headers(headers).build();
+
+            RecordingResponseHandler recorder = new RecordingResponseHandler();
+            client.execute(AsyncExecuteRequest.builder().request(request).requestContentPublisher(new EmptyPublisher()).responseHandler(recorder).build());
+            // invalid request should have returned an HttpException and not an IOException.
+            assertThatThrownBy(() -> recorder.completeFuture().get(5, TimeUnit.SECONDS))
+                .hasCauseInstanceOf(HttpException.class).hasMessageContaining("does not match the previously declared length");
+        }
+    }
+
     @Test
     public void callsOnStreamForEmptyResponseContent() throws Exception {
         stubFor(any(urlEqualTo("/")).willReturn(aResponse().withStatus(204).withHeader("foo", "bar")));
diff --git a/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/AwsCrtHttpClientWireMockTest.java b/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/AwsCrtHttpClientWireMockTest.java
@@ -48,7 +48,7 @@ public class AwsCrtHttpClientWireMockTest {
 
     @BeforeClass
     public static void setup() {
-        System.setProperty("aws.crt.debugnative", "true");
+        System.setProperty("aws.crt.debugnative", "false");
     }
 
     @AfterClass
diff --git a/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/internal/CrtRequestExecutorTest.java b/http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/internal/CrtRequestExecutorTest.java
@@ -16,6 +16,7 @@
 package software.amazon.awssdk.http.crt.internal;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static software.amazon.awssdk.http.HttpTestUtils.createProvider;
 import static software.amazon.awssdk.http.crt.CrtHttpClientTestUtils.createRequest;
 
@@ -33,12 +34,14 @@
 import software.amazon.awssdk.crt.CrtRuntimeException;
 import software.amazon.awssdk.crt.http.HttpClientConnection;
 import software.amazon.awssdk.crt.http.HttpClientConnectionManager;
+import software.amazon.awssdk.crt.http.HttpException;
 import software.amazon.awssdk.crt.http.HttpRequest;
 import software.amazon.awssdk.http.SdkCancellationException;
 import software.amazon.awssdk.http.SdkHttpFullRequest;
 import software.amazon.awssdk.http.async.AsyncExecuteRequest;
 import software.amazon.awssdk.http.async.SdkAsyncHttpResponseHandler;
 import software.amazon.awssdk.http.crt.internal.response.CrtResponseAdapter;
+import software.amazon.awssdk.utils.CompletableFutureUtils;
 
 @RunWith(MockitoJUnitRunner.class)
 public class CrtRequestExecutorTest {
@@ -91,16 +94,7 @@ public void acquireConnectionThrowException_shouldInvokeOnError() {
     @Test
     public void makeRequestThrowException_shouldInvokeOnError() {
         CrtRuntimeException exception = new CrtRuntimeException("");
-        SdkHttpFullRequest request = createRequest(URI.create("http://localhost"));
-        CrtRequestContext context = CrtRequestContext.builder()
-                                                     .readBufferSize(2000)
-                                                     .crtConnPool(connectionManager)
-                                                     .request(AsyncExecuteRequest.builder()
-                                                                                 .request(request)
-                                                                                 .requestContentPublisher(createProvider(""))
-                                                                                 .responseHandler(responseHandler)
-                                                                                 .build())
-                                                     .build();
+        CrtRequestContext context = crtRequestContext();
         CompletableFuture<HttpClientConnection> completableFuture = new CompletableFuture<>();
 
         Mockito.when(connectionManager.acquireConnection()).thenReturn(completableFuture);
@@ -122,16 +116,7 @@ public void makeRequestThrowException_shouldInvokeOnError() {
 
     @Test
     public void makeRequest_success() {
-        SdkHttpFullRequest request = createRequest(URI.create("http://localhost"));
-        CrtRequestContext context = CrtRequestContext.builder()
-                                                     .readBufferSize(2000)
-                                                     .crtConnPool(connectionManager)
-                                                     .request(AsyncExecuteRequest.builder()
-                                                                                 .request(request)
-                                                                                 .requestContentPublisher(createProvider(""))
-                                                                                 .responseHandler(responseHandler)
-                                                                                 .build())
-                                                     .build();
+        CrtRequestContext context = crtRequestContext();
         CompletableFuture<HttpClientConnection> completableFuture = new CompletableFuture<>();
         Mockito.when(connectionManager.acquireConnection()).thenReturn(completableFuture);
         completableFuture.complete(httpClientConnection);
@@ -162,4 +147,97 @@ public void cancelRequest_shouldInvokeOnError() {
         assertThat(actualException).hasMessageContaining("The request was cancelled");
         assertThat(actualException).isInstanceOf(SdkCancellationException.class);
     }
+
+    @Test
+    public void execute_AcquireConnectionFailure_shouldAlwaysWrapIOException() {
+        CrtRequestContext context = crtRequestContext();
+        RuntimeException exception = new RuntimeException("some failure");
+        CompletableFuture<HttpClientConnection> completableFuture = CompletableFutureUtils.failedFuture(exception);
+
+        Mockito.when(connectionManager.acquireConnection()).thenReturn(completableFuture);
+
+        CompletableFuture<Void> executeFuture = requestExecutor.execute(context);
+        assertThatThrownBy(executeFuture::join).hasCauseInstanceOf(IOException.class).hasRootCause(exception);
+    }
+
+    @Test
+    public void executeRequest_failedOfIllegalStateException_shouldWrapIOException() {
+        IllegalStateException exception = new IllegalStateException("connection closed");
+        CrtRequestContext context = crtRequestContext();
+        CompletableFuture<HttpClientConnection> completableFuture = new CompletableFuture<>();
+
+        Mockito.when(connectionManager.acquireConnection()).thenReturn(completableFuture);
+        completableFuture.complete(httpClientConnection);
+
+        Mockito.when(httpClientConnection.makeRequest(Mockito.any(HttpRequest.class), Mockito.any(CrtResponseAdapter.class)))
+               .thenThrow(exception);
+
+        CompletableFuture<Void> executeFuture = requestExecutor.execute(context);
+
+        ArgumentCaptor<Exception> argumentCaptor = ArgumentCaptor.forClass(Exception.class);
+        Mockito.verify(responseHandler).onError(argumentCaptor.capture());
+
+        Exception actualException = argumentCaptor.getValue();
+        assertThat(actualException).hasMessageContaining("An exception occurred when making the request").hasCause(exception);
+        assertThatThrownBy(executeFuture::join).hasCauseInstanceOf(IOException.class).hasRootCause(exception);
+    }
+
+    @Test
+    public void executeRequest_failedOfRetryableHttpException_shouldWrapIOException() {
+        HttpException exception = new HttpException(0x080a); // AWS_ERROR_HTTP_CONNECTION_CLOSED
+        CrtRequestContext context = crtRequestContext();
+        CompletableFuture<HttpClientConnection> completableFuture = new CompletableFuture<>();
+
+        Mockito.when(connectionManager.acquireConnection()).thenReturn(completableFuture);
+        completableFuture.complete(httpClientConnection);
+
+        Mockito.when(httpClientConnection.makeRequest(Mockito.any(HttpRequest.class), Mockito.any(CrtResponseAdapter.class)))
+               .thenThrow(exception);
+
+        CompletableFuture<Void> executeFuture = requestExecutor.execute(context);
+
+        ArgumentCaptor<Exception> argumentCaptor = ArgumentCaptor.forClass(Exception.class);
+        Mockito.verify(responseHandler).onError(argumentCaptor.capture());
+
+        Exception actualException = argumentCaptor.getValue();
+        assertThat(actualException).hasCause(exception);
+        assertThatThrownBy(executeFuture::join).hasCauseInstanceOf(IOException.class).hasRootCause(exception);
+    }
+
+
+    @Test
+    public void executeRequest_failedOfNonRetryableHttpException_shouldNotWrapIOException() {
+        HttpException exception = new HttpException(0x0801); // AWS_ERROR_HTTP_HEADER_NOT_FOUND
+        CrtRequestContext context = crtRequestContext();
+        CompletableFuture<HttpClientConnection> completableFuture = new CompletableFuture<>();
+
+        Mockito.when(connectionManager.acquireConnection()).thenReturn(completableFuture);
+        completableFuture.complete(httpClientConnection);
+
+        Mockito.when(httpClientConnection.makeRequest(Mockito.any(HttpRequest.class), Mockito.any(CrtResponseAdapter.class)))
+               .thenThrow(exception);
+
+        CompletableFuture<Void> executeFuture = requestExecutor.execute(context);
+
+        ArgumentCaptor<Exception> argumentCaptor = ArgumentCaptor.forClass(Exception.class);
+        Mockito.verify(responseHandler).onError(argumentCaptor.capture());
+
+        Exception actualException = argumentCaptor.getValue();
+        assertThat(actualException).isEqualTo(exception);
+        assertThatThrownBy(executeFuture::join).hasCause(exception);
+    }
+
+    private CrtRequestContext crtRequestContext() {
+        SdkHttpFullRequest request = createRequest(URI.create("http://localhost"));
+        return CrtRequestContext.builder()
+                                .readBufferSize(2000)
+                                .crtConnPool(connectionManager)
+                                .request(AsyncExecuteRequest.builder()
+                                                            .request(request)
+                                                            .requestContentPublisher(createProvider(""))
+                                                            .responseHandler(responseHandler)
+                                                            .build())
+                                .build();
+    }
+
 }
diff --git a/pom.xml b/pom.xml
@@ -114,7 +114,7 @@
         <rxjava.version>2.2.21</rxjava.version>
         <commons-codec.verion>1.10</commons-codec.verion>
         <jmh.version>1.29</jmh.version>
-        <awscrt.version>0.19.6</awscrt.version>
+        <awscrt.version>0.19.9</awscrt.version>
 
         <!--Test dependencies -->
         <junit5.version>5.8.1</junit5.version>

Original file line number	Diff line number	Diff line change
`@@ -48,7 +48,7 @@ public class AwsCrtHttpClientWireMockTest {`
`48`	`48`
`49`	`49`	`@BeforeClass`
`50`	`50`	`public static void setup() {`
`51`		`- System.setProperty("aws.crt.debugnative", "true");`
	`51`	`+ System.setProperty("aws.crt.debugnative", "false");`
`52`	`52`	`}`
`53`	`53`
`54`	`54`	`@AfterClass`