Fix the bug where channel is not being released back to the pool when there is an exception thrown

Author: zoewangg

.changes/next-release/bugfix-NettyNioHttpClient-ed05f35.json

@@ -0,0 +1,5 @@
+{
+    "category": "Netty Nio Http Client", 
+    "type": "bugfix", 
+    "description": "Fix a bug where the channel fails to be released if there is an exception thrown."
+}

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

@@ -312,7 +312,8 @@ private String getMessageForTooManyAcquireOperationsError() {
      */
     private void closeAndRelease(Channel channel) {
         log.trace("closing and releasing channel {}", channel.id().asLongText());
-        channel.close().addListener(ignored -> context.channelPool().release(channel));
+        channel.close();
+        context.channelPool().release(channel);
     }
 
     /**

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

@@ -165,7 +165,8 @@ public static ResponseHandler getInstance() {
     private static void closeAndRelease(ChannelHandlerContext ctx) {
         Channel channel = ctx.channel();
         RequestContext requestContext = channel.attr(REQUEST_CONTEXT_KEY).get();
-        channel.close().addListener(i -> requestContext.channelPool().release(channel));
+        ctx.close();
+        requestContext.channelPool().release(channel);
     }
 
     /**

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

@@ -41,14 +41,17 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.when;
 
+import com.github.tomakehurst.wiremock.http.Fault;
 import com.github.tomakehurst.wiremock.http.trafficlistener.WiremockNetworkTrafficListener;
 import com.github.tomakehurst.wiremock.junit.WireMockRule;
+import io.netty.channel.Channel;
 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.pool.ChannelPool;
 import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.util.AttributeKey;
 import java.io.IOException;
 import java.net.Socket;
 import java.net.URI;
@@ -225,6 +228,110 @@ protected ChannelPool newPool(URI key) {
         Mockito.verify(channelPool).close();
     }
 
+    @Test
+    public void responseConnectionReused_shouldReleaseChannel() throws Exception {
+
+        ChannelFactory channelFactory = mock(ChannelFactory.class);
+        EventLoopGroup customEventLoopGroup = new NioEventLoopGroup(1);
+        NioSocketChannel channel = new NioSocketChannel();
+
+        when(channelFactory.newChannel()).thenAnswer((Answer<NioSocketChannel>) invocationOnMock -> channel);
+        SdkEventLoopGroup eventLoopGroup = SdkEventLoopGroup.create(customEventLoopGroup, channelFactory);
+
+        NettyNioAsyncHttpClient customClient =
+            (NettyNioAsyncHttpClient) NettyNioAsyncHttpClient.builder()
+                                                             .eventLoopGroup(eventLoopGroup)
+                                                             .maxConcurrency(1)
+                                                             .build();
+
+        makeSimpleRequest(customClient);
+        verifyChannelRelease(channel);
+        assertThat(channel.isShutdown()).isFalse();
+
+        customClient.close();
+        eventLoopGroup.eventLoopGroup().shutdownGracefully().awaitUninterruptibly();
+    }
+
+    @Test
+    public void connectionInactive_shouldReleaseChannel() throws Exception {
+
+        ChannelFactory channelFactory = mock(ChannelFactory.class);
+        EventLoopGroup customEventLoopGroup = new NioEventLoopGroup(1);
+        NioSocketChannel channel = new NioSocketChannel();
+
+        when(channelFactory.newChannel()).thenAnswer((Answer<NioSocketChannel>) invocationOnMock -> channel);
+        SdkEventLoopGroup eventLoopGroup = SdkEventLoopGroup.create(customEventLoopGroup, channelFactory);
+
+        NettyNioAsyncHttpClient customClient =
+            (NettyNioAsyncHttpClient) NettyNioAsyncHttpClient.builder()
+                                                             .eventLoopGroup(eventLoopGroup)
+                                                             .maxConcurrency(1)
+                                                             .build();
+
+
+        String body = randomAlphabetic(10);
+        URI uri = URI.create("http://localhost:" + mockServer.port());
+        SdkHttpRequest request = createRequest(uri);
+        RecordingResponseHandler recorder = new RecordingResponseHandler();
+
+
+        stubFor(any(urlPathEqualTo("/")).willReturn(aResponse().withBody(body)
+                                                               .withStatus(500)
+                                                               .withFault(Fault.RANDOM_DATA_THEN_CLOSE)));
+
+        customClient.execute(AsyncExecuteRequest.builder()
+                                                .request(request)
+                                                .requestContentPublisher(createProvider(""))
+                                                .responseHandler(recorder).build());
+
+        verifyChannelRelease(channel);
+        assertThat(channel.isShutdown()).isTrue();
+
+        customClient.close();
+        eventLoopGroup.eventLoopGroup().shutdownGracefully().awaitUninterruptibly();
+    }
+
+    @Test
+    public void responseConnectionClosed_shouldCloseAndReleaseChannel() throws Exception {
+
+        ChannelFactory channelFactory = mock(ChannelFactory.class);
+        EventLoopGroup customEventLoopGroup = new NioEventLoopGroup(1);
+        NioSocketChannel channel = new NioSocketChannel();
+
+        when(channelFactory.newChannel()).thenAnswer((Answer<NioSocketChannel>) invocationOnMock -> channel);
+
+        URI uri = URI.create("http://localhost:" + mockServer.port());
+        SdkHttpRequest request = createRequest(uri);
+        RecordingResponseHandler recorder = new RecordingResponseHandler();
+
+        SdkEventLoopGroup eventLoopGroup = SdkEventLoopGroup.create(customEventLoopGroup, channelFactory);
+
+        NettyNioAsyncHttpClient customClient =
+            (NettyNioAsyncHttpClient) NettyNioAsyncHttpClient.builder()
+                                                             .eventLoopGroup(eventLoopGroup)
+                                                             .maxConcurrency(1)
+                                                             .build();
+
+        String body = randomAlphabetic(10);
+
+        stubFor(any(urlPathEqualTo("/")).willReturn(aResponse().withBody(body)
+                                                               .withStatus(500)
+                                                               .withHeader("Connection", "close")
+        ));
+
+        customClient.execute(AsyncExecuteRequest.builder()
+                                                .request(request)
+                                                .requestContentPublisher(createProvider(""))
+                                                .responseHandler(recorder).build());
+        recorder.completeFuture.get(5, TimeUnit.SECONDS);
+
+        verifyChannelRelease(channel);
+        assertThat(channel.isShutdown()).isTrue();
+
+        customClient.close();
+        eventLoopGroup.eventLoopGroup().shutdownGracefully().awaitUninterruptibly();
+    }
+
     /**
      * Make a simple async request and wait for it to fiish.
      *
@@ -491,6 +598,11 @@ public void testExceptionMessageChanged_WhenConnectionTimeoutErrorEncountered()
         customClient.close();
     }
 
+    private void verifyChannelRelease(Channel channel) throws InterruptedException {
+        Thread.sleep(1000);
+        assertThat(channel.attr(AttributeKey.valueOf("channelPool")).get()).isNull();
+    }
+
     private RecordingResponseHandler makeSimpleRequestAndReturnResponseHandler(SdkAsyncHttpClient client) throws Exception {
         String body = randomAlphabetic(10);
         URI uri = URI.create("http://localhost:" + mockServer.port());