Skip to content

Storage - Content Validation Message Encoder Perf Updates #47158

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ibrandes merged 3 commits into from
Feb 25, 2026
Merged

Storage - Content Validation Message Encoder Perf Updates #47158

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
87b3f7b
all perf adjustments
ibrandes Nov 4, 2025
87420bd
Merge branch 'feature/storage/content-validation' into content-valida…
ibrandes Feb 25, 2026
5623071
addressing copilot comments
ibrandes Feb 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions ...com/azure/storage/common/implementation/structuredmessage/StructuredMessageConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,4 +26,19 @@ public final class StructuredMessageConstants {
* The length of the CRC64 checksum.
*/
public static final int CRC64_LENGTH = 8;

/**
* The default length of segments for version 1.
*/
public static final int V1_DEFAULT_SEGMENT_CONTENT_LENGTH = 4 * 1024 * 1024; // 4 MiB

/**
* The maximum amount of data to encode at once.
*/
public static final int STATIC_MAXIMUM_ENCODED_DATA_LENGTH = 4 * 1024 * 1024; // 4 MiB

/**
* The header name for the CRC64 checksum.
*/
public static final String STRUCTURED_BODY_TYPE_VALUE = "XSM/1.0; properties=crc64";
}
13 changes: 5 additions & 8 deletions ...a/com/azure/storage/common/implementation/structuredmessage/StructuredMessageEncoder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@
package com.azure.storage.common.implementation.structuredmessage;

import com.azure.core.util.logging.ClientLogger;
import com.azure.storage.common.implementation.BufferStagingArea;
import com.azure.storage.common.implementation.StorageCrc64Calculator;
import com.azure.storage.common.implementation.StorageImplUtils;
import reactor.core.publisher.Flux;
Expand Down Expand Up @@ -109,7 +108,9 @@ private byte[] generateMessageHeader() {
return buffer.array();
}

