Minor cleanups + add changelog

Author: alextwoods

.changes/next-release/feature-AmazonS3-d7fcc2e.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "Amazon S3",
+    "contributor": "",
+    "description": "Added BufferedSplittableAsyncRequestBody.builder() with fullBufferingEnabled option that fully buffers each multipart upload part before sending, fixing NonRetryableException when retrying parts from slow streaming sources."
+}

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

@@ -34,11 +34,19 @@
  * If the first subscriber fails to consume all the data (e.g., due to early cancellation or errors),
  * subsequent retry attempts will fail since the complete data set is not available for resubscription.</p>
  *
- * <p><b>Usage Example:</b></p>
+ * <p><b>Usage Examples:</b></p>
  * {@snippet :
+ * // Simple usage (default behavior, immediate send):
  * AsyncRequestBody originalBody = AsyncRequestBody.fromString("Hello World");
  * BufferedSplittableAsyncRequestBody retryableBody =
  *     BufferedSplittableAsyncRequestBody.create(originalBody);
+ *
+ * // With full buffering enabled for slow streaming sources:
+ * BufferedSplittableAsyncRequestBody fullBufferedBody =
+ *     BufferedSplittableAsyncRequestBody.builder()
+ *         .asyncRequestBody(originalBody)
+ *         .fullBufferingEnabled(true)
+ *         .build();
  * }
  *
  * <p><b>Performance Considerations:</b></p>
@@ -56,44 +64,33 @@ public final class BufferedSplittableAsyncRequestBody implements AsyncRequestBod
     private final AsyncRequestBody delegate;
     private final boolean fullBufferingEnabled;
 
