Preserve content-length when calling mark in SdkLengthAwareInputStream (#6408)

* Preserve content-length when calling mark in SdkLengthAwareInputStream

* Add test case

* Ensure that markedRemaining is initialized

Author: alextwoods

.changes/next-release/bugfix-AWSSDKforJavav2-08c895c.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "Preserve the initial content-length when calling mark in SdkLengthAwareInputStream."
+}

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

@@ -32,13 +32,15 @@
 @SdkInternalApi
 public class SdkLengthAwareInputStream extends FilterInputStream {
     private static final Logger LOG = Logger.loggerFor(SdkLengthAwareInputStream.class);
-    private long length;
+    private final long length;
     private long remaining;
+    private long markedRemaining;
 
     public SdkLengthAwareInputStream(InputStream in, long length) {
         super(in);
         this.length = Validate.isNotNegative(length, "length");
         this.remaining = this.length;
+        this.markedRemaining = this.remaining;
     }
 
     @Override
@@ -104,15 +106,14 @@ public int available() throws IOException {
     @Override
     public void mark(int readlimit) {
         super.mark(readlimit);
-        // mark() causes reset() to change the stream's position back to the current position. Therefore, when reset() is called,
-        // the new length of the stream will be equal to the current value of 'remaining'.
-        length = remaining;
+        // Store the current remaining bytes to restore on reset()
+        markedRemaining = remaining;
     }
 
     @Override
     public void reset() throws IOException {
         super.reset();
-        remaining = length;
+        remaining = markedRemaining;
     }
 
     public long remaining() {

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

@@ -249,6 +249,27 @@ void readByte_readExactLength_doesNotThrow() throws IOException {
         assertThat(total).isEqualTo(delegateLength);
     }
 
+    @Test
+    void readBytePartialThenMark_doesNotResetContentLength() throws IOException {
+        int delegateLength = 16;
+        int expectedContentLength = delegateLength + 1;
+        ByteArrayInputStream delegate = new ByteArrayInputStream(new byte[delegateLength]);
+
+        SdkLengthAwareInputStream is = new SdkLengthAwareInputStream(delegate, expectedContentLength);
+        is.read(); // read one byte
+        is.mark(1024);
+        // read another byte and reset, the length should not be reset based on the byte that was already read
+        is.read();
+        is.reset();
+        assertThatThrownBy(() -> {
+            int read;
+            while ((read = is.read()) != -1) {
+            }
+        })
+            .isInstanceOf(IllegalStateException.class)
+            .hasMessageContaining("The request content has fewer bytes than the specified content-length: " + expectedContentLength);
+    }
+
     @Test
     void readByte_delegateLongerThanRequired_truncated() throws IOException {
         int delegateLength = 32;