Updating SDK to work with most recent CRT

Factoring out EventLoopGroup/HostResolver from client constructor, fixing bug reported by Sonar

Author: rccarper

http-clients/aws-crt-client/pom.xml

@@ -21,7 +21,7 @@
     <parent>
         <artifactId>http-clients</artifactId>
         <groupId>software.amazon.awssdk</groupId>
-        <version>2.10.9-SNAPSHOT</version>
+        <version>2.10.55-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
@@ -33,7 +33,7 @@
         <dependency>
             <groupId>software.amazon.awssdk.crt</groupId>
             <artifactId>aws-crt</artifactId>
-            <version>0.4.1</version>
+            <version>0.4.19</version>
         </dependency>
 
         <!--SDK dependencies-->

http-clients/aws-crt-client/src/it/java/software/amazon/awssdk/http/crt/AwsCrtClientCallingPatternIntegrationTest.java

@@ -32,6 +32,8 @@
 import org.junit.experimental.theories.Theory;
 import org.junit.runner.RunWith;
 import software.amazon.awssdk.crt.CrtResource;
+import software.amazon.awssdk.crt.io.EventLoopGroup;
+import software.amazon.awssdk.crt.io.HostResolver;
 import software.amazon.awssdk.http.SdkHttpConfigurationOption;
 import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
 import software.amazon.awssdk.regions.Region;
