Add trace logs for debugging multipart download (#5434)

* testing without DelegatingBufferingSubscriber + added logs

* more logs

Author: L-Applin

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

@@ -249,11 +249,14 @@ public void onError(Throwable t) {
 
         @Override
         public void onComplete() {
+            log.trace(() -> "onComplete");
             // if write in progress, tell write to close on finish.
             synchronized (this) {
                 if (writeInProgress) {
+                    log.trace(() -> "writeInProgress = true, not closing");
                     closeOnLastWrite = true;
                 } else {
+                    log.trace(() -> "writeInProgress = false, closing");
                     close();
                 }
             }
@@ -264,6 +267,7 @@ private void close() {
                 if (fileChannel != null) {
                     invokeSafely(fileChannel::close);
                 }
+                log.trace(() -> "Completing File async transformer future future");
                 future.complete(null);
             } catch (RuntimeException exception) {
                 future.completeExceptionally(exception);

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

@@ -123,10 +123,12 @@ 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);
             }
         });
@@ -167,6 +169,7 @@ public void request(long n) {
 
         @Override
         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();
@@ -207,26 +210,31 @@ private boolean doEmit() {
     /**
      * Handle the {@code .cancel()} signal received from the downstream subscription. Data that is being sent to the upstream
      * transformer need to finish processing before we complete. One typical use case for this is completing the multipart
-     * download, the subscriber having reached the final part will signal that it doesn't need more parts by calling {@code
-     * .cancel()} on the subscription.
+     * download, the subscriber having reached the final part will signal that it doesn't need more parts by calling
+     * {@code .cancel()} on the subscription.
      */
     private void handleSubscriptionCancel() {
         synchronized (cancelLock) {
             if (downstreamSubscriber == null) {
+                log.trace(() -> "downstreamSubscriber already null, skipping downstreamSubscriber.onComplete()");
                 return;
             }
             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");
                     downstreamSubscriber.onComplete();
                 }
                 downstreamSubscriber = null;
@@ -236,8 +244,8 @@ private void handleSubscriptionCancel() {
 
     /**
      * Handle when the {@link SplittingTransformer#resultFuture} is cancelled or completed exceptionally from the outside. Data
-     * need to stop being sent to the upstream transformer immediately. One typical use case for this is transfer manager
-     * needing to pause download by calling {@code .cancel(true)} on the future.
+     * need to stop being sent to the upstream transformer immediately. One typical use case for this is transfer manager needing
+     * to pause download by calling {@code .cancel(true)} on the future.
      *
      * @param e The exception the future was complete exceptionally with.
      */
@@ -281,6 +289,7 @@ public CompletableFuture<ResponseT> prepare() {
             });
             individualFuture.whenComplete((r, e) -> {
                 if (isCancelled.get()) {
+                    log.trace(() -> "Individual future completed .");
                     handleSubscriptionCancel();
                 }
             });
@@ -308,8 +317,8 @@ public void onStream(SdkPublisher<ByteBuffer> publisher) {
                         DelegatingBufferingSubscriber.builder()
                                                      .maximumBufferInBytes(maximumBufferInBytes)
                                                      .delegate(upstreamSubscriber)
-                                                     .build()
-                    ));
+                                                     .build())
+                    );
                 }
             }
             publisher.subscribe(new IndividualPartSubscriber<>(this.individualFuture, response));

services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/GenericS3TransferManager.java

@@ -362,6 +362,13 @@ public final <ResultT> Download<ResultT> download(DownloadRequest<ResultT> downl
                                                               r -> CompletedDownload.builder()
                                                                                     .result(r)
                                                                                     .build());
+            future.whenComplete((r, e) -> {
+                if (e != null) {
+                    log.error(() -> String.format("s3 client future completed exceptionally"), e);
+                } else {
+                    log.trace(() -> String.format("s3 client future completed : %s", r));
+                }
+            });
         } catch (Throwable throwable) {
             returnFuture.completeExceptionally(throwable);
         }

services-custom/s3-transfer-manager/src/test/resources/log4j2.properties

@@ -23,6 +23,12 @@ appender.console.layout.pattern = %d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %
 rootLogger.level = info
 rootLogger.appenderRef.stdout.ref = ConsoleAppender
 
+logger.split.name = software.amazon.awssdk.core.internal.async
+logger.split.level = trace
+
+logger.multi.name = software.amazon.awssdk.services.s3.internal.multipart
+logger.multi.level = trace
+
 # Uncomment below to enable more specific logging
 #
 #logger.sdk.name = software.amazon.awssdk