Improve error message for the error case where a request using RequestBody#fromInputStream failed to retry due to lack of mark and reset support (#6254)

Author: zoewangg

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

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "Improve error message for the error case where a request using RequestBody#fromInputStream failed to retry due to lack of mark and reset support. See [#6174](https://github.com/aws/aws-sdk-java-v2/issues/6174)"
+}

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

@@ -38,7 +38,6 @@
 import software.amazon.awssdk.http.ContentStreamProvider;
 import software.amazon.awssdk.http.Header;
 import software.amazon.awssdk.utils.BinaryUtils;
-import software.amazon.awssdk.utils.IoUtils;
 
 /**
  * Represents the body of an HTTP request. Must be provided for operations that have a streaming input.
@@ -120,38 +119,29 @@ public static RequestBody fromFile(File file) {
      * Creates a {@link RequestBody} from an input stream. {@value Header#CONTENT_LENGTH} must
      * be provided so that the SDK does not have to make two passes of the data.
      * <p>
-     * The stream will not be closed by the SDK. It is up to to caller of this method to close the stream. The stream
-     * should not be read outside of the SDK (by another thread) as it will change the state of the {@link InputStream} and
+     * The stream will not be closed by the SDK. It is up to caller of this method to close the stream. The stream
+     * should not be read outside the SDK (by another thread) as it will change the state of the {@link InputStream} and
      * could tamper with the sending of the request.
      * <p>
      * To support resetting via {@link ContentStreamProvider}, this uses {@link InputStream#reset()} and uses a read limit of
      * 128 KiB. If you need more control, use {@link #fromContentProvider(ContentStreamProvider, long, String)} or
      * {@link #fromContentProvider(ContentStreamProvider, String)}.
      *
+     * <p>
+     * It is recommended to provide a stream that supports mark and reset for retry. If the stream does not support mark and
+     * reset, an {@link IllegalStateException} will be thrown during retry.
+     *
      * @param inputStream   Input stream to send to the service. The stream will not be closed by the SDK.
      * @param contentLength Content length of data in input stream. If a content length smaller than the actual size of the
      *                      object is set, the client will truncate the stream to the specified content length and only send
      *                      exactly the number of bytes equal to the content length.
      * @return RequestBody instance.
      */
     public static RequestBody fromInputStream(InputStream inputStream, long contentLength) {
-        IoUtils.markStreamWithMaxReadLimit(inputStream);
-        InputStream nonCloseable = nonCloseableInputStream(inputStream);
-        ContentStreamProvider provider = new ContentStreamProvider() {
-            @Override
-            public InputStream newStream() {
-                if (nonCloseable.markSupported()) {
-                    invokeSafely(nonCloseable::reset);
-                }
-                return nonCloseable;
-            }
-
-            @Override
-            public String name() {
-                return ProviderType.STREAM.getName();
-            }
-        };
-        return fromContentProvider(provider, contentLength, Mimetype.MIMETYPE_OCTET_STREAM);
+        ContentStreamProvider contentStreamProvider = ContentStreamProvider.fromInputStream(
+            nonCloseableInputStream(inputStream));
+        return fromContentProvider(contentStreamProvider,
+                                   contentLength, Mimetype.MIMETYPE_OCTET_STREAM);
     }
 
     /**

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

@@ -39,6 +39,7 @@
 import software.amazon.awssdk.checksums.SdkChecksum;
 import software.amazon.awssdk.core.internal.sync.BufferingContentStreamProvider;
 import software.amazon.awssdk.core.internal.util.Mimetype;
+import software.amazon.awssdk.testutils.RandomInputStream;
 import software.amazon.awssdk.utils.BinaryUtils;
 import software.amazon.awssdk.utils.IoUtils;
 import software.amazon.awssdk.utils.StringInputStream;
@@ -163,6 +164,17 @@ public void fromInputStream_streamSupportsReset_resetsTheStream() {
         assertThat(getCrc32(requestBody.contentStreamProvider().newStream())).isEqualTo(streamCrc32);
     }
 
+    @Test
+    public void fromInputStream_streamNotSupportReset_shouldThrowException() {
+        RandomInputStream stream = new RandomInputStream(100);
+        assertThat(stream.markSupported()).isFalse();
+        RequestBody requestBody = RequestBody.fromInputStream(stream, 100);
+        IoUtils.drainInputStream(requestBody.contentStreamProvider().newStream());
+        assertThatThrownBy(() -> requestBody.contentStreamProvider().newStream())
+            .isInstanceOf(IllegalStateException.class)
+            .hasMessageContaining("Content input stream does not support mark/reset");
+    }
+
     private static String getCrc32(InputStream inputStream) {
         byte[] buff = new byte[1024];
         int read;