@@ -95,8 +97,12 @@ private boolean testWithClient(KmsAsyncClient asyncKMSClient, int numberOfReques
     }
 
     private boolean testWithNewClient(int eventLoopSize, int numberOfRequests) {
-        try (SdkAsyncHttpClient newAwsCrtHttpClient = AwsCrtAsyncHttpClient.builder()
-                .eventLoopSize(eventLoopSize)
+
+        try (EventLoopGroup eventLoopGroup = new EventLoopGroup(eventLoopSize);
+             HostResolver hostResolver = new HostResolver(eventLoopGroup);
+             SdkAsyncHttpClient newAwsCrtHttpClient = AwsCrtAsyncHttpClient.builder()
+                .eventLoopGroup(eventLoopGroup)
+                .hostResolver(hostResolver)
                 .build()) {
             try (KmsAsyncClient newAsyncKMSClient = KmsAsyncClient.builder()
                     .region(REGION)
@@ -156,9 +162,12 @@ public void checkAllCombinations(@FromDataPoints("EventLoop") int eventLoopSize,
                     .put(SdkHttpConfigurationOption.MAX_CONNECTIONS, connectionPoolSize)
                     .build();
 
+            EventLoopGroup eventLoopGroup = new EventLoopGroup(eventLoopSize);
+            HostResolver hostResolver = new HostResolver(eventLoopGroup);
 
             SdkAsyncHttpClient awsCrtHttpClient = AwsCrtAsyncHttpClient.builder()
-                    .eventLoopSize(eventLoopSize)
+                    .eventLoopGroup(eventLoopGroup)
+                    .hostResolver(hostResolver)
                     .buildWithDefaults(attributes);
 
             KmsAsyncClient sharedAsyncKMSClient = KmsAsyncClient.builder()
@@ -194,6 +203,9 @@ public void checkAllCombinations(@FromDataPoints("EventLoop") int eventLoopSize,
             awsCrtHttpClient.close();
             Assert.assertFalse(failed.get());
 
+            hostResolver.close();
+            eventLoopGroup.close();
+
             CrtResource.waitForNoResources();
 
             float numSeconds = (float) ((System.currentTimeMillis() - start) / 1000.0);

http-clients/aws-crt-client/src/it/java/software/amazon/awssdk/http/crt/AwsCrtClientKmsIntegrationTest.java

@@ -11,6 +11,8 @@
 import org.junit.Test;
 import software.amazon.awssdk.core.SdkBytes;
 import software.amazon.awssdk.crt.CrtResource;
+import software.amazon.awssdk.crt.io.EventLoopGroup;
+import software.amazon.awssdk.crt.io.HostResolver;
 import software.amazon.awssdk.crt.io.TlsCipherPreference;
 import software.amazon.awssdk.crt.io.TlsContextOptions;
 import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
@@ -32,6 +34,8 @@ public class AwsCrtClientKmsIntegrationTest {
     private static String KEY_ALIAS = "alias/aws-sdk-java-v2-integ-test";
     private static Region REGION = Region.US_EAST_1;
     private static List<SdkAsyncHttpClient> awsCrtHttpClients = new ArrayList<>();
+    private static EventLoopGroup eventLoopGroup;
+    private static HostResolver hostResolver;
 
     @Before
     public void setup() {
@@ -43,9 +47,13 @@ public void setup() {
                 continue;
             }
 
+            int numThreads = 1;
+            eventLoopGroup = new EventLoopGroup(numThreads);
+            hostResolver = new HostResolver(eventLoopGroup);
 
             SdkAsyncHttpClient awsCrtHttpClient = AwsCrtAsyncHttpClient.builder()
-                    .eventLoopSize(1)
+                    .eventLoopGroup(eventLoopGroup)
+                    .hostResolver(hostResolver)
                     .build();
 
             awsCrtHttpClients.add(awsCrtHttpClient);
@@ -55,6 +63,8 @@ public void setup() {
 
     @After
     public void tearDown() {
+        hostResolver.close();
+        eventLoopGroup.close();
         CrtResource.waitForNoResources();
     }
 

http-clients/aws-crt-client/src/it/java/software/amazon/awssdk/http/crt/AwsCrtClientS3IntegrationTest.java

@@ -30,6 +30,8 @@
 import software.amazon.awssdk.core.ResponseBytes;
 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
 import software.amazon.awssdk.crt.CrtResource;
+import software.amazon.awssdk.crt.io.EventLoopGroup;
+import software.amazon.awssdk.crt.io.HostResolver;
 import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
@@ -50,6 +52,8 @@ public class AwsCrtClientS3IntegrationTest {
 
     private static Region REGION = Region.US_EAST_1;
 
+    private static EventLoopGroup eventLoopGroup;
+    private static HostResolver hostResolver;
     private static SdkAsyncHttpClient crtClient;
 
     private static S3AsyncClient s3;
@@ -58,8 +62,13 @@ public class AwsCrtClientS3IntegrationTest {
     public void setup() {
         CrtResource.waitForNoResources();
 
+        int numThreads = 4;
+        eventLoopGroup = new EventLoopGroup(numThreads);
+        hostResolver = new HostResolver(eventLoopGroup);
+
         crtClient = AwsCrtAsyncHttpClient.builder()
-                .eventLoopSize(4)
+                .eventLoopGroup(eventLoopGroup)
+                .hostResolver(hostResolver)
                 .build();
 
         s3 = S3AsyncClient.builder()
@@ -73,7 +82,8 @@ public void setup() {
     public void tearDown() {
         s3.close();
         crtClient.close();
-
+        hostResolver.close();
+        eventLoopGroup.close();
         CrtResource.waitForNoResources();
     }
 
@@ -108,4 +118,4 @@ public void testParallelDownloadFromS3() throws Exception {
         }
     }
 
-}
+}

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtAsyncHttpClient.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * Copyright 2010-2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License").
  * You may not use this file except in compliance with the License.
@@ -30,9 +30,12 @@
 import software.amazon.awssdk.annotations.SdkPublicApi;
 import software.amazon.awssdk.crt.CrtResource;
 import software.amazon.awssdk.crt.http.HttpClientConnectionManager;
+import software.amazon.awssdk.crt.http.HttpClientConnectionManagerOptions;
 import software.amazon.awssdk.crt.http.HttpHeader;
 import software.amazon.awssdk.crt.http.HttpRequest;
 import software.amazon.awssdk.crt.io.ClientBootstrap;
+import software.amazon.awssdk.crt.io.EventLoopGroup;
+import software.amazon.awssdk.crt.io.HostResolver;
 import software.amazon.awssdk.crt.io.SocketOptions;
 import software.amazon.awssdk.crt.io.TlsCipherPreference;
 import software.amazon.awssdk.crt.io.TlsContext;
@@ -80,9 +83,10 @@ public AwsCrtAsyncHttpClient(DefaultBuilder builder, AttributeMap config) {
         int maxConns = config.get(SdkHttpConfigurationOption.MAX_CONNECTIONS);
 
         Validate.isPositive(maxConns, "maxConns");
-        Validate.isPositive(builder.eventLoopSize, "eventLoopSize");
         Validate.notNull(builder.cipherPreference, "cipherPreference");
         Validate.isPositive(builder.windowSize, "windowSize");
+        Validate.notNull(builder.eventLoopGroup, "eventLoopGroup");
+        Validate.notNull(builder.hostResolver, "hostResolver");
 
         /**
          * Must call own() in same order that CrtResources are created in, so that they will be closed in reverse order.
@@ -93,12 +97,16 @@ public AwsCrtAsyncHttpClient(DefaultBuilder builder, AttributeMap config) {
          * in the correct order.
          */
 
-        bootstrap = own(new ClientBootstrap(builder.eventLoopSize));
-        socketOptions = own(new SocketOptions());
-        tlsContextOptions = own(TlsContextOptions.createDefaultClient().withCipherPreference(builder.cipherPreference));
-        tlsContextOptions.setVerifyPeer(builder.verifyPeer);
-        tlsContext = own(new TlsContext(tlsContextOptions));
+        this.bootstrap = own(new ClientBootstrap(builder.eventLoopGroup, builder.hostResolver));
+        this.socketOptions = own(new SocketOptions());
 
+        try (TlsContextOptions defaultClientOptions = TlsContextOptions.createDefaultClient()
+                .withCipherPreference(builder.cipherPreference)
+                .withVerifyPeer(builder.verifyPeer)) {
+            this.tlsContextOptions = own(defaultClientOptions);
+        }
+
+        this.tlsContext = own(new TlsContext(this.tlsContextOptions));
         this.windowSize = builder.windowSize;
         this.maxConnectionsPerEndpoint = maxConns;
     }
@@ -135,8 +143,15 @@ private HttpClientConnectionManager createConnectionPool(URI uri) {
         Validate.notNull(uri, "URI must not be null");
         log.debug(() -> "Creating ConnectionPool for: URI:" + uri + ", MaxConns: " + maxConnectionsPerEndpoint);
 
-        return new HttpClientConnectionManager(bootstrap, socketOptions, tlsContext, uri, windowSize,
-                                                maxConnectionsPerEndpoint);
+        HttpClientConnectionManagerOptions options = new HttpClientConnectionManagerOptions();
+        options.withClientBootstrap(bootstrap)
+               .withSocketOptions(socketOptions)
+               .withTlsContext(tlsContext)
+               .withUri(uri)
+               .withWindowSize(windowSize)
+               .withMaxConnections(maxConnectionsPerEndpoint);
+
+        return HttpClientConnectionManager.create(options);
     }
 
     private HttpClientConnectionManager getOrCreateConnectionPool(URI uri) {
@@ -194,7 +209,7 @@ private HttpHeader[] asArray(List<HttpHeader> crtHeaderList) {
         return crtHeaderList.toArray(new HttpHeader[crtHeaderList.size()]);
     }
 
-    private HttpRequest toCrtRequest(URI uri, AsyncExecuteRequest asyncRequest) {
+    private HttpRequest toCrtRequest(URI uri, AsyncExecuteRequest asyncRequest, AwsCrtAsyncHttpStreamAdapter crtToSdkAdapter) {
         SdkHttpRequest sdkRequest = asyncRequest.request();
         Validate.notNull(uri, "URI must not be null");
         Validate.notNull(sdkRequest, "SdkHttpRequest must not be null");
@@ -207,7 +222,7 @@ private HttpRequest toCrtRequest(URI uri, AsyncExecuteRequest asyncRequest) {
 
         HttpHeader[] crtHeaderArray = asArray(createHttpHeaderList(uri, asyncRequest));
 
-        return new HttpRequest(method, encodedPath + encodedQueryString, crtHeaderArray);
+        return new HttpRequest(method, encodedPath + encodedQueryString, crtHeaderArray, crtToSdkAdapter);
     }
 
     @Override
@@ -222,13 +237,12 @@ public CompletableFuture<Void> execute(AsyncExecuteRequest asyncRequest) {
 
         URI uri = toUri(asyncRequest.request());
         HttpClientConnectionManager crtConnPool = getOrCreateConnectionPool(uri);
-        HttpRequest crtRequest = toCrtRequest(uri, asyncRequest);
-
         CompletableFuture<Void> requestFuture = new CompletableFuture<>();
 
         // When a Connection is ready from the Connection Pool, schedule the Request on the connection
         crtConnPool.acquireConnection()
             .whenComplete((crtConn, throwable) -> {
+
                 // If we didn't get a connection for some reason, fail the request
                 if (throwable != null) {
                     requestFuture.completeExceptionally(throwable);
@@ -237,6 +251,7 @@ public CompletableFuture<Void> execute(AsyncExecuteRequest asyncRequest) {
 
                 AwsCrtAsyncHttpStreamAdapter crtToSdkAdapter =
                         new AwsCrtAsyncHttpStreamAdapter(crtConn, requestFuture, asyncRequest, windowSize);
+                HttpRequest crtRequest = toCrtRequest(uri, asyncRequest, crtToSdkAdapter);
 
                 // Submit the Request on this Connection
                 invokeSafely(() -> crtConn.makeRequest(crtRequest, crtToSdkAdapter));
@@ -263,24 +278,17 @@ public void close() {
      */
     public interface Builder extends SdkAsyncHttpClient.Builder<AwsCrtAsyncHttpClient.Builder> {
 
-        /**
-         * The number of Threads to use in the EventLoop.
-         * @param eventLoopSize The number of Threads to use in the EventLoop.
-         * @return the builder of the method chaining.
-         */
-        Builder eventLoopSize(int eventLoopSize);
-
         /**
          * The AWS CRT TlsCipherPreference to use for this Client
          * @param tlsCipherPreference The AWS Common Runtime TlsCipherPreference
-         * @return the builder of the method chaining.
+         * @return The builder of the method chaining.
          */
         Builder tlsCipherPreference(TlsCipherPreference tlsCipherPreference);
 
         /**
          * Whether or not to Verify the Peer's TLS Certificate Chain.
          * @param verifyPeer true if the Certificate Chain should be validated, false if validation should be skipped.
-         * @return the builder of the method chaining.
+         * @return The builder of the method chaining.
          */
         Builder verifyPeer(boolean verifyPeer);
 
@@ -290,9 +298,23 @@ public interface Builder extends SdkAsyncHttpClient.Builder<AwsCrtAsyncHttpClien
          * the Subscriber to read more data.
          *
          * @param windowSize The AWS Common Runtime WindowSize
-         * @return the builder of the method chaining.
+         * @return The builder of the method chaining.
          */
         Builder windowSize(int windowSize);
+
+        /**
+         * The AWS CRT EventLoopGroup to use for this Client.
+         * @param eventLoopGroup The AWS CRT EventLoopGroup to use for this client.
+         * @return The builder of the method chaining.
+         */
+        Builder eventLoopGroup(EventLoopGroup eventLoopGroup);
+
+        /**
+         * The AWS CRT HostResolver to use for this Client.
+         * @param hostResolver The AWS CRT HostResolver to use for this client.
+         * @return The builder of the method chaining.
+         */
+        Builder hostResolver(HostResolver hostResolver);
     }
 
     /**
@@ -301,10 +323,11 @@ public interface Builder extends SdkAsyncHttpClient.Builder<AwsCrtAsyncHttpClien
      */
     private static final class DefaultBuilder implements Builder {
         private final AttributeMap.Builder standardOptions = AttributeMap.builder();
-        private int eventLoopSize = Runtime.getRuntime().availableProcessors();
         private TlsCipherPreference cipherPreference = TlsCipherPreference.TLS_CIPHER_SYSTEM_DEFAULT;
         private int windowSize = DEFAULT_STREAM_WINDOW_SIZE;
         private boolean verifyPeer = true;
+        private EventLoopGroup eventLoopGroup;
+        private HostResolver hostResolver;
 
         private DefaultBuilder() {
         }
@@ -322,13 +345,6 @@ public SdkAsyncHttpClient buildWithDefaults(AttributeMap serviceDefaults) {
                                                            .merge(SdkHttpConfigurationOption.GLOBAL_HTTP_DEFAULTS));
         }
 
-        @Override
-        public Builder eventLoopSize(int eventLoopSize) {
-            Validate.isPositive(eventLoopSize, "eventLoopSize");
-            this.eventLoopSize = eventLoopSize;
-            return this;
-        }
-
         @Override
         public Builder tlsCipherPreference(TlsCipherPreference tlsCipherPreference) {
             Validate.notNull(tlsCipherPreference, "cipherPreference");
@@ -350,5 +366,17 @@ public Builder windowSize(int windowSize) {
             this.windowSize = windowSize;
             return this;
         }
+
+        @Override
+        public Builder eventLoopGroup(EventLoopGroup eventLoopGroup) {
+            this.eventLoopGroup = eventLoopGroup;
+            return this;
+        }
+
+        @Override
+        public Builder hostResolver(HostResolver hostResolver) {
+            this.hostResolver = hostResolver;
+            return this;
+        }
     }
 }