Remove user configurable stream read limit

This is a fairly low level configuration that we don't foresee many customers
using; can be added back in if requested.

For now, when marking, we set the read limit to previous default of 128K.

Author: dagnir

core/auth/src/main/java/software/amazon/awssdk/auth/signer/AwsS3V4Signer.java

@@ -33,6 +33,7 @@
 import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
 import software.amazon.awssdk.http.SdkHttpFullRequest;
 import software.amazon.awssdk.utils.BinaryUtils;
+import software.amazon.awssdk.utils.IoUtils;
 
 /**
  * AWS4 signer implementation for AWS S3
@@ -242,7 +243,7 @@ private static long getContentLength(SdkHttpFullRequest.Builder requestBuilder)
         long contentLength = 0;
         byte[] tmp = new byte[4096];
         int read;
-        content.mark(getReadLimit());
+        IoUtils.markStreamWithMaxReadLimit(content);
         while ((read = content.read(tmp)) != -1) {
             contentLength += read;
         }

core/auth/src/main/java/software/amazon/awssdk/auth/signer/internal/AbstractAws4Signer.java

@@ -39,6 +39,7 @@
 import software.amazon.awssdk.core.signer.Presigner;
 import software.amazon.awssdk.http.SdkHttpFullRequest;
 import software.amazon.awssdk.utils.BinaryUtils;
+import software.amazon.awssdk.utils.IoUtils;
 import software.amazon.awssdk.utils.Logger;
 import software.amazon.awssdk.utils.http.SdkHttpUtils;
 
@@ -146,7 +147,7 @@ protected void addSessionCredentials(SdkHttpFullRequest.Builder mutableRequest,
      */
     protected String calculateContentHash(SdkHttpFullRequest.Builder mutableRequest, T signerParams) {
         InputStream payloadStream = getBinaryRequestPayloadStream(mutableRequest.content());
-        payloadStream.mark(getReadLimit());
+        IoUtils.markStreamWithMaxReadLimit(payloadStream);
         String contentSha256 = BinaryUtils.toHex(hash(payloadStream));
         try {
             payloadStream.reset();

core/auth/src/main/java/software/amazon/awssdk/auth/signer/internal/AbstractAwsSigner.java

@@ -29,12 +29,10 @@
 import java.util.TreeMap;
 import javax.crypto.Mac;
 import javax.crypto.spec.SecretKeySpec;
-import software.amazon.awssdk.annotations.ReviewBeforeRelease;
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
 import software.amazon.awssdk.auth.credentials.AwsCredentials;
 import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
-import software.amazon.awssdk.core.RequestOption;
 import software.amazon.awssdk.core.exception.SdkClientException;
 import software.amazon.awssdk.core.io.SdkDigestInputStream;
 import software.amazon.awssdk.core.signer.Signer;
@@ -244,12 +242,6 @@ protected String getCanonicalizedQueryString(Map<String, List<String>> parameter
         return SdkHttpUtils.flattenQueryParameters(sorted).orElse("");
     }
 
-    @ReviewBeforeRelease("Do we still want to make read limit user-configurable as in V1?")
-    protected static int getReadLimit() {
-        return RequestOption.DEFAULT_STREAM_BUFFER_SIZE;
-
-    }
-
     protected InputStream getBinaryRequestPayloadStream(InputStream stream) {
         try {
             if (stream == null) {

core/sdk-core/src/main/java/software/amazon/awssdk/core/RequestOption.java

@@ -22,7 +22,5 @@
  */
 @SdkProtectedApi
 public final class RequestOption {
-    public static final int DEFAULT_STREAM_BUFFER_SIZE = (1 << 17) + 1;
-
     private RequestOption() {}
 }

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

@@ -21,10 +21,7 @@
 import java.util.Optional;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
-import software.amazon.awssdk.annotations.ReviewBeforeRelease;
 import software.amazon.awssdk.annotations.SdkInternalApi;
-import software.amazon.awssdk.core.RequestOption;
-import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
 import software.amazon.awssdk.http.SdkHttpFullRequest;
 import software.amazon.awssdk.http.async.SdkHttpContentPublisher;
 import software.amazon.awssdk.utils.IoUtils;
@@ -39,10 +36,10 @@ public final class SimpleHttpContentPublisher implements SdkHttpContentPublisher
     private final byte[] content;
     private final int length;
 
-    public SimpleHttpContentPublisher(SdkHttpFullRequest request, ExecutionAttributes executionAttributes) {
+    public SimpleHttpContentPublisher(SdkHttpFullRequest request) {
         this.content = request.content().map(content -> {
             try {
-                content.mark(getReadLimit(executionAttributes));
+                IoUtils.markStreamWithMaxReadLimit(content);
                 return invokeSafely(() -> IoUtils.toByteArray(content));
             } finally {
                 invokeSafely(content::reset);
@@ -51,11 +48,6 @@ public SimpleHttpContentPublisher(SdkHttpFullRequest request, ExecutionAttribute
         this.length = content.length;
     }
 
-    @ReviewBeforeRelease("Do we still want to make read limit user-configurable as in V1?")
-    private int getReadLimit(ExecutionAttributes executionAttributes) {
-        return RequestOption.DEFAULT_STREAM_BUFFER_SIZE;
-    }
-
     @Override
     public Optional<Long> contentLength() {
         return Optional.of(Long.valueOf(length));

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStage.java

@@ -24,9 +24,7 @@
 import java.util.concurrent.TimeUnit;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import software.amazon.awssdk.annotations.ReviewBeforeRelease;
 import software.amazon.awssdk.annotations.SdkInternalApi;
-import software.amazon.awssdk.core.RequestOption;
 import software.amazon.awssdk.core.SdkStandardLogger;
 import software.amazon.awssdk.core.client.config.SdkClientOption;
 import software.amazon.awssdk.core.exception.NonRetryableException;
@@ -44,6 +42,7 @@
 import software.amazon.awssdk.core.retry.RetryUtils;
 import software.amazon.awssdk.http.SdkHttpFullRequest;
 import software.amazon.awssdk.http.SdkHttpFullResponse;
+import software.amazon.awssdk.utils.IoUtils;
 
 /**
  * Wrapper around the pipeline for a single request to provide retry functionality.
@@ -200,30 +199,12 @@ private CompletableFuture<Response<OutputT>> doExecute() throws Exception {
                 request.content().ifPresent(AsyncRetryableStage::resetRequestInputStream);
             }
 
-            request.content().ifPresent(this::markInputStream);
+            request.content().ifPresent(IoUtils::markStreamWithMaxReadLimit);
 
             SdkStandardLogger.REQUEST_LOGGER.debug(() -> (retryHandler.isRetry() ? "Retrying " : "Sending ") +
                                                          "Request: " + request);
 
             return requestPipeline.execute(retryHandler.addRetryInfoHeader(request, requestCount), context);
         }
-
-        /**
-         * Mark the input stream at the current position to allow a reset on retries.
-         */
-        private void markInputStream(InputStream originalContent) {
-            if (originalContent.markSupported()) {
-                originalContent.mark(readLimit());
-            }
-        }
-
-        /**
-         * @return Allowed read limit that we can mark request input stream. If we read past this limit we cannot reset the stream
-         * so we cannot retry the request.
-         */
-        @ReviewBeforeRelease("Do we still want to make read limit user-configurable as in V1?")
-        private int readLimit() {
-            return RequestOption.DEFAULT_STREAM_BUFFER_SIZE;
-        }
     }
 }

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/MakeAsyncHttpRequestStage.java

@@ -95,7 +95,7 @@ private CompletableFuture<Response<OutputT>> executeHttpRequest(SdkHttpFullReque
         final ResponseHandler handler = new ResponseHandler(responseHandler.prepare(), errorResponseFuture);
 
         SdkHttpContentPublisher requestProvider = context.requestProvider() == null
-                ? new SimpleHttpContentPublisher(request, context.executionAttributes())
+                ? new SimpleHttpContentPublisher(request)
                 : context.requestProvider();
         // Set content length if it hasn't been set already.
         SdkHttpFullRequest requestWithContentLength = getRequestWithContentLength(request, requestProvider);

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/RetryableStage.java

@@ -19,9 +19,8 @@
 import java.io.InputStream;
 import java.time.Duration;
 import java.util.concurrent.TimeUnit;
-import software.amazon.awssdk.annotations.ReviewBeforeRelease;
+
 import software.amazon.awssdk.annotations.SdkInternalApi;
-import software.amazon.awssdk.core.RequestOption;
 import software.amazon.awssdk.core.SdkStandardLogger;
 import software.amazon.awssdk.core.client.config.SdkClientOption;
 import software.amazon.awssdk.core.exception.ResetException;
@@ -39,6 +38,7 @@
 import software.amazon.awssdk.core.retry.RetryPolicy;
 import software.amazon.awssdk.core.retry.RetryUtils;
 import software.amazon.awssdk.http.SdkHttpFullRequest;
+import software.amazon.awssdk.utils.IoUtils;
 import software.amazon.awssdk.utils.Logger;
 
 /**
@@ -131,7 +131,7 @@ private Response<OutputT> doExecute() throws Exception {
                 doPauseBeforeRetry();
             }
 
-            request.content().ifPresent(this::markInputStream);
+            request.content().ifPresent(IoUtils::markStreamWithMaxReadLimit);
 
             SdkStandardLogger.REQUEST_LOGGER.debug(() -> (retryHandler.isRetry() ? "Retrying " : "Sending ") + "Request: " +
                                                          request);
@@ -173,24 +173,6 @@ private SdkException handleThrownException(Exception e) {
             return sdkClientException;
         }
 
-        /**
-         * Mark the input stream at the current position to allow a reset on retries.
-         */
-        private void markInputStream(InputStream originalContent) {
-            if (originalContent.markSupported()) {
-                originalContent.mark(readLimit());
-            }
-        }
-
-        /**
-         * @return Allowed read limit that we can mark request input stream. If we read past this limit we cannot reset the stream
-         * so we cannot retry the request.
-         */
-        @ReviewBeforeRelease("Do we still want to make read limit user-configurable as in V1?")
-        private int readLimit() {
-            return RequestOption.DEFAULT_STREAM_BUFFER_SIZE;
-        }
-
         /**
          * Sleep for a period of time on failed request to avoid flooding a service with retries.
          */

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

@@ -20,7 +20,7 @@ public SimpleRequestProviderTckTest() {
 
     @Override
     public Publisher<ByteBuffer> createPublisher(long l) {
-        return new SimpleHttpContentPublisher(makeFullRequest(), new ExecutionAttributes());
+        return new SimpleHttpContentPublisher(makeFullRequest());
     }
 
     @Override

utils/src/main/java/software/amazon/awssdk/utils/IoUtils.java

@@ -124,4 +124,16 @@ public static void drainInputStream(InputStream in) {
             // Stream may be self closed by HTTP client so we ignore any failures.
         }
     }
+
+    /**
+     * If the stream supports marking, marks the stream at the current position with a {@code readLimit} value of
+     * 128 KiB.
+     *
+     * @param s The stream.
+     */
+    public static void markStreamWithMaxReadLimit(InputStream s) {
+        if (s.markSupported()) {
+            s.mark(1 << 17);
+        }
+    }
 }