Storage - Content Validation Message Encoder Perf Updates

[ibrandes]

📜 Overview

This PR introduces significant updates to the StructuredMessageEncoder to enhance performance and enable reactive streaming of encoded data. The changes reduce memory overhead from 3 times the size of the payload to around the same size of the payload.

---

🔑 Key Changes

1. Reactive Encoding API

   • Changed the encode method signature from:

     public ByteBuffer encode(ByteBuffer unencodedBuffer)

     to:

     public Flux<ByteBuffer> encode(ByteBuffer unencodedBuffer)

   • Enables streaming of ByteBuffer segments for better memory efficiency and non-blocking I/O.

2. Segmented Encoding Logic

   • Refactored encoding flow to emit:
     • Message header
     • Segment headers and content
     • Segment footers
     • Message footer
   • Introduced helper methods like generateMessageHeader() and generateSegmentFooter() for cleaner composition.

3. Error Handling

   • Updated validation to return Flux.error(...) instead of throwing exceptions directly.

4. Performance Improvements

   • Removed intermediate ByteArrayOutputStream aggregation in favor of streaming buffers.
   • Reduced unnecessary copying and allocations by emitting buffers as soon as they are ready.

---

📦 Other Minor Changes

• StructuredMessageConstants.java: Added constants for default segment size and maximum encoded data length.
• MessageEncoderTests.java: Updated tests for Flux<ByteBuffer> and added large-scale encoding test (bigEncode) for performance validation.

[github-actions]

API Change Check

APIView identified API level changes in this PR and created the following API reviews

com.azure:azure-storage-common

[github-actions]

Hi @ibrandes. Thank you for your interest in helping to improve the Azure SDK experience and for your contribution. We've noticed that there hasn't been recent engagement on this pull request. If this is still an active work stream, please let us know by pushing some changes or leaving a comment. Otherwise, we'll close this out in 7 days.

[ibrandes]

comment

[Copilot]

Pull request overview

This PR updates the storage structured message encoding implementation to expose a reactive Flux<ByteBuffer>-based API and aims to reduce memory overhead by emitting encoded pieces (headers/content/footers) instead of aggregating into a single buffer.

Changes:

• Updated StructuredMessageEncoder.encode to return Flux<ByteBuffer> and adjusted segment header generation logic.
• Added new structured message constants (default segment content length and a maximum encoded-data length).
• Updated encoder unit tests to consume Flux<ByteBuffer> and added a large-scale encoding test.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 5 comments.

File	Description
sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/implementation/structuredmessage/StructuredMessageEncoder.java	Converts encoding to a reactive Flux<ByteBuffer> API and refactors segment header generation and validation flow.
sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/implementation/structuredmessage/StructuredMessageConstants.java	Adds constants for default segment content length and a maximum encoded-data length.
sdk/storage/azure-storage-common/src/test/java/com/azure/storage/common/implementation/structuredmessage/MessageEncoderTests.java	Updates tests to collect bytes from the new Flux<ByteBuffer> API and adds a very large encode test.

Comments suppressed due to low confidence (1)

sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/implementation/structuredmessage/StructuredMessageEncoder.java:183

• This implementation still buffers the entire encoded output into an in-memory List<ByteBuffer> before returning Flux.fromIterable(buffers). That means it isn’t actually streaming (and doesn’t benefit from backpressure), which contradicts the PR description’s claim of reactive streaming and reduced memory overhead. Consider emitting incrementally using Flux.generate/Flux.create (or Flux.push) so buffers can be consumed as they’re produced.

        return Flux.defer(() -> {
            List<ByteBuffer> buffers = new ArrayList<>();

            // if we are at the beginning of the message, encode message header
            if (currentContentOffset == 0) {
                buffers.add(ByteBuffer.wrap(generateMessageHeader()));
            }

            while (unencodedBuffer.hasRemaining()) {
                // if we are at the beginning of a segment's content, encode segment header
                if (currentSegmentOffset == 0) {
                    incrementCurrentSegment();
                    buffers.add(ByteBuffer.wrap(generateSegmentHeader()));
                }

                buffers.add(encodeSegmentContent(unencodedBuffer));

                // if we are at the end of a segment's content, encode segment footer
                if (currentSegmentOffset == getSegmentContentLength()) {
                    byte[] footer = generateSegmentFooter();
                    if (footer.length > 0) {
                        buffers.add(ByteBuffer.wrap(footer));
                    }
                    currentSegmentOffset = 0;
                }
            }

            // if all content has been encoded, encode message footer
            if (currentContentOffset == contentLength) {
                byte[] footer = generateMessageFooter();
                if (footer.length > 0) {
                    buffers.add(ByteBuffer.wrap(footer));
                }
            }

            return Flux.fromIterable(buffers);
        });