Fix race codition between read and finalizeRequest

HttpStreamsClientHandler ignores the LastHttpContent decoded by the HTTP
codec when we get an empty response from the service, but if the channel
gets released (which causes the streams handler to get removed) before
the last content message comes through and is ignored by the handler,
the new HttpStreamsHandler (added by RunnableRquest) will not be in a
state to anticipate and ignore the LastHttpContent and will instead do
the wrong thing and attempt to remove a non-existent handler.

This commit fixes that race condition by explicitly replacing the
HttpStreamsClientHandler in the pipeline with a new handler that ignores
the next message read in the pipeline if it's a LastHttpContent.

Fixes #202

Author: dagnir

.changes/next-release/bugfix-NettyNIOAsyncHTTPClient-e5d4229.json

@@ -0,0 +1,5 @@
+{
+    "category": "Netty NIO Async HTTP Client", 
+    "type": "bugfix", 
+    "description": "Fix race condition in the async client causing instability when making multiple concurent requests. Fixes [#202](https://github.com/aws/aws-sdk-java-v2/issues/202)"
+}

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

@@ -20,6 +20,7 @@
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKeys.REQUEST_CONTEXT_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKeys.RESPONSE_COMPLETE_KEY;
 
+import com.typesafe.netty.http.HttpStreamsClientHandler;
 import com.typesafe.netty.http.StreamedHttpResponse;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandler.Sharable;
@@ -31,6 +32,7 @@
 import io.netty.handler.codec.http.HttpObject;
 import io.netty.handler.codec.http.HttpResponse;
 import io.netty.handler.codec.http.HttpUtil;
+import io.netty.handler.codec.http.LastHttpContent;
 import io.netty.util.AttributeKey;
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -75,12 +77,10 @@ protected void channelRead0(ChannelHandlerContext channelContext, HttpObject msg
         if (msg instanceof StreamedHttpResponse) {
             requestContext.handler().onStream(new PublisherAdapter((StreamedHttpResponse) msg, channelContext, requestContext));
         } else if (msg instanceof FullHttpResponse) {
-            // TODO: HttpStreamsClientHandler leaves a dangling LastHttpContent
-            // in the pipeline. Consume it ourselves to make sure the channel
-            // is empty at the end of stream and before releasing it back to he
-            // pool.  The HttpStreamsClientHandler should really be doing this
-            // for us.
-            channelContext.read();
+            // Be prepared to take care of (ignore) a trailing LastHttpResponse
+            // from the HttpClientCodec if there is one.
+            channelContext.pipeline().replace(HttpStreamsClientHandler.class,
+                    channelContext.name() + "-LastHttpContentSwallower", new LastHttpContentSwallower());
 
             ByteBuf fullContent = ((FullHttpResponse) msg).content();
             final ByteBuffer bb = copyToByteBuffer(fullContent);
@@ -239,4 +239,20 @@ public void subscribe(Subscriber<? super ByteBuffer> subscriber) {
                     .set(subscriber);
         }
     }
+
+
+    private static class LastHttpContentSwallower extends SimpleChannelInboundHandler<HttpObject> {
+        @Override
+        protected void channelRead0(ChannelHandlerContext ctx, HttpObject obj) throws Exception {
+            if (obj instanceof LastHttpContent) {
+                // Queue another read to make up for the one we just ignored
+                ctx.read();
+            } else {
+                ctx.fireChannelRead(obj);
+            }
+            // Remove self from pipeline since we only care about potentially
+            // ignoring the very first message
+            ctx.pipeline().remove(this);
+        }
+    }
 }