Enable zero-copy ByteBuffer publishing in AsyncRequestBody "unsafe" constructors

Author: StephenFlavin

.changes/next-release/feature-AWSSDKforJavav2-5e523a4.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "AWS SDK for Java v2",
+    "contributor": "StephenFlavin",
+    "description": "Enable zero-copy ByteBuffer publishing in AsyncRequestBody via \"unsafe\" constructors"
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/ByteBuffersAsyncRequestBody.java

@@ -96,11 +96,6 @@ public void request(long n) {
                             do {
                                 ByteBuffer buffer = buffers[i];
 
-                                // Pending discussions on https://github.com/aws/aws-sdk-java-v2/issues/3928
-                                if (buffer.isDirect()) {
-                                    buffer = BinaryUtils.toNonDirectBuffer(buffer);
-                                }
-
                                 s.onNext(buffer.asReadOnlyBuffer());
                                 remaining--;
                             } while (remaining > 0 && (i = index.getAndIncrement()) < buffers.length);

core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/ByteBuffersAsyncRequestBodyTest.java

@@ -186,26 +186,6 @@ public void canceledSubscriberDoesNotReturnNewResults() {
         assertTrue(subscriber.publishedResults.isEmpty());
     }
 
-    // Pending discussions on https://github.com/aws/aws-sdk-java-v2/issues/3928
-    @Test
-    public void directBuffersAreCoppiedToNonDirectBuffers() {
-        byte[] bytes = "Hello World!".getBytes(StandardCharsets.UTF_8);
-        ByteBuffer buffer = ByteBuffer.allocateDirect(bytes.length)
-                                      .put(bytes);
-        buffer.flip();
-        AsyncRequestBody requestBody = ByteBuffersAsyncRequestBody.of(buffer);
-
-        TestSubscriber subscriber = new TestSubscriber();
-        requestBody.subscribe(subscriber);
-        subscriber.request(1);
-
-        ByteBuffer publishedBuffer = subscriber.publishedResults.get(0);
-        assertFalse(publishedBuffer.isDirect());
-        byte[] publishedBytes = new byte[publishedBuffer.remaining()];
-        publishedBuffer.get(publishedBytes);
-        assertArrayEquals(bytes, publishedBytes);
-    }
-
     @Test
     public void staticOfByteBufferConstructorSetsLengthBasedOnBufferRemaining() {
         ByteBuffer bb1 = ByteBuffer.allocate(2);