DataLakeServiceAsyncClient chunked Flux<ByteBuffer> read()

[o-shevchenko]

Query/Question
Is there a way to configure chunk size for data stream?
As far as I see it returns 8kb chunks and it can't be chaged.
We would like to set chunk size:

val chunkSize = DataSize.ofMegabytes(1).toBytes()
val data: Flux<ByteBuffer> = client
            .getFileSystemAsyncClient(fileSystemName)
            .getFileAsyncClient(path)
            .read(chunkSize) // <--- Set chunk size

Why is this not a Bug or a feature Request?
I'm not sure if such a common functionality already exists. If not, will be great to move it to a Feature Request.

Setup (please complete the following information if applicable):

• OS: MacOS
• IDE: IntelliJ
• Library/Libraries: com.azure:azure-storage-file-datalake:12.23.0

Information Checklist
Kindly make sure that you have added all the following information above and checkoff the required fields otherwise we will treat the issuer as an incomplete report

• Query Added
• Setup information Added

[github-actions]

@gunjansingh-msft @ibrandes @kyleknap @seanmcc-msft

[github-actions]

Thank you for your feedback. Tagging and routing to the team member best able to assist.

[alzimmermsft]

Thanks for asking this question @o-shevchenko

Adding HttpClient tag and myself as this is more of an azure-core-http-netty thing and DataLake happens to be the SDK used.

In our azure-core-http-netty HttpClient we just flow back the chunks of data that Reactor Netty is returning from Netty. Per Netty's design the default max chunk size is 8 KB (https://netty.io/4.1/api/io/netty/handler/codec/http/HttpObjectDecoder.html), and if you were to inspect a lot of chunks you'd also see on occasion that vary to something smaller than 8 KB.

We do have a sample on how to customize the HttpClient created by azure-core-http-netty to return a chunk size larger than 8 KB:

https://github.com/Azure/azure-sdk-for-java/tree/main/sdk/core/azure-core-http-netty#create-a-client-with-custom-max-chunk-size

The API is deprecated in later versions of Reactor Netty in preparation for Netty 5. But that shouldn't be a concern as we'd likely create a new package for a Netty 5-based azure-sdk-http-* implementation.

Unfortunately, after doing this review it appears our azure-sdk-http-okhttp hardcodes to 8 KB, it's unclear on what azure-sdk-http-jdk-httpclient emits as there isn't a clear configuration option, and azure-sdk-http-vertx should be the same as Netty as Vert.x appears to wrap Netty (only difference would be passing a custom Vert.x HttpClientOptions with setMaxChunkSize set, but setting HttpClientOptions in azure-core-http-vertx ignores all other VertxHttpClientBuilder configurations).

[o-shevchenko]

Thanks @alzimmermsft for the quick response.
I tried to set chunk size in https://github.com/Azure/azure-sdk-for-java/tree/main/sdk/core/azure-core-http-netty#create-a-client-with-custom-max-chunk-size . But client still returns small chunks (e.g. I get now 15637 instead of 8192 even if I set 5mb chunkSize)
That's not really what we are looking for. It tells the HTTP‐decoder “don’t hand me buffers larger than X” – but we will still get buffers of any size up to that maximum. That's just a random size for such a buffer, and not really chunking. So, we need to manually rechunk it somehow and it still doesn't fix a bottleneck from the source.
For S3 client we can set chunk size and pushdown reading with chunks to S3 directly to ensure max throughput.

val builder = S3AsyncClient.crtBuilder()
            .minimumPartSizeInBytes(chunkSize.toLong())
            .initialReadBufferSizeInBytes(chunkSize.toLong())

We need to read hundreds of GBs using such a streaming client and that will take ages if we stream it in x-kb size without ability of setting the chunkSize for Azure client.

[alzimmermsft]

Thanks for the additional information @o-shevchenko, want to clarify a few things.

From reading the S3AsyncClient.crtBuilder() documentation it appears what I shared is different.

minimumPartSizeInBytes(Long) appears to be the total network operation size. As it then links out to this https://docs.aws.amazon.com/AmazonS3/latest/userguide/qfacts.html. This is akin to calling DataLakeFileAsyncClient.readToFile(ReadToFileOptions) and configuring the ParallelTransferOptions used as that can mutate how large each read request to DataLake is. @kyleknap this appears to be a gap between Blobs and DataLake where DataLake doesn't have a Flux<ByteBuffer> read API that allows changing how big each network call is.

initialReadBufferSizeInBytes(Long) is a functionality that doesn't exist in either Blobs or DataLake. Those SDKs have left it to the user to manage aggregating buffers emitted from the request. I will call out though, if your read request is moving data between different Azure Storage locations the upload APIs that take ParallelTransferOptions will handle aggregating content. For example, if you passed the read Flux<ByteBuffer> from your sample code to DataLakeFileAsyncClient.upload(data, parallelTransferOptions, true) (https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/storage/azure-storage-file-datalake/src/main/java/com/azure/storage/file/datalake/DataLakeFileAsyncClient.java#L392) it would aggregate using the configured ParallelTransferOptions.getBlockSizeLong() (which defaults to 4MB or 8MB).

[o-shevchenko]

Could you suggest how we can pushdown chunking to ADLS to read files from Azure and receive chunked data with a fixed size?
e.g. Flux with each ByteBuffer size 1MB.

[o-shevchenko]

I also tried to use readWithResponse since it allows you to set FileRange. And it works fine for chunking, but that creates a lot of HTTP requests instead of one when we use just read(), that might impact performance or we might face problems with authentication (cred expired) in the middle of reading a big file and we still have a bottleneck since HTTP decoder returns smaller chunks anyway that we should merge.

val fileClient = client // <-- Client with HttpClient.create().httpResponseDecoder { spec -> spec.maxChunkSize(chunkSize) }
            .getFileSystemAsyncClient(fileSystemName)
            .getFileAsyncClient(path)

        val dataBuffers: Flux<DataBuffer> = fileClient.properties
            .flatMapMany { props ->
                val numChunks = ((props.fileSize + chunkSize - 1) / chunkSize).toInt() // chunkSize=1MB

                Flux.range(0, numChunks).concatMap { idx ->
                    val start = idx.toLong() * chunkSize
                    fileClient.readWithResponse(
                        FileRange(start, chunkSize.toLong()),
                        DownloadRetryOptions(),
                        DataLakeRequestConditions(),
                        false
                    ).flatMap { resp ->
                        DataBufferUtils.join(
                            resp.value.map { bb -> bufferFactory.wrap(bb) } // we get here a lot of small buffers (e.g. 15574 even with chunk size 1MB)
                        ).map { merged -> // need to merge it but it gives us exactly FileRange(start, chunkSize.toLong())
                            val total = merged.readableByteCount()
                            val single = bufferFactory.allocateBuffer(total)
                            single.write(merged)
                            DataBufferUtils.release(merged)
                            single
                        }
                    }
                }
            }