test(s3): add regression test for onError cause preservation

Add ParallelMultipartDownloaderSubscriberTest verifying that onError completes
the result future with the original throwable (not a CancellationException) and
still cancels in-flight part requests. The test fails against the previous
cancel-before-complete ordering. Restore the original onError comments.

Author: Rishabh Jain

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

@@ -412,6 +412,9 @@ private boolean shouldProcessPendingTransformers() {
 
     @Override
     public void onError(Throwable t) {
+        // Signal received from the publisher this is subscribed to
+        // (in the case of file download, that's FileAsyncResponseTransformerPublisher)
+        // Failed state, something really wrong has happened, cancel everything
         // Complete the result future with the original cause before cancelling in-flight parts.
         // Cancelling first races a CancellationException onto resultFuture and masks t.
         log.debug(() -> "Error in parallel multipart download", t);

services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/multipart/ParallelMultipartDownloaderSubscriberTest.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.services.s3.internal.multipart;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.reactivestreams.Subscription;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+
+/**
+ * Unit tests for {@link ParallelMultipartDownloaderSubscriber}.
+ */
+class ParallelMultipartDownloaderSubscriberTest {
+
+    private static final int MAX_IN_FLIGHT_PARTS = 10;
+
+    private S3AsyncClient s3;
+    private CompletableFuture<GetObjectResponse> resultFuture;
+    private ParallelMultipartDownloaderSubscriber subscriber;
+
+    @BeforeEach
+    void setup() {
+        s3 = mock(S3AsyncClient.class);
+        resultFuture = new CompletableFuture<>();
+        GetObjectRequest request = GetObjectRequest.builder()
+                                                   .bucket("test-bucket")
+                                                   .key("test-key")
+                                                   .build();
+        subscriber = new ParallelMultipartDownloaderSubscriber(s3, request, resultFuture, MAX_IN_FLIGHT_PARTS);
+    }
+
+    @Test
+    void onError_withInFlightRequests_shouldCompleteResultFutureWithOriginalCause() {
+        CompletableFuture<GetObjectResponse> firstPartFuture = new CompletableFuture<>();
+        CompletableFuture<GetObjectResponse> secondPartFuture = new CompletableFuture<>();
+        when(s3.getObject(any(GetObjectRequest.class), any(AsyncResponseTransformer.class)))
+            .thenReturn(firstPartFuture, secondPartFuture);
+
+        subscriber.onSubscribe(mock(Subscription.class));
+        subscriber.onNext(mock(AsyncResponseTransformer.class));
+        firstPartFuture.complete(GetObjectResponse.builder()
+                                                  .partsCount(3)
+                                                  .eTag("etag")
+                                                  .build());
+        // Second part is now in flight and never completes on its own.
+        subscriber.onNext(mock(AsyncResponseTransformer.class));
+
+        RuntimeException cause = new RuntimeException("original failure");
+        subscriber.onError(cause);
+
+        assertThat(resultFuture).isCompletedExceptionally();
+        Throwable thrown = resultFuture.handle((r, t) -> t).join();
+        assertThat(thrown).isNotInstanceOf(CancellationException.class);
+        assertThat(thrown).isSameAs(cause);
+    }
+
+    @Test
+    void onError_withoutInFlightRequests_shouldCompleteResultFutureWithOriginalCause() {
+        RuntimeException cause = new RuntimeException("original failure");
+        subscriber.onError(cause);
+
+        assertThat(resultFuture).isCompletedExceptionally();
+        Throwable thrown = resultFuture.handle((r, t) -> t).join();
+        assertThat(thrown).isSameAs(cause);
+    }
+
+    @Test
+    void onError_shouldCancelInFlightRequests() {
+        CompletableFuture<GetObjectResponse> firstPartFuture = new CompletableFuture<>();
+        CompletableFuture<GetObjectResponse> secondPartFuture = new CompletableFuture<>();
+        when(s3.getObject(any(GetObjectRequest.class), any(AsyncResponseTransformer.class)))
+            .thenReturn(firstPartFuture, secondPartFuture);
+
+        subscriber.onSubscribe(mock(Subscription.class));
+        subscriber.onNext(mock(AsyncResponseTransformer.class));
+        firstPartFuture.complete(GetObjectResponse.builder()
+                                                  .partsCount(3)
+                                                  .eTag("etag")
+                                                  .build());
+        subscriber.onNext(mock(AsyncResponseTransformer.class));
+
+        subscriber.onError(new RuntimeException("original failure"));
+
+        assertThat(secondPartFuture).isCompletedExceptionally();
+    }
+}