Do not add chunk data for empty ByteBuffer (#5320)

* Do not add chunk data for empty ByteBuffer

* Do not add chunk data for empty ByteBuffer

* Add unit test

* Add changelog

* Remove integ test

Author: davidh44

.changes/next-release/bugfix-AmazonS3-e33d66a.json

@@ -0,0 +1,6 @@
+{
+    "category": "Amazon S3", 
+    "contributor": "", 
+    "type": "bugfix", 
+    "description": "Fixes bug where empty non-final chunk is wrapped with headers and trailers during PutObject when using flexible checksums with S3AsyncClient"
+}

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

@@ -197,9 +197,11 @@ public void onNext(ByteBuffer byteBuffer) {
                     checksumBytes = checksum.getChecksumBytes();
                     ByteBuffer allocatedBuffer = getFinalChecksumAppendedChunk(byteBuffer);
                     wrapped.onNext(allocatedBuffer);
-                } else {
+                } else if (byteBuffer.hasRemaining()) {
                     ByteBuffer allocatedBuffer = createChunk(byteBuffer, false);
                     wrapped.onNext(allocatedBuffer);
+                } else {
+                    wrapped.onNext(byteBuffer);
                 }
             } catch (SdkException sdkException) {
                 this.subscription.cancel();

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

@@ -22,6 +22,7 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -37,6 +38,7 @@
 import org.junit.jupiter.params.provider.MethodSource;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
 import software.amazon.awssdk.core.async.AsyncRequestBody;
 import software.amazon.awssdk.core.checksums.Algorithm;
 import software.amazon.awssdk.core.internal.util.Mimetype;
@@ -97,7 +99,10 @@ private static Stream<Arguments> publishers() {
                          expectedEmptyString),
             Arguments.of("RequestBody from string, random pos, empty string",
                          checksumPublisher(AsyncRequestBody.fromRemainingByteBufferUnsafe(nonPosZeroByteBuffer(emptyString))),
-                         expectedEmptyString));
+                         expectedEmptyString),
+            Arguments.of("EmptyBufferPublisher, test string",
+                         checksumPublisher(new EmptyBufferPublisher(testString)),
+                         expectedTestString));
     }
 
     private static ChecksumCalculatingAsyncRequestBody checksumPublisher(AsyncRequestBody sourcePublisher) {
@@ -274,4 +279,40 @@ public void fromBytes_byteArrayNotNullChecksumSupplied() {
         ByteBuffer publishedBb = Flowable.fromPublisher(body).toList().blockingGet().get(0);
         assertThat(BinaryUtils.copyAllBytesFrom(publishedBb)).isEqualTo(expected);
     }
+
+    static class EmptyBufferPublisher implements AsyncRequestBody {
+
+        private final ByteBuffer[] buffers = new ByteBuffer[2];
+        private final String payload;
+
+        EmptyBufferPublisher(String payload) {
+            buffers[0] = ByteBuffer.wrap(new byte[0]);
+            buffers[1] = ByteBuffer.wrap(payload.getBytes(StandardCharsets.UTF_8));
+            this.payload = payload;
+        }
+
+        @Override
+        public void subscribe(Subscriber<? super ByteBuffer> subscriber) {
+            subscriber.onSubscribe(new Subscription() {
+                private int count = 0;
+
+                @Override
+                public void request(long n) {
+                    if (count < 2) {
+                        subscriber.onNext(buffers[count++]);
+                    } else {
+                        subscriber.onComplete();
+                    }
+                }
+
+                @Override
+                public void cancel() {}
+            });
+        }
+
+        @Override
+        public Optional<Long> contentLength() {
+            return Optional.of((long) payload.length());
+        }
+    }
 }