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. The user/caller is responsible for closing the underlying stream's lifecycle,
+        // in line with the BinaryData.fromStream contract. Closing it here breaks stream replayability on retries.
     }
 
     @Override

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

@@ -210,6 +210,28 @@ public void expectedBodyLengthSync() throws IOException {
         }
     }
 
+    @Test
+    public void lengthValidatingInputStreamCloseDoesNotCloseUnderlying() throws IOException {
+        final boolean[] closed = new boolean[1];
+
+        try (InputStream innerStream = new ByteArrayInputStream(EXPECTED) {
+            @Override
+            public void close() throws IOException {
+                closed[0] = true;
+                super.close();
+            }
+        }) {
+            LengthValidatingInputStream validatingStream
+                = new LengthValidatingInputStream(innerStream, EXPECTED.length);
+
+            validatingStream.close();
+
+            // Validate that closing the wrapper does not close the underlying stream.
+            org.junit.jupiter.api.Assertions.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()));

sdk/core/azure-core/src/test/java/com/azure/core/util/UrlBuilderTests.java

@@ -707,7 +707,7 @@ public void parallelParsing() throws InterruptedException {
             return UrlBuilder.parse("https://example" + i + ".com");
         }).collect(Collectors.toCollection(() -> new ArrayList<>(20000)));
 
-        List<Future<UrlBuilder>> futures = SharedExecutorService.getInstance().invokeAll(tasks, 10, TimeUnit.SECONDS);
+        List<Future<UrlBuilder>> futures = SharedExecutorService.getInstance().invokeAll(tasks, 60, TimeUnit.SECONDS);
         for (Future<UrlBuilder> future : futures) {
             assertTrue(future.isDone());
             assertDoesNotThrow(() -> future.get());
@@ -726,7 +726,7 @@ public void fluxParallelParsing() {
                 return UrlBuilder.parse("https://example" + i + ".com");
             })
             .then()
-            .timeout(Duration.ofSeconds(10));
+            .timeout(Duration.ofSeconds(60));
 
         StepVerifier.create(mono).verifyComplete();
         assertEquals(20000, callCount.get());