Close created ChannelPools in close method

Author: dagnir

.changes/next-release/bugfix-NettyNIOAsyncHTTPClient-07f0d1e.json

@@ -0,0 +1,5 @@
+{
+    "category": "Netty NIO Async HTTP Client", 
+    "type": "bugfix", 
+    "description": "Close created `ChannelPool`s in `close()` method."
+}

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/NettyNioAsyncHttpClient.java

@@ -22,12 +22,12 @@
 import static software.amazon.awssdk.http.SdkHttpConfigurationOption.READ_TIMEOUT;
 import static software.amazon.awssdk.http.SdkHttpConfigurationOption.WRITE_TIMEOUT;
 import static software.amazon.awssdk.utils.FunctionalUtils.invokeSafely;
+import static software.amazon.awssdk.utils.FunctionalUtils.runAndLogError;
 
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.pool.ChannelPool;
-import io.netty.channel.pool.ChannelPoolMap;
 import io.netty.handler.codec.http2.Http2SecurityUtil;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
@@ -39,6 +39,9 @@
 import java.util.concurrent.atomic.AtomicReference;
 import javax.net.ssl.SSLException;
 import javax.net.ssl.TrustManagerFactory;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.annotations.SdkPublicApi;
 import software.amazon.awssdk.http.Protocol;
 import software.amazon.awssdk.http.SdkHttpConfigurationOption;
@@ -67,8 +70,10 @@
  */
 @SdkPublicApi
 public final class NettyNioAsyncHttpClient implements SdkAsyncHttpClient {
+    private static final Logger log = LoggerFactory.getLogger(NettyNioAsyncHttpClient.class);
+
     private final SdkEventLoopGroup sdkEventLoopGroup;
-    private final ChannelPoolMap<URI, ChannelPool> pools;
+    private final SdkChannelPoolMap<URI, ChannelPool> pools;
     private final SdkChannelOptions sdkChannelOptions;
     private final NettyConfiguration configuration;
     private final long maxStreams;
@@ -134,7 +139,7 @@ private TrustManagerFactory getTrustManager() {
         return configuration.trustAllCertificates() ? InsecureTrustManagerFactory.INSTANCE : null;
     }
 
-    private ChannelPoolMap<URI, ChannelPool> createChannelPoolMap() {
+    private SdkChannelPoolMap<URI, ChannelPool> createChannelPoolMap() {
         return new SdkChannelPoolMap<URI, ChannelPool>() {
             @Override
             protected ChannelPool newPool(URI key) {
@@ -166,7 +171,8 @@ private SdkEventLoopGroup nonManagedEventLoopGroup(SdkEventLoopGroup eventLoopGr
 
     @Override
     public void close() {
-        sdkEventLoopGroup.eventLoopGroup().shutdownGracefully();
+        pools.forEach(e -> runAndLogError(log, "Unable to close channel pool for " + e.getKey(), e.getValue()::close));
+        runAndLogError(log, "Unable to shutdown event loop", sdkEventLoopGroup.eventLoopGroup()::shutdownGracefully);
     }
 
     /**

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/SdkEventLoopGroup.java

@@ -52,7 +52,7 @@ public final class SdkEventLoopGroup {
     private final EventLoopGroup eventLoopGroup;
     private final ChannelFactory<? extends Channel> channelFactory;
 
-    private SdkEventLoopGroup(EventLoopGroup eventLoopGroup, ChannelFactory<? extends Channel> channelFactory) {
+    SdkEventLoopGroup(EventLoopGroup eventLoopGroup, ChannelFactory<? extends Channel> channelFactory) {
         Validate.paramNotNull(eventLoopGroup, "eventLoopGroup");
         Validate.paramNotNull(channelFactory, "channelFactory");
         this.eventLoopGroup = eventLoopGroup;

http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/NettyNioAsyncHttpClientWireMockTest.java

@@ -44,6 +44,7 @@
 import com.github.tomakehurst.wiremock.http.trafficlistener.WiremockNetworkTrafficListener;
 import com.github.tomakehurst.wiremock.junit.WireMockRule;
 import io.netty.channel.ChannelFactory;
+import io.netty.channel.ChannelFuture;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioSocketChannel;
@@ -284,6 +285,47 @@ public void requestContentOnlyEqualToContentLengthHeaderFromProvider() throws In
         assertThat(wiremockTrafficListener.requests.toString()).endsWith(content);
     }
 
+    @Test
+    public void closeMethodClosesOpenedChannels() throws InterruptedException, TimeoutException, ExecutionException {
+        String body = randomAlphabetic(10);
+        URI uri = URI.create("https://localhost:" + mockServer.httpsPort());
+        stubFor(any(urlPathEqualTo("/")).willReturn(aResponse().withHeader("Some-Header", "With Value").withBody(body)));
+
+        SdkHttpFullRequest request = createRequest(uri, "/", body, SdkHttpMethod.POST, Collections.emptyMap());
+        RecordingResponseHandler recorder = new RecordingResponseHandler();
+
+        CompletableFuture<Boolean> channelClosedFuture = new CompletableFuture<>();
+        ChannelFactory<NioSocketChannel> channelFactory = new ChannelFactory<NioSocketChannel>() {
+            @Override
+            public NioSocketChannel newChannel() {
+                return new NioSocketChannel() {
+                    @Override
+                    public ChannelFuture close() {
+                        ChannelFuture cf = super.close();
+                        channelClosedFuture.complete(true);
+                        return cf;
+                    }
+                };
+            }
+        };
+
+        SdkAsyncHttpClient customClient = NettyNioAsyncHttpClient.builder()
+                .eventLoopGroup(new SdkEventLoopGroup(new NioEventLoopGroup(1), channelFactory))
+                .buildWithDefaults(mapWithTrustAllCerts());
+
+        try {
+            customClient.execute(AsyncExecuteRequest.builder()
+                    .request(request)
+                    .requestContentPublisher(createProvider(body))
+                    .responseHandler(recorder).build())
+                    .join();
+        } finally {
+            customClient.close();
+        }
+
+        assertThat(channelClosedFuture.get(5, TimeUnit.SECONDS)).isTrue();
+    }
+
     private void assertCanReceiveBasicRequest(URI uri, String body) throws Exception {
         stubFor(any(urlPathEqualTo("/")).willReturn(aResponse().withHeader("Some-Header", "With Value").withBody(body)));