Close and release channel upon validation failure (#3855)

When content length validation fails when PublisherAdapter#onComplete is
invoked, behave as on PublisherAdapter#onError and also close and release the
channel; otherwise the channel would be left in the leased state forever.

.changes/next-release/bugfix-NettyNIOHTTPClient-7b511ce.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "Netty NIO HTTP Client",
+    "contributor": "",
+    "description": "Fix a bug where, if validation of of the amount of expected data to be received (HTTP `Content-Length`) fails, the connection would be left dangling, consuming a connection from the pool until the client is shut down."
+}

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

@@ -376,6 +376,8 @@ public void onComplete() {
                         }
                     } catch (IOException e) {
                         notifyError(e);
+                        runAndLogError(channelContext.channel(), () -> "Could not release channel back to the pool",
+                                       () -> closeAndRelease(channelContext));
                     }
                 }
 

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

@@ -235,6 +235,56 @@ public void onComplete() {
         }
     }
 
+    @Test
+    public void contentLengthValidationFails_closesAndReleasesConnection() {
+        channel.attr(ChannelAttributeKey.RESPONSE_CONTENT_LENGTH).set(1L);
+        channel.attr(ChannelAttributeKey.RESPONSE_DATA_READ).set(0L);
+
+        Publisher<HttpContent> publisher = subscriber -> subscriber.onSubscribe(new Subscription() {
+            @Override
+            public void request(long l) {
+                subscriber.onComplete();
+            }
+
+            @Override
+            public void cancel() {
+            }
+        });
+
+        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) {
+            }
+
+            @Override
+            public void onError(Throwable throwable) {
+            }
+
+            @Override
+            public void onComplete() {
+            }
+        };
+
+        ResponseHandler.PublisherAdapter publisherAdapter = new ResponseHandler.PublisherAdapter(streamedResponse, ctx,
+                                                                                                 requestContext, executeFuture);
+
+        publisherAdapter.subscribe(subscriber);
+
+        verify(ctx).close();
+        verify(channelPool).release(channel);
+    }
+
     static final class TestSubscriber implements Subscriber<ByteBuffer> {
 
         private Subscription subscription;