Bug fix for S3AsyncClient.putObject hangs if there is a connection reset set while uploading of object (#3535)

* Bug fix for S3AsyncClient.putObject hangs if there is a connection reset while uploading of objects

* Changing LAST_HTTP_CONTENT_RECEIVED_KEY key name to STREAMING_COMPLETE_KEY and setting it irrespective of ignoreBodyRead flag

* Added SimpleHttpContentPublisher and removed SdkTestHttpContentPublisher

* Moving log in handleReadHttpContent() to log irrespective of ignoreBodyRead flag

Author: joviegas

.changes/next-release/bugfix-NettyNIOHttpClient-b7e834c.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "Netty NIO Http Client",
+    "contributor": "",
+    "description": "Fix for Netty based client request getting stuck if connection is reset after recieveing Http Continue response."
+}

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

@@ -73,6 +73,13 @@ public final class ChannelAttributeKey {
     public static final AttributeKey<ChannelDiagnostics> CHANNEL_DIAGNOSTICS = NettyUtils.getOrCreateAttributeKey(
         "aws.http.nio.netty.async.channelDiagnostics");
 
+    /**
+     * {@link AttributeKey} to keep track of whether the streaming is completed and this is set to true when we receive the *
+     * {@link LastHttpContent}.
+     */
+    public static final AttributeKey<Boolean> STREAMING_COMPLETE_KEY = NettyUtils.getOrCreateAttributeKey(
+        "aws.http.nio.netty.async.streamingComplete");
+
     /**
      * {@link AttributeKey} to keep track of whether we should close the connection after this request
      * has completed.
@@ -100,12 +107,6 @@ public final class ChannelAttributeKey {
     static final AttributeKey<Long> RESPONSE_DATA_READ = NettyUtils.getOrCreateAttributeKey(
         "aws.http.nio.netty.async.responseDataRead");
 
-    /**
-     * {@link AttributeKey} to keep track of whether we have received the {@link LastHttpContent}.
-     */
-    static final AttributeKey<Boolean> LAST_HTTP_CONTENT_RECEIVED_KEY = NettyUtils.getOrCreateAttributeKey(
-        "aws.http.nio.netty.async.lastHttpContentReceived");
-
     static final AttributeKey<CompletableFuture<Void>> EXECUTE_FUTURE_KEY = NettyUtils.getOrCreateAttributeKey(
             "aws.http.nio.netty.async.executeFuture");
 

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

@@ -49,7 +49,6 @@ public void channelReleased(Channel channel) {
         if (channel.isOpen() || channel.isRegistered()) {
             removeIfExists(channel.pipeline(),
                            HttpStreamsClientHandler.class,
-                           LastHttpContentHandler.class,
                            FlushOnReadHandler.class,
                            ResponseHandler.class,
                            ReadTimeoutHandler.class,

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

@@ -21,11 +21,11 @@
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.EXECUTION_ID_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.IN_USE;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.KEEP_ALIVE;
-import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.LAST_HTTP_CONTENT_RECEIVED_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.REQUEST_CONTEXT_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.RESPONSE_COMPLETE_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.RESPONSE_CONTENT_LENGTH;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.RESPONSE_DATA_READ;
+import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.STREAMING_COMPLETE_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.NettyRequestMetrics.measureTimeTaken;
 
 import io.netty.buffer.ByteBuf;
@@ -195,7 +195,7 @@ private void configureChannel() {
         channel.attr(EXECUTE_FUTURE_KEY).set(executeFuture);
         channel.attr(REQUEST_CONTEXT_KEY).set(context);
         channel.attr(RESPONSE_COMPLETE_KEY).set(false);
-        channel.attr(LAST_HTTP_CONTENT_RECEIVED_KEY).set(false);
+        channel.attr(STREAMING_COMPLETE_KEY).set(false);
         channel.attr(RESPONSE_CONTENT_LENGTH).set(null);
         channel.attr(RESPONSE_DATA_READ).set(null);
         channel.attr(CHANNEL_DIAGNOSTICS).get().incrementRequestCount();
@@ -220,7 +220,6 @@ private void configurePipeline() throws IOException {
                 throw new IOException("Unknown protocol: " + protocol);
         }
 
-        pipeline.addLast(LastHttpContentHandler.create());
         if (protocol == Protocol.HTTP2) {
             pipeline.addLast(FlushOnReadHandler.getInstance());
         }

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

@@ -19,12 +19,12 @@
 import static java.util.stream.Collectors.mapping;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.EXECUTE_FUTURE_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.KEEP_ALIVE;
-import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.LAST_HTTP_CONTENT_RECEIVED_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.REQUEST_CONTEXT_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.RESPONSE_COMPLETE_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.RESPONSE_CONTENT_LENGTH;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.RESPONSE_DATA_READ;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.RESPONSE_STATUS_CODE;
+import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.STREAMING_COMPLETE_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.utils.ExceptionHandlingUtils.tryCatch;
 import static software.amazon.awssdk.http.nio.netty.internal.utils.ExceptionHandlingUtils.tryCatchFinally;
 
@@ -463,10 +463,10 @@ public void cancel() {
     private void notifyIfResponseNotCompleted(ChannelHandlerContext handlerCtx) {
         RequestContext requestCtx = handlerCtx.channel().attr(REQUEST_CONTEXT_KEY).get();
         Boolean responseCompleted = handlerCtx.channel().attr(RESPONSE_COMPLETE_KEY).get();
-        Boolean lastHttpContentReceived = handlerCtx.channel().attr(LAST_HTTP_CONTENT_RECEIVED_KEY).get();
+        Boolean isStreamingComplete = handlerCtx.channel().attr(STREAMING_COMPLETE_KEY).get();
         handlerCtx.channel().attr(KEEP_ALIVE).set(false);
 
-        if (!Boolean.TRUE.equals(responseCompleted) && !Boolean.TRUE.equals(lastHttpContentReceived)) {
+        if (!Boolean.TRUE.equals(responseCompleted) && !Boolean.TRUE.equals(isStreamingComplete)) {
             IOException err = new IOException(NettyUtils.closedChannelMessage(handlerCtx.channel()));
             runAndLogError(handlerCtx.channel(), () -> "Fail to execute SdkAsyncHttpResponseHandler#onError",
                            () -> requestCtx.handler().onError(err));

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

@@ -15,6 +15,8 @@
 
 package software.amazon.awssdk.http.nio.netty.internal.nrs;
 
+import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.STREAMING_COMPLETE_KEY;
+
 import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
@@ -30,6 +32,7 @@
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.http.nio.netty.internal.utils.NettyClientLogger;
 
 /**
  * This class contains source imported from https://github.com/playframework/netty-reactive-streams,
@@ -42,6 +45,7 @@
 @SdkInternalApi
 abstract class HttpStreamsHandler<InT extends HttpMessage, OutT extends HttpMessage> extends ChannelDuplexHandler {
 
+    private static final NettyClientLogger logger = NettyClientLogger.getLogger(HttpStreamsHandler.class);
     private final Queue<Outgoing> outgoing = new LinkedList<>();
     private final Class<InT> inClass;
     private final Class<OutT> outClass;
@@ -209,11 +213,16 @@ private void handleCancelled(ChannelHandlerContext ctx, InT msg) {
     }
 
     private void handleReadHttpContent(ChannelHandlerContext ctx, HttpContent content) {
+        boolean lastHttpContent = content instanceof LastHttpContent;
+        if (lastHttpContent) {
+            logger.debug(ctx.channel(),
+                         () -> "Received LastHttpContent " + ctx.channel() + " with ignoreBodyRead as " + ignoreBodyRead);
+            ctx.channel().attr(STREAMING_COMPLETE_KEY).set(true);
+        }
         if (!ignoreBodyRead) {
-            if (content instanceof LastHttpContent) {
-
+            if (lastHttpContent) {
                 if (content.content().readableBytes() > 0 ||
-                        !((LastHttpContent) content).trailingHeaders().isEmpty()) {
+                    !((LastHttpContent) content).trailingHeaders().isEmpty()) {
                     // It has data or trailing headers, send them
                     ctx.fireChannelRead(content);
                 } else {
@@ -230,7 +239,7 @@ private void handleReadHttpContent(ChannelHandlerContext ctx, HttpContent conten
 
         } else {
             ReferenceCountUtil.release(content);
-            if (content instanceof LastHttpContent) {
+            if (lastHttpContent) {
                 ignoreBodyRead = false;
                 if (currentlyStreamedMessage != null) {
                     removeHandlerIfActive(ctx, ctx.name() + "-body-publisher");