#6949 Fix Netty streaming completion guard

Signed-off-by: Arnab Nandy <arnab_nandy7@yahoo.com>

Author: arnabnandy7

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

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "Netty NIO Async HTTP Client",
+    "contributor": "",
+    "description": "Fix async streaming responses so channel-inactive errors are surfaced until the response has been finalized."
+}

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

@@ -25,7 +25,6 @@
 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;
 
@@ -462,10 +461,9 @@ 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 isStreamingComplete = handlerCtx.channel().attr(STREAMING_COMPLETE_KEY).get();
         handlerCtx.channel().attr(KEEP_ALIVE).set(false);
 
-        if (!Boolean.TRUE.equals(responseCompleted) && !Boolean.TRUE.equals(isStreamingComplete)) {
+        if (!Boolean.TRUE.equals(responseCompleted)) {
             IOException err = new IOException(NettyUtils.closedChannelMessage(handlerCtx.channel()));
             runAndLogError(handlerCtx.channel(), () -> "Fail to execute SdkAsyncHttpResponseHandler#onError",
                            () -> requestCtx.handler().onError(err));
@@ -514,4 +512,4 @@ public void onComplete() {
             });
         }
     }
-}
+}

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

@@ -25,6 +25,7 @@
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.EXECUTE_FUTURE_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.PROTOCOL_FUTURE;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.REQUEST_CONTEXT_KEY;
+import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.STREAMING_COMPLETE_KEY;
 
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.EmptyByteBuf;
@@ -315,6 +316,20 @@ public void onComplete() {
         verify(channelPool).release(channel);
     }
 
+    @Test
+    public void channelInactive_streamingCompleteButResponseNotComplete_shouldInvokeResponseHandler() {
+        channel.attr(STREAMING_COMPLETE_KEY).set(true);
+
+        nettyResponseHandler.channelInactive(ctx);
+
+        ArgumentCaptor<Throwable> errorCaptor = ArgumentCaptor.forClass(Throwable.class);
+        verify(responseHandler).onError(errorCaptor.capture());
+        assertThat(errorCaptor.getValue()).isInstanceOf(IOException.class);
+        assertThat(executeFuture).isCompletedExceptionally();
+        verify(ctx).close();
+        verify(channelPool).release(channel);
+    }
+
     static final class TestSubscriber implements Subscriber<ByteBuffer> {
 
         private Subscription subscription;