fix: InputStreamContent closes caller InputStream between retry attempts

sdk/core/azure-core/CHANGELOG.md

@@ -8,6 +8,8 @@
 
 ### Bugs Fixed
 
+- Fixed a bug where retrying requests with bodies created from a mark/reset-capable `InputStream` could fail because the stream was closed between retry attempts. ([#49650](https://github.com/Azure/azure-sdk-for-java/pull/49650))
+
 ### Other Changes
 
 ## 1.58.1 (2026-06-08)

sdk/core/azure-core/src/main/java/com/azure/core/implementation/http/rest/LengthValidatingInputStream.java

@@ -81,7 +81,8 @@ public int available() throws IOException {
 
     @Override
     public void close() throws IOException {
-        inner.close();
+        // No-op. RestProxy length validation can wrap caller-owned InputStreamContent. Closing this wrapper must not
+        // close the underlying stream, including between retry attempts where the caller-owned stream needs to be reset.
     }
 
     @Override

sdk/core/azure-core/src/test/java/com/azure/core/implementation/http/rest/RestProxyUtilsTests.java

@@ -29,6 +29,7 @@
 
 import static com.azure.core.CoreTestUtils.assertArraysEqual;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -210,6 +211,67 @@ public void expectedBodyLengthSync() throws IOException {
         }
     }
 
+    @Test
+    public void lengthValidatingInputStreamCloseDoesNotCloseUnderlying() throws IOException {
+        final boolean[] closed = new boolean[1];
+
+        try (InputStream innerStream = new InputStream() {
+            private final ByteArrayInputStream delegate = new ByteArrayInputStream(EXPECTED);
+
+            private void ensureOpen() throws IOException {
+                if (closed[0]) {
+                    throw new IOException("Stream closed");
+                }
+            }
+
+            @Override
+            public void close() throws IOException {
+                closed[0] = true;
+                delegate.close();
+            }
+
+            @Override
+            public synchronized int read() throws IOException {
+                ensureOpen();
+                return delegate.read();
+            }
+
+            @Override
+            public synchronized void mark(int readlimit) {
+                delegate.mark(readlimit);
+            }
+
+            @Override
+            public synchronized void reset() throws IOException {
+                ensureOpen();
+                delegate.reset();
+            }
+
+            @Override
+            public boolean markSupported() {
+                return delegate.markSupported();
+            }
+        }) {
+            LengthValidatingInputStream validatingStream
+                = new LengthValidatingInputStream(innerStream, EXPECTED.length);
+            // Simulate the retry scenario: the wrapper is closed after an attempt, then the pipeline tries to reset.
+            validatingStream.mark(EXPECTED.length);
+            assertTrue(validatingStream.read() != -1);
+
+            validatingStream.close();
+
+            try {
+                validatingStream.reset();
+            } catch (IOException e) {
+                fail("Expected LengthValidatingInputStream.reset() to succeed after close(), but it failed.", e);
+            }
+            byte[] actual = new byte[EXPECTED.length];
+            assertEquals(EXPECTED.length, validatingStream.read(actual));
+            assertArraysEqual(EXPECTED, actual);
+            assertFalse(closed[0], "LengthValidatingInputStream.close() must not close the underlying stream.");
+        }
+    }
+
     private static Mono<byte[]> validateAndCollectRequestAsync(HttpRequest request) {
         return RestProxyUtils.validateLengthAsync(request)
             .flatMap(r -> FluxUtil.collectBytesInByteBufferStream(r.getBody()));