Released ignored HTTP content in Netty client

When cancelling a subscrption, it's possible onNext() is still invoked by the
publisher if it processes signals asynchronously. While the adapter ignores the
content rather forwarding it to the downstream subscriber, it leaves a dangling
reference on the input HttpCotent. Fix this issue by decrementing the reference
before returning.

Author: dagnir

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

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "Netty NIO HTTP Client",
+    "contributor": "",
+    "description": "Fix a bug where the Netty HTTP client can leak memory when a response stream is cancelled prematurely but the upstream publisher continues to invoke onNext for some time before stopping. Fixes [#2051](https://github.com/aws/aws-sdk-java-v2/issues/2051)."
+}

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

@@ -38,6 +38,7 @@
 import io.netty.handler.codec.http.HttpUtil;
 import io.netty.handler.timeout.ReadTimeoutException;
 import io.netty.handler.timeout.WriteTimeoutException;
+import io.netty.util.ReferenceCountUtil;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.List;
@@ -251,6 +252,7 @@ private void onCancel() {
                 public void onNext(HttpContent httpContent) {
                     // isDone may be true if the subscriber cancelled
                     if (isDone.get()) {
+                        ReferenceCountUtil.release(httpContent);
                         return;
                     }
 

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

@@ -20,13 +20,15 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 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 io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.EmptyByteBuf;
+import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.EventLoopGroup;
 import io.netty.handler.codec.http.DefaultHttpContent;
@@ -43,6 +45,7 @@
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
+import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import software.amazon.awssdk.http.Protocol;
@@ -155,6 +158,75 @@ public void errorOccurred_shouldInvokeResponseHandler() {
         verify(responseHandler).onError(exception);
     }
 
+    @Test
+    public void subscriptionCancelled_upstreamPublisherCallsOnNext_httpContentReleased() {
+        HttpContent firstContent = mock(HttpContent.class);
+        when(firstContent.content()).thenReturn(Unpooled.EMPTY_BUFFER);
+
+        HttpContent[] contentToIgnore = new HttpContent[8];
+        for (int i = 0; i < contentToIgnore.length; ++i) {
+            contentToIgnore[i] = mock(HttpContent.class);
+            when(contentToIgnore[i].content()).thenReturn(Unpooled.EMPTY_BUFFER);
+        }
+
+        Publisher<HttpContent> publisher = subscriber -> subscriber.onSubscribe(new Subscription() {
+            @Override
+            public void request(long l) {
+                // We ignore any cancel signal and just publish all the content
+                subscriber.onNext(firstContent);
+
+                for (int i = 0; i < l && i < contentToIgnore.length; ++i) {
+                    subscriber.onNext(contentToIgnore[i]);
+                }
+            }
+
+            @Override
+            public void cancel() {
+                // no-op
+            }
+        });
+
+        DefaultStreamedHttpResponse streamedResponse = new DefaultStreamedHttpResponse(HttpVersion.HTTP_1_1,
+                HttpResponseStatus.OK, publisher);
+
+        Subscriber<ByteBuffer> subscriber = new Subscriber<ByteBuffer>() {
+            private Subscription subscription;
+
+            @Override
+            public void onSubscribe(Subscription subscription) {
+                this.subscription = subscription;
+                subscription.request(Long.MAX_VALUE);
+            }
+
+            @Override
+            public void onNext(ByteBuffer byteBuffer) {
+                subscription.cancel();
+            }
+
+            @Override
+            public void onError(Throwable throwable) {
+            }
+
+            @Override
+            public void onComplete() {
+            }
+        };
+
+        ResponseHandler.PublisherAdapter publisherAdapter = new ResponseHandler.PublisherAdapter(streamedResponse, ctx,
+                requestContext, executeFuture);
+
+        publisherAdapter.subscribe(subscriber);
+
+        // First one should be accessed as normal
+        verify(firstContent).content();
+        verify(firstContent).release();
+
+        for (int i = 0; i < contentToIgnore.length; ++i) {
+            verify(contentToIgnore[i]).release();
+            verifyNoMoreInteractions(contentToIgnore[i]);
+        }
+    }
+
     static final class TestSubscriber implements Subscriber<ByteBuffer> {
 
         private Subscription subscription;