Fix bug in FileAsyncRequestBody where inflight parts were negative (#6564)

* Fix bug in FileAsyncRequestBody where inflight parts were negative

* cleanups

* Update test

* Alternative approach

* Cleanup/rename

* Fix spotbugs

Author: alextwoods

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

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "Amazon S3",
+    "contributor": "",
+    "description": "Fix bug in S3 Multipart uploads with FileAsyncRequestBody - ensure that concurrency is limited correctly by bufferSizeInBytes"
+}

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

@@ -166,6 +166,7 @@ private final class FileAsyncRequestBodyWrapper implements AsyncRequestBody {
 
         private final FileAsyncRequestBody fileAsyncRequestBody;
         private final SimplePublisher<AsyncRequestBody> simplePublisher;
+        private final AtomicBoolean hasCompleted = new AtomicBoolean(false);
 
         FileAsyncRequestBodyWrapper(FileAsyncRequestBody fileAsyncRequestBody,
                                     SimplePublisher<AsyncRequestBody> simplePublisher) {
@@ -175,16 +176,22 @@ private final class FileAsyncRequestBodyWrapper implements AsyncRequestBody {
 
         @Override
         public void subscribe(Subscriber<? super ByteBuffer> s) {
-            fileAsyncRequestBody.doAfterOnComplete(() -> startNextRequestBody(simplePublisher))
+            fileAsyncRequestBody.doAfterOnComplete(this::startNextIfNeeded)
                                 // The reason we still need to call startNextRequestBody when the subscription is
                                 // cancelled is that upstream could cancel the subscription even though the stream has
                                 // finished successfully before onComplete. If this happens, doAfterOnComplete callback
                                 // will never be invoked, and if the current buffer is full, the publisher will stop
                                 // sending new FileAsyncRequestBody, leading to uncompleted future.
-                                .doAfterOnCancel(() -> startNextRequestBody(simplePublisher))
+                                .doAfterOnCancel(this::startNextIfNeeded)
                                 .subscribe(s);
         }
 
+        private void startNextIfNeeded() {
+            if (hasCompleted.compareAndSet(false, true)) {
+                startNextRequestBody(simplePublisher);
+            }
+        }
+
         @Override
         public Optional<Long> contentLength() {
             return fileAsyncRequestBody.contentLength();

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

@@ -92,7 +92,7 @@ private static Runnable verifyConcurrentRequests(FileAsyncRequestBodySplitHelper
             if (concurrency > maxConcurrency.get()) {
                 maxConcurrency.set(concurrency);
             }
-            assertThat(helper.numAsyncRequestBodiesInFlight()).hasValueLessThan(10);
+            assertThat(helper.numAsyncRequestBodiesInFlight()).hasValueBetween(0,10);
         };
     }
 }