Storage - [Content Validation] Make synchronous stageBlock request body replayable on retry (#49612)

* Make synchronous stageBlock request body replayable on retry

* updating assets (unrelated to the changes in this PR)

* updating recordings

Author: ibrandes

sdk/storage/azure-storage-blob/assets.json

@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "java",
   "TagPrefix": "java/storage/azure-storage-blob",
-  "Tag": "java/storage/azure-storage-blob_afc0d343f3"
+  "Tag": "java/storage/azure-storage-blob_494b72fca7"
 }

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlockBlobClient.java

@@ -38,14 +38,17 @@
 import com.azure.storage.blob.options.BlockBlobSimpleUploadOptions;
 import com.azure.storage.blob.options.BlockBlobStageBlockFromUrlOptions;
 import com.azure.storage.blob.options.BlockBlobStageBlockOptions;
+import com.azure.storage.common.Utility;
 import com.azure.storage.common.implementation.Constants;
 import com.azure.storage.common.implementation.StorageImplUtils;
 import com.azure.storage.common.implementation.StorageSeekableByteChannel;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.net.URL;
+import java.nio.ByteBuffer;
 import java.nio.channels.SeekableByteChannel;
 import java.time.Duration;
 import java.util.List;
@@ -783,10 +786,13 @@ public Response<Void> stageBlockWithResponse(String base64BlockId, InputStream d
         String leaseId, Duration timeout, Context context) {
         StorageImplUtils.assertNotNull("data", data);
 
-        Mono<Response<Void>> response = client.stageBlockWithResponseInternal(
-            new BlockBlobStageBlockOptions(base64BlockId, BinaryData.fromStream(data, length)).setContentMd5(contentMd5)
-                .setLeaseId(leaseId),
-            context);
+        Flux<ByteBuffer> fbb
+            = Utility.convertStreamToByteBuffer(data, length, BlobAsyncClient.BLOB_DEFAULT_UPLOAD_BLOCK_SIZE, true);
+
+        Mono<Response<Void>> response = BinaryData.fromFlux(fbb, length, false)
+            .flatMap(binaryData -> client.stageBlockWithResponseInternal(
+                new BlockBlobStageBlockOptions(base64BlockId, binaryData).setContentMd5(contentMd5).setLeaseId(leaseId),
+                context));
         return blockWithOptionalTimeout(response, timeout);
     }