Prevent multipart download from hanging (#5466)

* remove cancel lock in downloader subscriber

* re-add synchronized block

Author: L-Applin

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

@@ -123,12 +123,10 @@ private SplittingTransformer(AsyncResponseTransformer<ResponseT, ResultT> upstre
             maximumBufferSizeInBytes, "maximumBufferSizeInBytes");
 
         this.resultFuture.whenComplete((r, e) -> {
-            log.trace(() -> "result future whenComplete");
             if (e == null) {
                 return;
             }
             if (isCancelled.compareAndSet(false, true)) {
-                log.trace(() -> "result future whenComplete with isCancelled=true");
                 handleFutureCancel(e);
             }
         });
@@ -171,7 +169,6 @@ public void request(long n) {
         public void cancel() {
             log.trace(() -> String.format("received cancel signal. Current cancel state is 'isCancelled=%s'", isCancelled.get()));
             if (isCancelled.compareAndSet(false, true)) {
-                log.trace(() -> "Cancelling splitting transformer");
                 handleSubscriptionCancel();
             }
         }
@@ -221,20 +218,17 @@ private void handleSubscriptionCancel() {
             }
             if (!onStreamCalled.get()) {
                 // we never subscribe publisherToUpstream to the upstream, it would not complete
-                log.trace(() -> "publisherToUpstream never subscribed, skipping downstreamSubscriber.onComplete()");
                 downstreamSubscriber = null;
                 return;
             }
-            log.trace(() -> "publisherToUpstream.complete()");
             publisherToUpstream.complete().whenComplete((v, t) -> {
                 if (downstreamSubscriber == null) {
-                    log.trace(() -> "[in future] downstreamSubscriber already null, skipping downstreamSubscriber.onComplete()");
                     return;
                 }
                 if (t != null) {
                     downstreamSubscriber.onError(t);
                 } else {
-                    log.trace(() -> "[in future] calling downstreamSubscriber.onComplete");
+                    log.trace(() -> "calling downstreamSubscriber.onComplete()");
                     downstreamSubscriber.onComplete();
                 }
                 downstreamSubscriber = null;
@@ -289,7 +283,6 @@ public CompletableFuture<ResponseT> prepare() {
             });
             individualFuture.whenComplete((r, e) -> {
                 if (isCancelled.get()) {
-                    log.trace(() -> "Individual future completed .");
                     handleSubscriptionCancel();
                 }
             });

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartDownloaderSubscriber.java

@@ -69,13 +69,16 @@ public class MultipartDownloaderSubscriber implements Subscriber<AsyncResponseTr
      */
     private final CompletableFuture<Void> future = new CompletableFuture<>();
 
-    private final Object lock = new Object();
-
     /**
      * The etag of the object being downloaded.
      */
     private volatile String eTag;
 
+    /**
+     * The Subscription lock
+     */
+    private final Object lock = new Object();
+
     public MultipartDownloaderSubscriber(S3AsyncClient s3, GetObjectRequest getObjectRequest) {
         this(s3, getObjectRequest, 0);
     }
@@ -88,7 +91,6 @@ public MultipartDownloaderSubscriber(S3AsyncClient s3, GetObjectRequest getObjec
 
     @Override
     public void onSubscribe(Subscription s) {
-        log.trace(() -> "onSubscribe");
         if (this.subscription != null) {
             s.cancel();
             return;
@@ -99,20 +101,19 @@ public void onSubscribe(Subscription s) {
 
     @Override
     public void onNext(AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> asyncResponseTransformer) {
-        log.trace(() -> String.format("onNext, completed part = %d", completedParts.get()));
         if (asyncResponseTransformer == null) {
             subscription.cancel();
             throw new NullPointerException("onNext must not be called with null asyncResponseTransformer");
         }
 
         int nextPartToGet = completedParts.get() + 1;
 
-        if (totalParts != null && nextPartToGet > totalParts) {
-            synchronized (lock) {
-                logMulitpartComplete(totalParts);
+        synchronized (lock) {
+            if (totalParts != null && nextPartToGet > totalParts) {
+                log.debug(() -> String.format("Completing multipart download after a total of %d parts downloaded.", totalParts));
                 subscription.cancel();
+                return;
             }
-            return;
         }
 
         GetObjectRequest actualRequest = nextRequest(nextPartToGet);
@@ -128,10 +129,6 @@ public void onNext(AsyncResponseTransformer<GetObjectResponse, GetObjectResponse
         });
     }
 
-    private void logMulitpartComplete(int totalParts) {
-        log.debug(() -> String.format("Completing multipart download after a total of %d parts downloaded.", totalParts));
-    }
-
     private void requestMoreIfNeeded(GetObjectResponse response) {
         int totalComplete = completedParts.incrementAndGet();
         MultipartDownloadUtils.multipartDownloadResumeContext(getObjectRequest)
@@ -153,14 +150,15 @@ private void requestMoreIfNeeded(GetObjectResponse response) {
         if (partCount != null && totalParts == null) {
             log.debug(() -> String.format("Total amount of parts of the object to download: %d", partCount));
             MultipartDownloadUtils.multipartDownloadResumeContext(getObjectRequest)
-                .ifPresent(ctx -> ctx.totalParts(partCount));
+                                  .ifPresent(ctx -> ctx.totalParts(partCount));
             totalParts = partCount;
         }
+
         synchronized (lock) {
             if (totalParts != null && totalParts > 1 && totalComplete < totalParts) {
                 subscription.request(1);
             } else {
-                logMulitpartComplete(totalParts);
+                log.debug(() -> String.format("Completing multipart download after a total of %d parts downloaded.", totalParts));
                 subscription.cancel();
             }
         }