Add changelog files, fix tests, update to throw IllegalStateException and add more tests

Author: zoewangg

.changes/next-release/feature-AWSSDKforJavav2-02883e2.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "The SDK now throws exception for input streaming operation if the stream has fewer bytes (i.e. reaches EOF) before the expected length is reached."
+}

.changes/next-release/feature-AWSSDKforJavav2-9dffbaf.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "The SDK now does not buffer input data from `RequestBody#fromInputStream` in cases where the InputStream does not support mark and reset."
+}

.changes/next-release/feature-AWSSDKforJavav2-ee5927f.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "The SDK now does not buffer input data from ContentStreamProvider in cases where content length is known."
+}

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

@@ -15,6 +15,8 @@
 
 package software.amazon.awssdk.core.internal.http.pipeline.stages;
 
+import static software.amazon.awssdk.http.Header.CONTENT_LENGTH;
+
 import java.time.Duration;
 import java.util.Optional;
 import software.amazon.awssdk.annotations.SdkInternalApi;
@@ -111,7 +113,7 @@ private static SdkHttpFullRequest enforceContentLengthIfPresent(SdkHttpFullReque
 
         Optional<Long> contentLength = contentLength(request);
         if (!contentLength.isPresent()) {
-            LOG.warn(() -> String.format("Request contains a body but does not have a Content-Length header. Not validating "
+            LOG.debug(() -> String.format("Request contains a body but does not have a Content-Length header. Not validating "
                                          + "the amount of data sent to the service: %s", request));
             return request;
         }
@@ -125,7 +127,7 @@ private static SdkHttpFullRequest enforceContentLengthIfPresent(SdkHttpFullReque
     }
 
     private static Optional<Long> contentLength(SdkHttpFullRequest request) {
-        Optional<String> contentLengthHeader = request.firstMatchingHeader("Content-Length");
+        Optional<String> contentLengthHeader = request.firstMatchingHeader(CONTENT_LENGTH);
 
         if (contentLengthHeader.isPresent()) {
             try {

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/io/SdkLengthAwareInputStream.java

@@ -53,7 +53,9 @@ public int read() throws IOException {
         if (read != -1) {
             remaining--;
         } else if (remaining != 0) { // EOF, ensure we've read the number of expected bytes
-            throw new IOException("Reached EOF before reading entire expected content");
+            throw new IllegalStateException("The request content has fewer bytes than the "
+                                            + "specified "
+                                            + "content-length: " + length + " bytes.");
         }
         return read;
     }
@@ -74,7 +76,9 @@ public int read(byte[] b, int off, int len) throws IOException {
 
         // EOF, ensure we've read the number of expected bytes
         if (read == -1 && remaining != 0) {
-            throw new IOException("Reached EOF before reading entire expected content");
+            throw new IllegalStateException("The request content has fewer bytes than the "
+                                            + "specified "
+                                            + "content-length: " + length + " bytes.");
         }
 
         return read;

core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/io/SdkLengthAwareInputStreamTest.java

@@ -179,8 +179,8 @@ void readArray_delegateShorterThanExpected_throws() {
             while ((read = is.read(buff, 0, buff.length)) != -1) {
             }
         })
-            .isInstanceOf(IOException.class)
-            .hasMessage("Reached EOF before reading entire expected content");
+            .isInstanceOf(IllegalStateException.class)
+            .hasMessageContaining("The request content has fewer bytes than the specified content-length");
     }
 
     @Test
@@ -230,8 +230,8 @@ void readByte_delegateShorterThanExpected_throws() {
             while ((read = is.read()) != -1) {
             }
         })
-            .isInstanceOf(IOException.class)
-            .hasMessage("Reached EOF before reading entire expected content");
+            .isInstanceOf(IllegalStateException.class)
+            .hasMessageContaining("The request content has fewer bytes than the specified content-length");
     }
 
     @Test

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

@@ -122,49 +122,11 @@ public void newStream_delegateStreamClosedOnBufferingStreamClose() throws IOExce
     }
 
     @Test
-    public void newStream_lengthKnown_readUpToLengthThenClosed_newStreamUsesBufferedData() throws IOException {
+    public void newStream_lengthKnown_doesNotBuffer() throws IOException {
         ByteArrayInputStream stream = new ByteArrayInputStream(TEST_DATA);
         requestBody = RequestBody.fromContentProvider(() -> stream, TEST_DATA.length, "text/plain");
-
-        int totalRead = 0;
-        int read;
-
-        InputStream stream1 = requestBody.contentStreamProvider().newStream();
-        do {
-            read = stream1.read();
-            if (read != -1) {
-                ++totalRead;
-            }
-        } while (read != -1);
-
-        assertThat(totalRead).isEqualTo(TEST_DATA.length);
-
-        stream1.close();
-
         assertThat(requestBody.contentStreamProvider().newStream())
-            .isInstanceOf(BufferingContentStreamProvider.ByteArrayStream.class);
-    }
-
-    @Test
-    public void newStream_lengthKnown_partialRead_close_doesNotBufferData() throws IOException {
-        // We need a large buffer because BufferedInputStream buffers data in chunks. If the buffer is small enough, a single
-        // read() on the BufferedInputStream might actually buffer all the delegate's data.
-
-        byte[] newData = new byte[16536];
-        new Random().nextBytes(newData);
-        ByteArrayInputStream stream = new ByteArrayInputStream(newData);
-        requestBody = RequestBody.fromContentProvider(() -> stream, newData.length, "text/plain");
-
-        InputStream stream1 = requestBody.contentStreamProvider().newStream();
-        int read = stream1.read();
-        assertThat(read).isNotEqualTo(-1);
-
-        stream1.close();
-
-        InputStream stream2 = requestBody.contentStreamProvider().newStream();
-        assertThat(stream2).isInstanceOf(BufferingContentStreamProvider.BufferStream.class);
-
-        assertThat(getCrc32(stream2)).isEqualTo(getCrc32(new ByteArrayInputStream(newData)));
+            .isInstanceOf(ByteArrayInputStream.class);
     }
 
     @Test

test/architecture-tests/src/test/java/software/amazon/awssdk/archtests/CodingConventionWithSuppressionTest.java

@@ -32,6 +32,7 @@
 import java.util.Set;
 import java.util.regex.Pattern;
 import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.core.internal.http.pipeline.stages.MakeHttpRequestStage;
 import software.amazon.awssdk.metrics.publishers.emf.EmfMetricLoggingPublisher;
 import software.amazon.awssdk.metrics.publishers.emf.internal.MetricEmfConverter;
 import software.amazon.awssdk.utils.Logger;
@@ -44,25 +45,13 @@
  */
 public class CodingConventionWithSuppressionTest {
 
-    /**
-     * Suppressions for APIs used in generated code to avoid having to update archunit_store for new services. Unfortunately, we
-     * can't change the following because it may break people :(
-     * <p>
-     * DO NOT ADD NEW EXCEPTIONS
-     */
     private static final Set<Pattern> ALLOWED_WARN_LOG_SUPPRESSION = new HashSet<>(
-        Arrays.asList(ArchUtils.classNameToPattern(EmfMetricLoggingPublisher.class), ArchUtils.classNameToPattern(MetricEmfConverter.class))
-    );
+        Arrays.asList(ArchUtils.classNameToPattern(EmfMetricLoggingPublisher.class),
+                      ArchUtils.classNameToPattern(MetricEmfConverter.class),
+                      ArchUtils.classNameToPattern(MakeHttpRequestStage.class)));
 
-    /**
-     * Suppressions for APIs used in generated code to avoid having to update archunit_store for new services. Unfortunately, we
-     * can't change the following because it may break people :(
-     * <p>
-     * DO NOT ADD NEW EXCEPTIONS
-     */
     private static final Set<Pattern> ALLOWED_ERROR_LOG_SUPPRESSION = new HashSet<>(
-        Arrays.asList(ArchUtils.classNameToPattern(EmfMetricLoggingPublisher.class))
-    );
+        Arrays.asList(ArchUtils.classNameToPattern(EmfMetricLoggingPublisher.class)));
 
     @Test
     void shouldNotAbuseWarnLog() {

test/protocol-tests/src/test/java/software/amazon/awssdk/protocol/tests/TruncatedRequestStreamTest.java renamed to test/protocol-tests/src/test/java/software/amazon/awssdk/protocol/tests/StreamingRequestInputStreamLengthAwareTest.java

@@ -16,6 +16,7 @@
 package software.amazon.awssdk.protocol.tests;
 
 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.Mockito.mock;
 import static org.mockito.Mockito.verify;
@@ -25,12 +26,17 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
+import java.util.stream.Stream;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.mockito.ArgumentCaptor;
 import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.internal.util.Mimetype;
 import software.amazon.awssdk.core.sync.RequestBody;
 import software.amazon.awssdk.http.AbortableInputStream;
 import software.amazon.awssdk.http.ContentStreamProvider;
@@ -43,10 +49,11 @@
 import software.amazon.awssdk.services.protocolrestjson.ProtocolRestJsonClient;
 import software.amazon.awssdk.services.protocolrestjson.model.StreamingInputOperationRequest;
 
-public class TruncatedRequestStreamTest {
+public class StreamingRequestInputStreamLengthAwareTest {
     private ProtocolRestJsonClient client;
     private SdkHttpClient mockHttpClient;
 
+
     @BeforeEach
     public void setup() throws IOException {
         mockHttpClient = mock(SdkHttpClient.class);
@@ -82,8 +89,7 @@ public void streamingRequest_specifiedLengthTruncatesStream_lengthHonored() thro
         int bodyLength = 16;
 
         client.streamingInputOperation(StreamingInputOperationRequest.builder().build(),
-                                       bodyFromStream(32, bodyLength));
-
+                                       bodyFromInputStream(32, bodyLength));
         ArgumentCaptor<HttpExecuteRequest> requestCaptor = ArgumentCaptor.forClass(HttpExecuteRequest.class);
 
         verify(mockHttpClient).prepareRequest(requestCaptor.capture());
@@ -96,9 +102,30 @@ public void streamingRequest_specifiedLengthTruncatesStream_lengthHonored() thro
         assertThat(drainStream(streamProvider.newStream())).isEqualTo(bodyLength);
     }
 
-    private static RequestBody bodyFromStream(int streamLength, int bodyLength) {
+
+    public static Stream<Arguments> requestBodies() {
+        return Stream.of(Arguments.of(bodyFromInputStream(32, 64), 64),
+                         Arguments.of(bodyFromContentProvider(32, 50), 50));
+    }
+
+    @ParameterizedTest
+    @MethodSource("requestBodies")
+    public void streamingRequest_streamLessThanContentLength_shouldThrowException(RequestBody body, long contentLength) throws IOException {
+        assertThatThrownBy(() -> client.streamingInputOperation(StreamingInputOperationRequest.builder().build(),
+                                                                body)).isInstanceOf(IllegalStateException.class)
+                                                                      .hasMessageContaining("The request content has fewer "
+                                                                                            + "bytes than the specified "
+                                                                                            + "content-length");
+    }
+
+    private static RequestBody bodyFromInputStream(int streamLength, int contentLength) {
+        InputStream is = new ByteArrayInputStream(new byte[streamLength]);
+        return RequestBody.fromInputStream(is, contentLength);
+    }
+
+    private static RequestBody bodyFromContentProvider(int streamLength, int contentLength) {
         InputStream is = new ByteArrayInputStream(new byte[streamLength]);
-        return RequestBody.fromInputStream(is, bodyLength);
+        return RequestBody.fromContentProvider(() -> is, contentLength, Mimetype.MIMETYPE_OCTET_STREAM);
     }
 
     private static int drainStream(InputStream is) throws IOException {