Fix error propogation

Author: alextwoods

services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/internal/S3JavaMultipartTransferProgressListenerTest.java

@@ -449,6 +449,48 @@ void unknownContentLength_multiChunk_allCallbacksFire() {
         Mockito.verify(transferListenerMock, atLeastOnce()).bytesTransferred(ArgumentMatchers.any());
     }
 
+    /**
+     * Verifies that when an unknown-content-length upload fails on the single-chunk path,
+     * the completionFuture completes exceptionally and transferFailed fires.
+     * This guards against regressions where the failure path in uploadInOneChunk does not
+     * propagate the exception to returnFuture, causing the upload to hang indefinitely.
+     */
+    @Test
+    void unknownContentLength_singleChunk_failurePropagates() {
+        S3AsyncClient s3Async = s3AsyncClient(true);
+
+        stubFor(put(urlPathEqualTo("/" + EXAMPLE_BUCKET + "/" + TEST_KEY))
+                    .willReturn(aResponse().withStatus(500).withBody(ERROR_BODY)));
+
+        S3TransferManager tm = new GenericS3TransferManager(s3Async, mock(UploadDirectoryHelper.class),
+                                                            mock(TransferManagerConfiguration.class),
+                                                            mock(DownloadDirectoryHelper.class));
+        CaptureTransferListener transferListener = new CaptureTransferListener();
+        TransferListener transferListenerMock = mock(TransferListener.class);
+
+        BlockingInputStreamAsyncRequestBody body = AsyncRequestBody.forBlockingInputStream(null);
+
+        Upload upload = tm.upload(u -> u.putObjectRequest(p -> p.bucket(EXAMPLE_BUCKET).key(TEST_KEY))
+                                        .requestBody(body)
+                                        .addTransferListener(transferListener)
+                                        .addTransferListener(transferListenerMock)
+                                        .build());
+
+        byte[] data = new byte[1024];
+        body.writeInputStream(new ByteArrayInputStream(data));
+
+        assertThatExceptionOfType(CompletionException.class).isThrownBy(() -> upload.completionFuture().join());
+
+        assertTransferListenerCompletion(transferListener);
+        assertThat(transferListener.isTransferInitiated()).isTrue();
+        assertThat(transferListener.isTransferComplete()).isFalse();
+        assertThat(transferListener.getExceptionCaught()).isNotNull();
+
+        Mockito.verify(transferListenerMock, times(1)).transferInitiated(ArgumentMatchers.any());
+        Mockito.verify(transferListenerMock, times(0)).transferComplete(ArgumentMatchers.any());
+        Mockito.verify(transferListenerMock, timeout(1000).times(1)).transferFailed(ArgumentMatchers.any());
+    }
+
     private static void assertTransferListenerCompletion(CaptureTransferListener transferListener) {
         Duration waitDuration = Duration.ofSeconds(5);
         assertTimeoutPreemptively(

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

@@ -168,14 +168,18 @@ void uploadInOneChunk(PutObjectRequest putObjectRequest,
                                                                                                           asyncRequestBody);
         CompletableFutureUtils.forwardExceptionTo(returnFuture, putObjectResponseCompletableFuture);
 
-        putObjectResponseCompletableFuture.thenAccept(response -> {
+        putObjectResponseCompletableFuture.whenComplete((response, throwable) -> {
+            if (throwable != null) {
+                returnFuture.completeExceptionally(throwable);
+                return;
+            }
             if (reportProgress) {
                 asyncRequestBody.contentLength().ifPresent(progressListener::subscriberOnNext);
             }
             // Always signal completion so that TransferProgressUpdater's endOfStreamFuture completes
-            // and the TransferListener's transferComplete callback fires.
-            // For unknown content length we don't know if it wil lbe one chunk or not ahead of time
-            // and so don't set REPORT_PROGRESS_IN_SINGLE_CHUNK attribute.
+            // and the TransferListener's transferComplete callback fires. This is needed for the
+            // unknown-content-length single-chunk path where splitCloseable() bypasses the wrapper
+            // body that would normally complete endOfStreamFuture via subscriberOnComplete.
             progressListener.subscriberOnComplete();
             returnFuture.complete(response);
         });