Fixed the issue where the position of the ByteBuffer from the request… (#4212)

* Fixed the issue where the position of the ByteBuffer from the request is not honored in NettyNioAsyncHttpClient.

* Add assertion

Author: zoewangg

.changes/next-release/bugfix-NettyNIOHTTPClient-6afb88b.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "Netty NIO HTTP Client",
+    "contributor": "",
+    "description": "Fixed the issue where the position of the ByteBuffer from the request is not honored in NettyNioAsyncHttpClient."
+}

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

@@ -445,7 +445,7 @@ public void onNext(ByteBuffer contentBytes) {
 
                     try {
                         int newLimit = clampedBufferLimit(contentBytes.remaining());
-                        contentBytes.limit(newLimit);
+                        contentBytes.limit(contentBytes.position() + newLimit);
                         ByteBuf contentByteBuf = Unpooled.wrappedBuffer(contentBytes);
                         HttpContent content = new DefaultHttpContent(contentByteBuf);
 

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

@@ -18,14 +18,17 @@
 import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
 import static com.github.tomakehurst.wiremock.client.WireMock.any;
 import static com.github.tomakehurst.wiremock.client.WireMock.equalTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.post;
 import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor;
 import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
 import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
 import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo;
 import static com.github.tomakehurst.wiremock.client.WireMock.verify;
 import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
+import static java.util.Collections.emptyMap;
 import static java.util.Collections.singletonMap;
 import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
+import static org.apache.commons.lang3.RandomStringUtils.randomAscii;
 import static org.apache.commons.lang3.StringUtils.reverse;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -41,6 +44,7 @@
 import static software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClientTestUtils.makeSimpleRequest;
 
 import com.github.tomakehurst.wiremock.WireMockServer;
+import com.github.tomakehurst.wiremock.client.WireMock;
 import com.github.tomakehurst.wiremock.http.Fault;
 import com.github.tomakehurst.wiremock.junit.WireMockRule;
 import io.netty.channel.Channel;
@@ -55,6 +59,7 @@
 import io.netty.util.AttributeKey;
 import java.io.IOException;
 import java.net.URI;
+import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -80,6 +85,7 @@
 import software.amazon.awssdk.http.SdkHttpFullRequest;
 import software.amazon.awssdk.http.SdkHttpMethod;
 import software.amazon.awssdk.http.SdkHttpRequest;
+import software.amazon.awssdk.http.SimpleHttpContentPublisher;
 import software.amazon.awssdk.http.async.AsyncExecuteRequest;
 import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
 import software.amazon.awssdk.http.nio.netty.internal.NettyConfiguration;
@@ -558,6 +564,25 @@ public ChannelFuture close() {
         assertThat(channelClosedFuture.get(5, TimeUnit.SECONDS)).isTrue();
     }
 
+    @Test
+    public void execute_requestByteBufferWithNonZeroPosition_shouldHonor() throws Exception {
+        String body = randomAlphabetic(70);
+        byte[] content = randomAscii(100).getBytes();
+        ByteBuffer requestContent = ByteBuffer.wrap(content);
+        requestContent.position(95);
+        String expected = new String(content, 95, 5);
+
+        URI uri = URI.create("http://localhost:" + mockServer.port());
+        stubFor(post(urlPathEqualTo("/"))
+                    .withRequestBody(equalTo(expected)).willReturn(aResponse().withBody(body)));
+        SdkHttpRequest request = createRequest(uri, "/", expected, SdkHttpMethod.POST, emptyMap());
+        RecordingResponseHandler recorder = new RecordingResponseHandler();
+
+        client.execute(AsyncExecuteRequest.builder().request(request).requestContentPublisher(new SimpleHttpContentPublisher(requestContent)).responseHandler(recorder).build());
+        recorder.completeFuture.get(5, TimeUnit.SECONDS);
+        verify(postRequestedFor(urlPathEqualTo("/")).withRequestBody(equalTo(expected)));
+    }
+
     // Needs to be a non-anon class in order to spy
     public static class CustomThreadFactory implements ThreadFactory {
         @Override

test/http-client-tests/src/main/java/software/amazon/awssdk/http/SimpleHttpContentPublisher.java

@@ -32,20 +32,24 @@
 @SdkInternalApi
 public final class SimpleHttpContentPublisher implements SdkHttpContentPublisher {
 
-    private final byte[] content;
+    private final ByteBuffer content;
     private final int length;
 
     public SimpleHttpContentPublisher(SdkHttpFullRequest request) {
-        this.content = request.contentStreamProvider().map(p -> invokeSafely(() -> IoUtils.toByteArray(p.newStream())))
-                                                      .orElseGet(() -> new byte[0]);
-        this.length = content.length;
+        this.content =
+            request.contentStreamProvider().map(p -> invokeSafely(() -> ByteBuffer.wrap(IoUtils.toByteArray(p.newStream()))))
+                                                      .orElseGet(() -> ByteBuffer.wrap(new byte[0]));
+        this.length = content.remaining();
     }
 
     public SimpleHttpContentPublisher(byte[] body) {
-        this.content = body;
-        this.length = body.length;
+        this(ByteBuffer.wrap(body));
     }
 
+    public SimpleHttpContentPublisher(ByteBuffer byteBuffer) {
+        this.content = byteBuffer;
+        this.length = byteBuffer.remaining();
+    }
 
     @Override
     public Optional<Long> contentLength() {
@@ -72,7 +76,7 @@ public void request(long n) {
                 if (n <= 0) {
                     s.onError(new IllegalArgumentException("Demand must be positive"));
                 } else {
-                    s.onNext(ByteBuffer.wrap(content));
+                    s.onNext(content);
                     s.onComplete();
                 }
             }