private byte[] generateSegmentHeader(int segmentContentSize) {
private byte[] generateSegmentHeader() {
int segmentContentSize = Math.min(segmentSize, contentLength - currentContentOffset);
// 2 byte number, 8 byte size
ByteBuffer buffer = ByteBuffer.allocate(getSegmentHeaderLength()).order(ByteOrder.LITTLE_ENDIAN);
buffer.putShort((short) currentSegmentNumber);
buffer.putLong(segmentContentSize);
Expand All @@ -119,9 +120,6 @@ private byte[] generateSegmentHeader(int segmentContentSize) {

/**
* Encodes the given buffer into a structured message format as a stream of ByteBuffers.
* The encoder maintains mutable state and is designed for single, sequential subscription only.
* Callers should pre-chunk input buffers to appropriate sizes (e.g., using {@link BufferStagingArea}) to
* control memory usage.
*
* @param unencodedBuffer The buffer to be encoded.
* @return A Flux of encoded ByteBuffers.
Comment thread
ibrandes marked this conversation as resolved.
Expand All @@ -146,6 +144,7 @@ public Flux<ByteBuffer> encode(ByteBuffer unencodedBuffer) {
return Flux.empty();
}

// create a list of buffers to store the encoded message
List<ByteBuffer> buffers = new ArrayList<>();

// if we are at the beginning of the message, encode message header
Expand All @@ -157,9 +156,7 @@ public Flux<ByteBuffer> encode(ByteBuffer unencodedBuffer) {
// if we are at the beginning of a segment's content, encode segment header
if (currentSegmentOffset == 0) {
incrementCurrentSegment();
// Calculate actual segment size based on remaining content
int actualSegmentSize = Math.min(segmentSize, contentLength - currentContentOffset);
buffers.add(ByteBuffer.wrap(generateSegmentHeader(actualSegmentSize)));
buffers.add(ByteBuffer.wrap(generateSegmentHeader()));
}

buffers.add(encodeSegmentContent(unencodedBuffer));
Expand Down
67 changes: 42 additions & 25 deletions ...t/java/com/azure/storage/common/implementation/structuredmessage/MessageEncoderTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,27 +3,28 @@

package com.azure.storage.common.implementation.structuredmessage;

import com.azure.core.util.FluxUtil;
import com.azure.storage.common.implementation.StorageCrc64Calculator;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import reactor.test.StepVerifier;
import org.junit.jupiter.api.Disabled;
import reactor.core.publisher.Flux;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.Arrays;
import java.util.Objects;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Stream;

import static com.azure.core.util.FluxUtil.collectBytesInByteBufferStream;
import static com.azure.storage.common.implementation.structuredmessage.StructuredMessageConstants.V1_DEFAULT_SEGMENT_CONTENT_LENGTH;
import static com.azure.storage.common.implementation.structuredmessage.StructuredMessageConstants.CRC64_LENGTH;
import static com.azure.storage.common.implementation.structuredmessage.StructuredMessageConstants.V1_HEADER_LENGTH;
import static com.azure.storage.common.implementation.structuredmessage.StructuredMessageConstants.V1_SEGMENT_HEADER_LENGTH;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
Comment thread
ibrandes marked this conversation as resolved.
Expand Down Expand Up @@ -159,10 +160,11 @@ public void readAll(int size, int segmentSize, StructuredMessageFlags flags) thr

StructuredMessageEncoder structuredMessageEncoder = new StructuredMessageEncoder(size, segmentSize, flags);

byte[] actual = collectBytesInByteBufferStream(structuredMessageEncoder.encode(unencodedBuffer)).block();
byte[] actual
= FluxUtil.collectBytesInByteBufferStream(structuredMessageEncoder.encode(unencodedBuffer)).block();
byte[] expected = buildStructuredMessage(unencodedBuffer, segmentSize, flags).array();

Assertions.assertArrayEquals(expected, actual);
assertArrayEquals(expected, actual);
}

private static Stream<Arguments> readMultipleSupplier() {
Expand Down Expand Up @@ -193,42 +195,40 @@ public void readMultiple(int segmentSize, StructuredMessageFlags flags) throws I

byte[] expected = buildStructuredMessage(allWrappedData, segmentSize, flags).array();

ByteArrayOutputStream allActualData = new ByteArrayOutputStream();
allActualData.write(Objects
.requireNonNull(collectBytesInByteBufferStream(structuredMessageEncoder.encode(wrappedData1)).block()));
allActualData.write(Objects
.requireNonNull(collectBytesInByteBufferStream(structuredMessageEncoder.encode(wrappedData2)).block()));
allActualData.write(Objects
.requireNonNull(collectBytesInByteBufferStream(structuredMessageEncoder.encode(wrappedData3)).block()));
Flux<ByteBuffer> allActualFlux = structuredMessageEncoder.encode(wrappedData1)
.concatWith(structuredMessageEncoder.encode(wrappedData2))
.concatWith(structuredMessageEncoder.encode(wrappedData3));

Assertions.assertArrayEquals(expected, allActualData.toByteArray());
byte[] actual = FluxUtil.collectBytesInByteBufferStream(allActualFlux).block();

assertArrayEquals(expected, actual);
}

@Test
public void emptyBuffer() {
StructuredMessageEncoder encoder = new StructuredMessageEncoder(10, 5, StructuredMessageFlags.NONE);
ByteBuffer emptyBuffer = ByteBuffer.allocate(0);
ByteBuffer result = ByteBuffer
.wrap(Objects.requireNonNull(collectBytesInByteBufferStream(encoder.encode(emptyBuffer)).block()));
assertEquals(0, result.remaining());
byte[] result = FluxUtil.collectBytesInByteBufferStream(encoder.encode(emptyBuffer)).block();
assertNotNull(result);
assertEquals(0, result.length);
}

@Test
public void contentAlreadyEncoded() {
StructuredMessageEncoder encoder = new StructuredMessageEncoder(4, 2, StructuredMessageFlags.NONE);
encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2, 3, 4 })).blockLast();
StepVerifier.create(encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2 })))
.expectError(IllegalArgumentException.class)
.verify();
FluxUtil.collectBytesInByteBufferStream(encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2, 3, 4 }))).block();
assertThrows(IllegalArgumentException.class,
() -> FluxUtil.collectBytesInByteBufferStream(encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2 })))
.block());
}

@Test
public void bufferLengthExceedsContentLength() {
StructuredMessageEncoder encoder = new StructuredMessageEncoder(4, 2, StructuredMessageFlags.NONE);
encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2, 3 })).blockLast();
StepVerifier.create(encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2 })))
.expectError(IllegalArgumentException.class)
.verify();
FluxUtil.collectBytesInByteBufferStream(encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2, 3 }))).block();
assertThrows(IllegalArgumentException.class,
() -> FluxUtil.collectBytesInByteBufferStream(encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2 })))
.block());
}

@Test
Expand All @@ -248,4 +248,21 @@ public void testNumSegmentsExceedsMaxValue() {
assertThrows(IllegalArgumentException.class,
() -> new StructuredMessageEncoder(Integer.MAX_VALUE, 1, StructuredMessageFlags.NONE));
}

@Test
@Disabled("For local testing only")
public void bigEncode() throws IOException {
byte[] data = getRandomData(262144000);

ByteBuffer unencodedBuffer = ByteBuffer.wrap(data);

StructuredMessageEncoder structuredMessageEncoder = new StructuredMessageEncoder(262144000,
V1_DEFAULT_SEGMENT_CONTENT_LENGTH, StructuredMessageFlags.STORAGE_CRC64);

byte[] actual
= FluxUtil.collectBytesInByteBufferStream(structuredMessageEncoder.encode(unencodedBuffer)).block();
byte[] expected = buildStructuredMessage(unencodedBuffer, V1_DEFAULT_SEGMENT_CONTENT_LENGTH,
StructuredMessageFlags.STORAGE_CRC64).array();
assertArrayEquals(expected, actual);
}
}
Loading