Enable auto reads on idling connections (#4655)

This change enables auto reads on idling connections, and disables them on
in-use connections (no auto read for in-use connections is existing behavior).
Auto reads while the channel is idling allows the channel to read and
immediately handle events that are triggered by the remote end such as TLS close
notifies. In particular this fixes issues where a channel is leased only for the
SDK to find that the remote end has initiated a shutdown of the TLS connection.

Author: dagnir

.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();
+    }
+
 }