Prevent retries on non-repeatable request bodies

.changes/next-release/bugfix-AWSSDKforJavav2-1d9fdb6.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "Prevent subsequent retries for request bodies that are not\nrepeatable (i.e. cannot be reset to the initial position), since they will\nalways fail."
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/sync/SingleShotContentStreamProvider.java

@@ -0,0 +1,42 @@
+/*
+ * 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.core.internal.sync;
+
+import java.io.InputStream;
+import software.amazon.awssdk.core.exception.NonRetryableException;
+import software.amazon.awssdk.http.ContentStreamProvider;
+
+/**
+ * A {@code ContentStreamProvider} that can only provide the stream once. It throws a {@link NonRetryableException} for all
+ * subsequent calls to {@link #newStream()}.
+ */
+public final class SingleShotContentStreamProvider implements ContentStreamProvider {
+    private final InputStream is;
+    private boolean available = true;
+
+    public SingleShotContentStreamProvider(InputStream is) {
+        this.is = is;
+    }
+
+    @Override
+    public InputStream newStream() {
+        if (available) {
+            available = false;
+            return is;
+        }
+        throw NonRetryableException.create("Stream cannot be reset to the initial position.");
+    }
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/sync/RequestBody.java

@@ -31,7 +31,9 @@
 import java.util.Arrays;
 import java.util.Optional;
 import software.amazon.awssdk.annotations.SdkPublicApi;
+import software.amazon.awssdk.core.exception.NonRetryableException;
 import software.amazon.awssdk.core.internal.sync.FileContentStreamProvider;
+import software.amazon.awssdk.core.internal.sync.SingleShotContentStreamProvider;
 import software.amazon.awssdk.core.internal.util.Mimetype;
 import software.amazon.awssdk.core.io.ReleasableInputStream;
 import software.amazon.awssdk.http.ContentStreamProvider;
@@ -130,12 +132,17 @@ public static RequestBody fromFile(File file) {
     public static RequestBody fromInputStream(InputStream inputStream, long contentLength) {
         IoUtils.markStreamWithMaxReadLimit(inputStream);
         InputStream nonCloseable = nonCloseableInputStream(inputStream);
-        return fromContentProvider(() -> {
-            if (nonCloseable.markSupported()) {
+
+        ContentStreamProvider provider;
+        if (nonCloseable.markSupported()) {
+            provider = () -> {
                 invokeSafely(nonCloseable::reset);
-            }
-            return nonCloseable;
-        }, contentLength, Mimetype.MIMETYPE_OCTET_STREAM);
+                return nonCloseable;
+            };
+        } else {
+            provider = new SingleShotContentStreamProvider(nonCloseable);
+        }
+        return fromContentProvider(provider, contentLength, Mimetype.MIMETYPE_OCTET_STREAM);
     }
 
     /**

core/sdk-core/src/test/java/software/amazon/awssdk/core/sync/RequestBodyTest.java

@@ -16,7 +16,10 @@
 package software.amazon.awssdk.core.sync;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
 
 import com.google.common.jimfs.Configuration;
 import com.google.common.jimfs.Jimfs;
@@ -33,6 +36,7 @@
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
+import software.amazon.awssdk.core.exception.NonRetryableException;
 import software.amazon.awssdk.core.internal.util.Mimetype;
 import software.amazon.awssdk.utils.IoUtils;
 import software.amazon.awssdk.utils.StringInputStream;
@@ -140,4 +144,19 @@ public void remainingByteBufferConstructorOnlyRemainingBytesCopied() throws IOEx
         byte[] requestBodyBytes = IoUtils.toByteArray(requestBody.contentStreamProvider().newStream());
         assertThat(ByteBuffer.wrap(requestBodyBytes)).isEqualTo(bb);
     }
+
+    @Test
+    public void fromInputStream_markNotSupported_throwsNonRetryableExceptionOnSuccessiveNewStreams() {
+        byte[] content = "hello".getBytes(StandardCharsets.UTF_8);
+        InputStream is = spy(new ByteArrayInputStream(content));
+        when(is.markSupported()).thenReturn(false);
+
+        RequestBody requestBody = RequestBody.fromInputStream(is, content.length);
+
+        assertThatNoException().isThrownBy(requestBody.contentStreamProvider()::newStream);
+
+        assertThatThrownBy(requestBody.contentStreamProvider()::newStream)
+            .isInstanceOf(NonRetryableException.class)
+            .hasMessageContaining("cannot be reset");
+    }
 }