Enable auto reads on idling connections

.changes/next-release/bugfix-NettyNIOHTTPClient-1f36854.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "Netty NIO HTTP Client",
+    "contributor": "",
+    "description": "Fix an issue where data received on a channel while it was idling was not handled until the channel was leased again for a request. This caused issues such as late notification of channel closes, manifesting as channels being closed at the beginning of a request."
+}

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

@@ -0,0 +1,41 @@
+/*
+ * Copyright 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.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.http.nio.netty.internal;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandler;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+
+/**
+ * Disables auto read on in-use channels to allow upper layers to take care of flow control.
+ */
+@SdkInternalApi
+@ChannelHandler.Sharable
+public final class AutoReadDisableChannelPoolListener implements ListenerInvokingChannelPool.ChannelPoolListener {
+    private static final AutoReadDisableChannelPoolListener INSTANCE = new AutoReadDisableChannelPoolListener();
+
+    private AutoReadDisableChannelPoolListener() {
+    }
+
+    @Override
+    public void channelAcquired(Channel channel) {
+        channel.config().setAutoRead(false);
+    }
+
+    public static AutoReadDisableChannelPoolListener create() {
+        return INSTANCE;
+    }
+}

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

@@ -0,0 +1,41 @@
+/*
+ * Copyright 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.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.http.nio.netty.internal;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandler;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+
+/**
+ * Enables auto read on idle channels so that any data that a service sends while it's idling can be handled.
+ */
+@SdkInternalApi
+@ChannelHandler.Sharable
+public final class AutoReadEnableChannelPoolListener implements ListenerInvokingChannelPool.ChannelPoolListener {
+    private static final AutoReadEnableChannelPoolListener INSTANCE = new AutoReadEnableChannelPoolListener();
+
+    private AutoReadEnableChannelPoolListener() {
+    }
+
+    @Override
+    public void channelReleased(Channel channel) {
+        channel.config().setAutoRead(true);
+    }
+
+    public static AutoReadEnableChannelPoolListener create() {
+        return INSTANCE;
+    }
+}

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

@@ -239,12 +239,19 @@ private SdkChannelPool wrapBaseChannelPool(Bootstrap bootstrap, ChannelPool chan
                                                                    configuration.maxConnections(),
                                                                    configuration);
 
+
         sdkChannelPool = new ListenerInvokingChannelPool(bootstrap.config().group(), sdkChannelPool, Arrays.asList(
+            // Add a listener that disables auto reads on acquired connections.
+            AutoReadDisableChannelPoolListener.create(),
+
             // Add a listener that ensures acquired channels are marked IN_USE and thus not eligible for certain idle timeouts.
             InUseTrackingChannelPoolListener.create(),
 
             // Add a listener that removes request-specific handlers with each request.
-            HandlerRemovingChannelPoolListener.create()
+            HandlerRemovingChannelPoolListener.create(),
+
+            // Add a listener that enables auto reads on released connections.
+            AutoReadEnableChannelPoolListener.create()
         ));
 
         // Wrap the channel pool such that an individual channel can only be released to the underlying pool once.

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

@@ -31,7 +31,6 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
-import io.netty.channel.ChannelOption;
 import io.netty.channel.ChannelPipeline;
 import io.netty.handler.codec.DecoderResult;
 import io.netty.handler.codec.http.DefaultHttpContent;
@@ -199,7 +198,6 @@ private void configureChannel() {
         channel.attr(RESPONSE_CONTENT_LENGTH).set(null);
         channel.attr(RESPONSE_DATA_READ).set(null);
         channel.attr(CHANNEL_DIAGNOSTICS).get().incrementRequestCount();
-        channel.config().setOption(ChannelOption.AUTO_READ, false);
     }
 
     private void configurePipeline() throws IOException {

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

@@ -268,4 +268,42 @@ public void usesProvidedKeyManagersProvider() {
         verify(provider).keyManagers();
     }
 
+    @Test
+    public void acquireChannel_autoReadDisabled() {
+        channelPoolMap = AwaitCloseChannelPoolMap.builder()
+                                                 .sdkChannelOptions(new SdkChannelOptions())
+                                                 .sdkEventLoopGroup(SdkEventLoopGroup.builder().build())
+                                                 .configuration(new NettyConfiguration(GLOBAL_HTTP_DEFAULTS))
+                                                 .protocol(Protocol.HTTP1_1)
+                                                 .maxStreams(100)
+                                                 .sslProvider(SslProvider.OPENSSL)
+                                                 .build();
+
+        ChannelPool channelPool = channelPoolMap.newPool(URI.create("https://localhost:" + mockProxy.port()));
+
+        Channel channel = channelPool.acquire().awaitUninterruptibly().getNow();
+
+        assertThat(channel.config().isAutoRead()).isFalse();
+    }
+
+    @Test
+    public void releaseChannel_autoReadEnabled() {
+        channelPoolMap = AwaitCloseChannelPoolMap.builder()
+                                                 .sdkChannelOptions(new SdkChannelOptions())
+                                                 .sdkEventLoopGroup(SdkEventLoopGroup.builder().build())
+                                                 .configuration(new NettyConfiguration(GLOBAL_HTTP_DEFAULTS))
+                                                 .protocol(Protocol.HTTP1_1)
+                                                 .maxStreams(100)
+                                                 .sslProvider(SslProvider.OPENSSL)
+                                                 .build();
+
+        ChannelPool channelPool = channelPoolMap.newPool(URI.create("https://localhost:" + mockProxy.port()));
+
+        Channel channel = channelPool.acquire().awaitUninterruptibly().getNow();
+
+        channelPool.release(channel).awaitUninterruptibly();
+
+        assertThat(channel.config().isAutoRead()).isTrue();
+    }
+
 }