Address feedback

Author: zoewangg

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

@@ -167,7 +167,7 @@ public void onNext(CloseableAsyncRequestBody asyncRequestBody) {
                 NullPointerException exception = new NullPointerException("asyncRequestBody passed to onNext MUST NOT be null.");
                 multipartUploadHelper.failRequestsElegantly(futures,
                                                             exception, uploadId, returnFuture, putObjectRequest);
-                return;
+                throw exception;
             }
 
             if (isDone) {

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

@@ -16,6 +16,7 @@
 package software.amazon.awssdk.services.s3.internal.multipart;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
@@ -164,7 +165,8 @@ void uploadObject_nullAsyncRequestBody_shouldFailRequest() {
 
         Subscription subscription = mock(Subscription.class);
         subscriber.onSubscribe(subscription);
-        subscriber.onNext(null);
+        assertThatThrownBy(() -> subscriber.onNext(null)).isInstanceOf(NullPointerException.class).hasMessageContaining(
+            "asyncRequestBody");
 
         assertThat(future).isCompletedExceptionally();
         future.exceptionally(throwable -> {