Fix OneTimeReadTimeoutHandler behavior

 - Extend ReadTimeoutHandler for timeout behavior
 - Remove self from pipeline once a message is read

Author: dagnir

.changes/next-release/bugfix-NettyNIOHTTPClient-aa3a23a.json

@@ -0,0 +1,5 @@
+{
+    "category": "Netty NIO HTTP Client", 
+    "type": "bugfix", 
+    "description": "Fix a bug in the Netty client where the read timeout isn't applied correctly for some requests."
+}

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

@@ -47,6 +47,7 @@
 import java.io.IOException;
 import java.net.URI;
 import java.nio.ByteBuffer;
+import java.time.Duration;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -210,8 +211,8 @@ private void writeRequest(HttpRequest request) {
 
             // Should only add an one-time ReadTimeoutHandler to 100 Continue request.
             if (is100ContinueExpected()) {
-                channel.pipeline().addFirst(new OneTimeReadTimeoutHandler(context.configuration().readTimeoutMillis(),
-                                                                          TimeUnit.MILLISECONDS));
+                channel.pipeline().addFirst(new OneTimeReadTimeoutHandler(Duration.ofMillis(context.configuration()
+                        .readTimeoutMillis())));
             } else {
                 channel.pipeline().addFirst(new ReadTimeoutHandler(context.configuration().readTimeoutMillis(),
                                                                    TimeUnit.MILLISECONDS));

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

@@ -16,34 +16,24 @@
 package software.amazon.awssdk.http.nio.netty.internal;
 
 import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.handler.timeout.ReadTimeoutHandler;
-import io.netty.util.ReferenceCountUtil;
+import java.time.Duration;
 import java.util.concurrent.TimeUnit;
 import software.amazon.awssdk.annotations.SdkInternalApi;
 
 /**
- * Handler to add an one-time {@link ReadTimeoutHandler} to the pipeline and remove it afterwards.
+ * A one-time read timeout handler that removes itself from the pipeline after
+ * the next successful read.
  */
 @SdkInternalApi
-public final class OneTimeReadTimeoutHandler extends SimpleChannelInboundHandler {
-
-    private static final String READ_TIMEOUT_HANDLER_NAME = "RemoveAfterReadTimeoutHandler";
-    private final long readTimeoutMillis;
-    private final TimeUnit timeUnit;
-
-    OneTimeReadTimeoutHandler(long readTimeoutMillis, TimeUnit timeUnit) {
-        this.readTimeoutMillis = readTimeoutMillis;
-        this.timeUnit = timeUnit;
+public final class OneTimeReadTimeoutHandler extends ReadTimeoutHandler {
+    OneTimeReadTimeoutHandler(Duration timeout) {
+        super(timeout.toMillis(), TimeUnit.MILLISECONDS);
     }
 
     @Override
-    public void channelRead0(ChannelHandlerContext ctx, Object msg) {
-        ReferenceCountUtil.retain(msg);
-        ctx.pipeline().addFirst(READ_TIMEOUT_HANDLER_NAME, new ReadTimeoutHandler(readTimeoutMillis, timeUnit));
-        ctx.fireChannelRead(msg);
-
-        ctx.pipeline().remove(READ_TIMEOUT_HANDLER_NAME);
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
         ctx.pipeline().remove(this);
+        super.channelRead(ctx, msg);
     }
 }

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

@@ -15,19 +15,17 @@
 
 package software.amazon.awssdk.http.nio.netty.internal;
 
-import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPipeline;
-import io.netty.handler.timeout.ReadTimeoutHandler;
-import java.util.concurrent.TimeUnit;
+import java.time.Duration;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
 
@@ -48,29 +46,16 @@ public class OneTimeReadTimeoutHandlerTest {
 
     @BeforeClass
     public static void setup() {
-        handler = new OneTimeReadTimeoutHandler(TIMEOUT_IN_MILLIS, TimeUnit.MILLISECONDS);
+        handler = new OneTimeReadTimeoutHandler(Duration.ofMillis(TIMEOUT_IN_MILLIS));
     }
 
     @Test
-    public void channelRead_shouldAddReadTimeoutHandlerBeforeRead() {
-        ArgumentCaptor<ReadTimeoutHandler> argumentCaptor = ArgumentCaptor.forClass(ReadTimeoutHandler.class);
-        ArgumentCaptor<String> handlerNameCaptor = ArgumentCaptor.forClass(String.class);
-
+    public void channelRead_removesSelf() throws Exception {
         when(context.pipeline()).thenReturn(channelPipeline);
 
-        handler.channelRead0(context, object);
-
-        verify(channelPipeline, times(1)).addFirst(handlerNameCaptor.capture(),
-                                                   argumentCaptor.capture());
+        handler.channelRead(context, object);
 
+        verify(channelPipeline, times(1)).remove(eq(handler));
         verify(context, times(1)).fireChannelRead(object);
-
-        ReadTimeoutHandler readTimeoutHandler = argumentCaptor.getValue();
-        assertThat(readTimeoutHandler.getReaderIdleTimeInMillis()).isEqualTo(TIMEOUT_IN_MILLIS);
-
-        verify(channelPipeline, times(1)).remove(handlerNameCaptor.getValue());
-
-        verify(channelPipeline, times(1)).remove(handler);
-
     }
 }