-    private BufferedSplittableAsyncRequestBody(AsyncRequestBody delegate, boolean fullBufferingEnabled) {
-        this.delegate = delegate;
-        this.fullBufferingEnabled = fullBufferingEnabled;
+    private BufferedSplittableAsyncRequestBody(Builder builder) {
+        this.delegate = Validate.paramNotNull(builder.asyncRequestBody, "asyncRequestBody");
+        this.fullBufferingEnabled = builder.fullBufferingEnabled;
     }
 
     /**
      * Creates a new {@link BufferedSplittableAsyncRequestBody} that wraps the provided {@link AsyncRequestBody}.
      *
      * <p>Full buffering is disabled by default. Each part is sent to the downstream subscriber immediately
-     * upon initialization in the known-content-length path (existing behavior).
+     * upon initialization in the known-content-length path.
      *
      * @param delegate the {@link AsyncRequestBody} to wrap and make retryable. Must not be null.
      * @return a new {@link BufferedSplittableAsyncRequestBody} instance
      * @throws NullPointerException if delegate is null
      */
     public static BufferedSplittableAsyncRequestBody create(AsyncRequestBody delegate) {
         Validate.paramNotNull(delegate, "delegate");
-        return new BufferedSplittableAsyncRequestBody(delegate, false);
+        return builder().asyncRequestBody(delegate).build();
     }
 
     /**
-     * Creates a new {@link BufferedSplittableAsyncRequestBody} that wraps the provided {@link AsyncRequestBody},
-     * with an option to enable full buffering before sending parts downstream.
-     *
-     * <p>When {@code fullBufferingEnabled} is {@code true}, each part is fully buffered before being sent to the
-     * downstream subscriber. This guarantees that the retry buffer is always populated before the HTTP layer
-     * subscribes, making per-part retry deterministically successful for slow streaming sources (e.g., SFTP).
+     * Returns a new {@link Builder} for creating a {@link BufferedSplittableAsyncRequestBody} with configuration options.
      *
-     * <p>When {@code fullBufferingEnabled} is {@code false}, behavior is identical to {@link #create(AsyncRequestBody)}.
-     *
-     * @param delegate the {@link AsyncRequestBody} to wrap and make retryable. Must not be null.
-     * @param fullBufferingEnabled whether to enable full buffering before sending parts downstream
-     * @return a new {@link BufferedSplittableAsyncRequestBody} instance
-     * @throws NullPointerException if delegate is null
+     * @return a new builder instance
      */
-    public static BufferedSplittableAsyncRequestBody create(AsyncRequestBody delegate, boolean fullBufferingEnabled) {
-        Validate.paramNotNull(delegate, "delegate");
-        return new BufferedSplittableAsyncRequestBody(delegate, fullBufferingEnabled);
+    public static Builder builder() {
+        return new Builder();
     }
 
     @Override
@@ -136,4 +133,55 @@ public void subscribe(Subscriber<? super ByteBuffer> s) {
     public String body() {
         return delegate.body();
     }
+
+    /**
+     * Builder for {@link BufferedSplittableAsyncRequestBody}.
+     */
+    public static final class Builder {
+        private AsyncRequestBody asyncRequestBody;
+        private boolean fullBufferingEnabled = false;
+
+        private Builder() {
+        }
+
+        /**
+         * Sets the {@link AsyncRequestBody} to wrap and make retryable.
+         *
+         * @param asyncRequestBody the request body to wrap. Must not be null.
+         * @return this builder for method chaining
+         */
+        public Builder asyncRequestBody(AsyncRequestBody asyncRequestBody) {
+            this.asyncRequestBody = asyncRequestBody;
+            return this;
+        }
+
+        /**
+         * Configures whether to fully buffer each part before sending it downstream.
+         *
+         * <p>When enabled, each part is fully buffered before being sent to the downstream subscriber.
+         * This guarantees that the retry buffer is always populated before the HTTP layer subscribes,
+         * making per-part retry deterministically successful for slow streaming sources (e.g., SFTP).
+         *
+         * <p>When disabled (the default), each part is sent immediately upon initialization in the
+         * known-content-length path, allowing the HTTP connection to open while data is still arriving.
+         *
+         * @param fullBufferingEnabled whether to enable full buffering before sending parts downstream.
+         *                             Defaults to {@code false}.
+         * @return this builder for method chaining
+         */
+        public Builder fullBufferingEnabled(boolean fullBufferingEnabled) {
+            this.fullBufferingEnabled = fullBufferingEnabled;
+            return this;
+        }
+
+        /**
+         * Builds a new {@link BufferedSplittableAsyncRequestBody} instance.
+         *
+         * @return a new instance configured by this builder
+         * @throws NullPointerException if asyncRequestBody is null
+         */
+        public BufferedSplittableAsyncRequestBody build() {
+            return new BufferedSplittableAsyncRequestBody(this);
+        }
+    }
 }

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

@@ -36,6 +36,8 @@
  *
  * <p>If content length is known, each {@link AsyncRequestBody} is sent to the subscriber right after it's initialized.
  * Otherwise, it is sent after the entire content for that chunk is buffered. This is required to get content length.
+ * When {@code fullBufferingEnabled} is set to {@code true}, the known-content-length path also defers sending until the
+ * part is fully buffered, guaranteeing that the retry buffer is populated before the downstream subscriber receives it.
  */
 @SdkInternalApi
 public class SplittingPublisher implements SdkPublisher<CloseableAsyncRequestBody> {

core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/SplittingPublisherTest.java

@@ -335,8 +335,11 @@ public void cancel() {
             }
         };
 
-        // Use create(body, true) to enable full buffering
-        BufferedSplittableAsyncRequestBody bufferedBody = BufferedSplittableAsyncRequestBody.create(sourceBody, true);
+        // Use builder to enable full buffering
+        BufferedSplittableAsyncRequestBody bufferedBody = BufferedSplittableAsyncRequestBody.builder()
+                .asyncRequestBody(sourceBody)
+                .fullBufferingEnabled(true)
+                .build();
 
         // Verify that content length is propagated
         assertThat(bufferedBody.contentLength()).hasValue((long) data.length);
@@ -457,7 +460,10 @@ void bufferedSplittable_createWithFullBufferingTrue_partsAreRetryable() throws E
         }
         AsyncRequestBody sourceBody = AsyncRequestBody.fromBytes(data);
 
-        BufferedSplittableAsyncRequestBody bufferedBody = BufferedSplittableAsyncRequestBody.create(sourceBody, true);
+        BufferedSplittableAsyncRequestBody bufferedBody = BufferedSplittableAsyncRequestBody.builder()
+                .asyncRequestBody(sourceBody)
+                .fullBufferingEnabled(true)
+                .build();
 
         AsyncRequestBodySplitConfiguration splitConfig = AsyncRequestBodySplitConfiguration.builder()
                 .chunkSizeInBytes(5L)

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

@@ -302,7 +302,10 @@ public void cancel() {
 
         // Wrap with full buffering enabled — this ensures retry buffer is populated before HTTP subscribe
         BufferedSplittableAsyncRequestBody bufferedBody =
-            BufferedSplittableAsyncRequestBody.create(slowStreamingBody, true);
+            BufferedSplittableAsyncRequestBody.builder()
+                .asyncRequestBody(slowStreamingBody)
+                .fullBufferingEnabled(true)
+                .build();
 
         // The upload should complete successfully — retry works because full buffering
         // ensures the retry buffer is populated before the HTTP layer subscribes