Fixed OOM bug (Azure#19600)

Author: gapra-msft

sdk/storage/azure-storage-blob/CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## 12.11.0-beta.2 (Unreleased)
 - Fixed a bug where downloading would throw a NPE on large downloads due to a lack of eTag.
+- Fixed a bug where more data would be buffered in buffered upload than expected due to Reactor's concatMap operator.
 
 ## 12.11.0-beta.1 (2021-02-10)
 - Added support for the 2020-06-12 service version. 

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobAsyncClient.java

@@ -530,6 +530,7 @@ private Mono<Response<BlockBlobItem>> uploadInChunks(BlockBlobAsyncClient blockB
          Write to the pool and upload the output.
          */
         return chunkedSource.concatMap(pool::write)
+            .limitRate(parallelTransferOptions.getMaxConcurrency()) // This guarantees that concatMap will only buffer maxConcurrency * chunkSize data
             .concatWith(Flux.defer(pool::flush))
             .flatMapSequential(bufferAggregator -> {
                 // Report progress as necessary.

sdk/storage/azure-storage-file-datalake/CHANGELOG.md

@@ -1,7 +1,7 @@
 # Release History
 
 ## 12.5.0-beta.2 (Unreleased)
-
+- Fixed a bug where more data would be buffered in buffered upload than expected due to Reactor's concatMap operator.
 
 ## 12.5.0-beta.1 (2021-02-10)
 - Added support for the 2020-06-12 service version. 

sdk/storage/azure-storage-file-datalake/src/main/java/com/azure/storage/file/datalake/DataLakeFileAsyncClient.java

@@ -362,6 +362,7 @@ private Mono<Response<PathInfo>> uploadInChunks(Flux<ByteBuffer> data, long file
          Write to the pool and upload the output.
          */
         return chunkedSource.concatMap(pool::write)
+            .limitRate(parallelTransferOptions.getMaxConcurrency()) // This guarantees that concatMap will only buffer maxConcurrency * chunkSize data
             .concatWith(Flux.defer(pool::flush))
             /* Map the data to a tuple 3, of buffer, buffer length, buffer offset */
             .map(bufferAggregator -> Tuples.of(bufferAggregator, bufferAggregator.length(), 0L))