diff --git a/sdk/storage/azure-storage-blob/assets.json b/sdk/storage/azure-storage-blob/assets.json index d7718b9a1d4c..7177fb18a721 100644 --- a/sdk/storage/azure-storage-blob/assets.json +++ b/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_cb46d93569" + "Tag": "java/storage/azure-storage-blob_beac25b9c7" } diff --git a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobContentValidationAsyncUploadTests.java b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobContentValidationAsyncUploadTests.java index 44514684002a..1ee2466e1fb6 100644 --- a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobContentValidationAsyncUploadTests.java +++ b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobContentValidationAsyncUploadTests.java @@ -5,6 +5,7 @@ import com.azure.core.http.HttpHeaders; import com.azure.core.util.BinaryData; +import com.azure.core.util.FluxUtil; import com.azure.storage.blob.models.BlobRequestConditions; import com.azure.storage.blob.models.BlobStorageException; import com.azure.storage.blob.models.PageRange; @@ -26,19 +27,23 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.MethodSource; import reactor.core.publisher.Flux; import reactor.test.StepVerifier; import java.io.File; +import java.io.FileInputStream; import java.io.IOException; +import java.io.InputStream; import java.nio.ByteBuffer; +import java.nio.channels.AsynchronousFileChannel; import java.nio.file.Files; -import java.util.Base64; +import java.nio.file.StandardOpenOption; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; +import java.util.Base64; import java.util.Collections; import java.util.List; -import java.util.Random; import java.util.concurrent.CopyOnWriteArrayList; import static org.junit.jupiter.api.Assertions.assertArrayEquals; @@ -57,17 +62,24 @@ public class BlobContentValidationAsyncUploadTests extends BlobTestBase { /* Single-part uploads with length < 4MB use CRC64 header; >= 4MB use structured message. */ private static final int UNDER_4MB = 2 * Constants.MB; - private static final long LARGE_UPLOAD_MIN_BYTES = 500L * Constants.MB; - private static final long LARGE_UPLOAD_MAX_BYTES = Constants.GB; - private static final long LARGE_UPLOAD_BLOCK_SIZE_BYTES = 8L * Constants.MB; - private static final int LARGE_UPLOAD_MAX_CONCURRENCY = 8; + /** + * Live-only random payload band (256–500 MiB, inclusive upper bound via {@code randomLongFromNamer}+1) for + * {@code uploadWithResponse}, {@code uploadFromFileWithResponse}, and single-block {@code stageBlock}. + */ + private static final long LIVE_RANDOM_PARALLEL_PAYLOAD_MIN_BYTES_EXCLUSIVE = 256L * Constants.MB; + private static final long LIVE_RANDOM_PARALLEL_PAYLOAD_MAX_BYTES_INCLUSIVE = 500L * Constants.MB; + + /** + * Live-only random payload band for sequential append-block puts only ({@link + * #appendBlockLiveRandomRoundTripDataIntegrity()}): {@code Flux.concatMap} issues one append REST call per chunk in + * order (not parallel staging); use a smaller band than {@link #LIVE_RANDOM_PARALLEL_PAYLOAD_MIN_BYTES_EXCLUSIVE}. + */ + private static final long LIVE_RANDOM_SEQUENTIAL_APPEND_PAYLOAD_MIN_BYTES_EXCLUSIVE = 32L * Constants.MB; + private static final long LIVE_RANDOM_SEQUENTIAL_APPEND_PAYLOAD_MAX_BYTES_INCLUSIVE = 64L * Constants.MB; private static final String MD5_AND_CRC64_EXCLUSIVE_MESSAGE = "Only one form of transactional content validation may be used."; - private static final String UPLOAD_FROM_FILE_MD5_NOT_SUPPORTED_MESSAGE - = "ContentValidationAlgorithm.MD5 is not supported for uploadFromFile. Use CRC64 or AUTO instead."; - // =========================================================================================== // BlobAsyncClient.uploadWithResponse // =========================================================================================== @@ -119,7 +131,7 @@ public void uploadWithStructuredMessage(ContentValidationAlgorithm algorithm) { } /** - * Multi-part (chunked) upload; content validation uses structured message on each stage block. + * Multipart (chunked) upload; content validation uses structured message on each stage block. */ @LiveOnly // Put Block URLs include random block IDs; not replayable with the test proxy. @ParameterizedTest @@ -261,9 +273,9 @@ public void stageBlockWithCrc64Header(ContentValidationAlgorithm algorithm) { BlockBlobStageBlockOptions options = new BlockBlobStageBlockOptions(getBlockID(), data).setContentValidationAlgorithm(algorithm); - StepVerifier.create(client.stageBlockWithResponse(options)).assertNext(response -> { - assertTrue(hasOnlyCrc64Headers(recorded)); - }).verifyComplete(); + StepVerifier.create(client.stageBlockWithResponse(options)) + .assertNext(response -> assertTrue(hasOnlyCrc64Headers(recorded))) + .verifyComplete(); } @ParameterizedTest @@ -279,9 +291,9 @@ public void stageBlockWithStructuredMessage(ContentValidationAlgorithm algorithm BlockBlobStageBlockOptions options = new BlockBlobStageBlockOptions(getBlockID(), data).setContentValidationAlgorithm(algorithm); - StepVerifier.create(client.stageBlockWithResponse(options)).assertNext(response -> { - assertTrue(hasOnlyStructuredMessageHeaders(recorded)); - }).verifyComplete(); + StepVerifier.create(client.stageBlockWithResponse(options)) + .assertNext(response -> assertTrue(hasOnlyStructuredMessageHeaders(recorded))) + .verifyComplete(); } @Test @@ -296,9 +308,9 @@ public void stageBlockWithNoContentValidation() { BlockBlobStageBlockOptions options = new BlockBlobStageBlockOptions(getBlockID(), data) .setContentValidationAlgorithm(ContentValidationAlgorithm.NONE); - StepVerifier.create(client.stageBlockWithResponse(options)).assertNext(response -> { - assertTrue(hasNoContentValidationHeaders(recorded)); - }).verifyComplete(); + StepVerifier.create(client.stageBlockWithResponse(options)) + .assertNext(response -> assertTrue(hasNoContentValidationHeaders(recorded))) + .verifyComplete(); } // =========================================================================================== @@ -621,10 +633,9 @@ public void uploadWithCrc64RoundTripDataIntegrity() { .setRequestConditions(new BlobRequestConditions()) .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - client.uploadWithResponse(options).block(); - - byte[] downloaded = client.downloadContent().block().toBytes(); - assertArrayEquals(randomData, downloaded, "Downloaded data must match uploaded data (CRC64 header path)"); + StepVerifier.create(client.uploadWithResponse(options).then(client.downloadContent())) + .assertNext(downloaded -> assertArrayEquals(randomData, downloaded.toBytes())) + .verifyComplete(); } @Test @@ -639,10 +650,9 @@ public void uploadWithStructuredMessageRoundTripDataIntegrity() { .setRequestConditions(new BlobRequestConditions()) .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - client.uploadWithResponse(options).block(); - - byte[] downloaded = client.downloadContent().block().toBytes(); - assertArrayEquals(randomData, downloaded, "Downloaded data must match uploaded data (structured message path)"); + StepVerifier.create(client.uploadWithResponse(options).then(client.downloadContent())) + .assertNext(downloaded -> assertArrayEquals(randomData, downloaded.toBytes())) + .verifyComplete(); } @LiveOnly // Put Block URLs include random block IDs; not replayable with the test proxy. @@ -660,11 +670,9 @@ public void uploadChunkedWithStructuredMessageRoundTripDataIntegrity() { .setRequestConditions(new BlobRequestConditions()) .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - client.uploadWithResponse(options).block(); - - byte[] downloaded = client.downloadContent().block().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match uploaded data (chunked structured message path)"); + StepVerifier.create(client.uploadWithResponse(options).then(client.downloadContent())) + .assertNext(downloaded -> assertArrayEquals(randomData, downloaded.toBytes())) + .verifyComplete(); } @Test @@ -678,11 +686,9 @@ public void blockBlobSimpleUploadRoundTripDataIntegrity() { BlockBlobSimpleUploadOptions options = new BlockBlobSimpleUploadOptions(data).setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - client.uploadWithResponse(options).block(); - - byte[] downloaded = blobClient.downloadContent().block().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match uploaded data (block blob simple upload)"); + StepVerifier.create(client.uploadWithResponse(options).then(client.downloadContent())) + .assertNext(downloaded -> assertArrayEquals(randomData, downloaded.toBytes())) + .verifyComplete(); } @Test @@ -697,10 +703,9 @@ public void appendBlockRoundTripDataIntegrity() { AppendBlobAppendBlockOptions options = new AppendBlobAppendBlockOptions(data, TEN_MB) .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - client.appendBlockWithResponse(options).block(); - - byte[] downloaded = blobClient.downloadContent().block().toBytes(); - assertArrayEquals(randomData, downloaded, "Downloaded data must match uploaded data (append block)"); + StepVerifier.create(client.appendBlockWithResponse(options).then(client.downloadContent())) + .assertNext(downloaded -> assertArrayEquals(randomData, downloaded.toBytes())) + .verifyComplete(); } @Test @@ -716,206 +721,263 @@ public void uploadPagesRoundTripDataIntegrity() { = new PageBlobUploadPagesOptions(new PageRange().setStart(0).setEnd(FOUR_MB_PAGE_ALIGNED - 1), data) .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - client.uploadPagesWithResponse(options).block(); - - byte[] downloaded = blobClient.downloadContent().block().toBytes(); - assertArrayEquals(randomData, downloaded, "Downloaded data must match uploaded data (page blob upload pages)"); + StepVerifier.create(client.uploadPagesWithResponse(options).then(client.downloadContent())) + .assertNext(downloaded -> assertArrayEquals(randomData, downloaded.toBytes())) + .verifyComplete(); } // =========================================================================================== - // Randomized payload sizes (exercises CRC64 header vs structured message lengths across runs) + // Live-only random payload bands. + // - 256–500 MiB: parallelUpload, uploadFromFile, stageBlock — parallel staging / giant block; default transfer + // options where applicable. + // - 32–64 MiB (sequential append blocks only): appendBlockLiveRandom… — one append REST call per chunk in order. // =========================================================================================== - @Test - public void uploadWithRandomSizeCrc64HeaderRoundTripDataIntegrity() { - int size = randomIntFromNamer(Constants.KB, EXACTLY_4MB); - - BlobAsyncClient client = createBlobAsyncClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - byte[] randomData = getRandomByteArray(size); - Flux data = Flux.just(ByteBuffer.wrap(randomData)); - - BlobParallelUploadOptions options = new BlobParallelUploadOptions(data) - .setParallelTransferOptions(new ParallelTransferOptions().setMaxSingleUploadSizeLong((long) size)) - .setRequestConditions(new BlobRequestConditions()) - .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - - client.uploadWithResponse(options).block(); - - byte[] downloaded = client.downloadContent().block().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match uploaded data (random size CRC64 header path, size=" + size + ")"); - } - @LiveOnly // This test is too large for the test proxy. @Test - public void uploadWithRandomSizeStructuredMessageRoundTripDataIntegrity() { - int size = randomIntFromNamer(EXACTLY_4MB, 48 * Constants.MB + 1); - - BlobAsyncClient client = createBlobAsyncClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - byte[] randomData = getRandomByteArray(size); - Flux data = Flux.just(ByteBuffer.wrap(randomData)); - - BlobParallelUploadOptions options = new BlobParallelUploadOptions(data) - .setParallelTransferOptions(new ParallelTransferOptions().setMaxSingleUploadSizeLong((long) size)) - .setRequestConditions(new BlobRequestConditions()) - .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - - client.uploadWithResponse(options).block(); - - byte[] downloaded = client.downloadContent().block().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match uploaded data (random size structured message path, size=" + size + ")"); - } - - @LiveOnly // Put Block URLs include random block IDs; not replayable with the test proxy. - @Test - public void uploadChunkedRandomSizesRoundTripDataIntegrity() { - Random rnd = newRandomFromNamer(); - long[] blockSizeChoices = { Constants.MB, 2L * Constants.MB, 4L * Constants.MB, 8L * Constants.MB }; - long blockSize = blockSizeChoices[rnd.nextInt(blockSizeChoices.length)]; - int minTotal = (int) Math.max(24L * Constants.MB, 2 * blockSize); - int totalSize = minTotal + rnd.nextInt(80 * Constants.MB + 1 - minTotal); - - BlobAsyncClient client = createBlobAsyncClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - byte[] randomData = getRandomByteArray(totalSize); - Flux data = Flux.just(ByteBuffer.wrap(randomData)); - - BlobParallelUploadOptions options = new BlobParallelUploadOptions(data) - .setParallelTransferOptions( - new ParallelTransferOptions().setBlockSizeLong(blockSize).setMaxSingleUploadSizeLong(blockSize)) - .setRequestConditions(new BlobRequestConditions()) - .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - - client.uploadWithResponse(options).block(); - - byte[] downloaded = client.downloadContent().block().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match uploaded data (random chunked path, total=" + totalSize + ", block=" + blockSize - + ")"); + public void parallelUploadLiveRandomRoundTripDataIntegrity() throws Exception { + int chosenPayloadSizeBytes = (int) randomLongFromNamer(LIVE_RANDOM_PARALLEL_PAYLOAD_MIN_BYTES_EXCLUSIVE + 1, + LIVE_RANDOM_PARALLEL_PAYLOAD_MAX_BYTES_INCLUSIVE + 1); + try { + String prefix = "chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". "; + BlobAsyncClient client = createBlobAsyncClientWithRequestSniffer(new CopyOnWriteArrayList<>()); + File sourceFile = getRandomFile(chosenPayloadSizeBytes); + File outFile = Files.createTempFile("blob-cv-live-par-dl-async", ".bin").toFile(); + outFile.deleteOnExit(); + + try { + try (InputStream data = new FileInputStream(sourceFile)) { + BlobParallelUploadOptions options + = new BlobParallelUploadOptions(data).setRequestConditions(new BlobRequestConditions()) + .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); + client.uploadWithResponse(options).block(); + } + + client.downloadToFile(outFile.getPath(), true).block(); + assertTrue(compareFiles(sourceFile, outFile, 0, chosenPayloadSizeBytes), prefix); + } finally { + if (!sourceFile.delete()) { + sourceFile.deleteOnExit(); + } + if (!outFile.delete()) { + outFile.deleteOnExit(); + } + } + } catch (Exception e) { + throw new Exception("chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". " + e.getMessage(), e); + } } @LiveOnly // This test is too large for the test proxy. @Test - public void blockBlobSimpleUploadRandomSizeRoundTripDataIntegrity() { - int size = randomIntFromNamer(Constants.KB, 48 * Constants.MB + 1); - - BlobAsyncClient blobClient = createBlobAsyncClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - BlockBlobAsyncClient client = blobClient.getBlockBlobAsyncClient(); - byte[] randomData = getRandomByteArray(size); - BinaryData data = BinaryData.fromBytes(randomData); - - BlockBlobSimpleUploadOptions options - = new BlockBlobSimpleUploadOptions(data).setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - - client.uploadWithResponse(options).block(); - - byte[] downloaded = blobClient.downloadContent().block().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match uploaded data (random block blob simple upload, size=" + size + ")"); + public void stageBlockLiveRandomRoundTripDataIntegrity() throws Exception { + int chosenPayloadSizeBytes = (int) randomLongFromNamer(LIVE_RANDOM_PARALLEL_PAYLOAD_MIN_BYTES_EXCLUSIVE + 1, + LIVE_RANDOM_PARALLEL_PAYLOAD_MAX_BYTES_INCLUSIVE + 1); + try { + String prefix = "chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". "; + BlobAsyncClient blobClient = createBlobAsyncClientWithRequestSniffer(new CopyOnWriteArrayList<>()); + BlockBlobAsyncClient client = blobClient.getBlockBlobAsyncClient(); + String blockId = getBlockID(); + + File sourceFile = getRandomFile(chosenPayloadSizeBytes); + File outFile = Files.createTempFile("blob-cv-live-stage-async-dl", ".bin").toFile(); + outFile.deleteOnExit(); + try { + BinaryData binaryData = BinaryData.fromFile(sourceFile.toPath()); + BlockBlobStageBlockOptions stageOptions = new BlockBlobStageBlockOptions(blockId, binaryData) + .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); + client.stageBlockWithResponse(stageOptions).block(); + client.commitBlockList(Collections.singletonList(blockId)).block(); + blobClient.downloadToFile(outFile.getPath(), true).block(); + assertTrue(compareFiles(sourceFile, outFile, 0, chosenPayloadSizeBytes), prefix); + } finally { + if (!sourceFile.delete()) { + sourceFile.deleteOnExit(); + } + if (!outFile.delete()) { + outFile.deleteOnExit(); + } + } + } catch (Exception e) { + throw new Exception("chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". " + e.getMessage(), e); + } } @LiveOnly // This test is too large for the test proxy. @Test - public void stageBlockRandomSizeRoundTripDataIntegrity() { - int size = randomIntFromNamer(Constants.KB, 40 * Constants.MB + 1); - - BlobAsyncClient blobClient = createBlobAsyncClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - BlockBlobAsyncClient client = blobClient.getBlockBlobAsyncClient(); - String blockId = getBlockID(); - byte[] randomData = getRandomByteArray(size); - BinaryData data = BinaryData.fromBytes(randomData); - - BlockBlobStageBlockOptions options = new BlockBlobStageBlockOptions(blockId, data) - .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - - client.stageBlockWithResponse(options).block(); - client.commitBlockList(Collections.singletonList(blockId)).block(); - - byte[] downloaded = blobClient.downloadContent().block().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match staged block (random size, size=" + size + ")"); + public void appendBlockLiveRandomRoundTripDataIntegrity() throws Exception { + int chosenPayloadSizeBytes + = (int) randomLongFromNamer(LIVE_RANDOM_SEQUENTIAL_APPEND_PAYLOAD_MIN_BYTES_EXCLUSIVE + 1, + LIVE_RANDOM_SEQUENTIAL_APPEND_PAYLOAD_MAX_BYTES_INCLUSIVE + 1); + try { + String prefix = "chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". "; + BlobAsyncClient blobClient = createBlobAsyncClientWithRequestSniffer(new CopyOnWriteArrayList<>()); + AppendBlobAsyncClient client = blobClient.getAppendBlobAsyncClient(); + client.create().block(); + + int maxAppendBlockBytes = client.getMaxAppendBlockBytes(); + File sourceFile = getRandomFile(chosenPayloadSizeBytes); + File outFile = Files.createTempFile("blob-cv-live-append-async-dl", ".bin").toFile(); + outFile.deleteOnExit(); + + try { + try (AsynchronousFileChannel channel + = AsynchronousFileChannel.open(sourceFile.toPath(), StandardOpenOption.READ)) { + FluxUtil.readFile(channel, maxAppendBlockBytes, 0, chosenPayloadSizeBytes).concatMap(bb -> { + AppendBlobAppendBlockOptions appendOptions + = new AppendBlobAppendBlockOptions(Flux.just(bb), bb.remaining()) + .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); + return client.appendBlockWithResponse(appendOptions); + }).then().block(); + } + + blobClient.downloadToFile(outFile.getPath(), true).block(); + assertTrue(compareFiles(sourceFile, outFile, 0, chosenPayloadSizeBytes), prefix); + } finally { + if (!sourceFile.delete()) { + sourceFile.deleteOnExit(); + } + if (!outFile.delete()) { + outFile.deleteOnExit(); + } + } + } catch (Exception e) { + throw new Exception("chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". " + e.getMessage(), e); + } } @LiveOnly // This test is too large for the test proxy. @Test - public void appendBlockRandomSizeRoundTripDataIntegrity() { - int size = randomIntFromNamer(Constants.KB, 80 * Constants.MB + 1); - - BlobAsyncClient blobClient = createBlobAsyncClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - AppendBlobAsyncClient client = blobClient.getAppendBlobAsyncClient(); - client.create().block(); - - byte[] randomData = getRandomByteArray(size); - Flux data = Flux.just(ByteBuffer.wrap(randomData)); - - AppendBlobAppendBlockOptions options = new AppendBlobAppendBlockOptions(data, size) - .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - - client.appendBlockWithResponse(options).block(); - - byte[] downloaded = blobClient.downloadContent().block().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match append block (random size, size=" + size + ")"); + public void uploadFromFileLiveRandomRoundTripDataIntegrity() throws Exception { + int chosenPayloadSizeBytes = (int) randomLongFromNamer(LIVE_RANDOM_PARALLEL_PAYLOAD_MIN_BYTES_EXCLUSIVE + 1, + LIVE_RANDOM_PARALLEL_PAYLOAD_MAX_BYTES_INCLUSIVE + 1); + try { + String prefix = "chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". "; + BlobAsyncClient client = createBlobAsyncClientWithRequestSniffer(new CopyOnWriteArrayList<>()); + File sourceFile = getRandomFile(chosenPayloadSizeBytes); + File outFile = Files.createTempFile("blob-cv-live-uploadfromfile-async-dl", ".bin").toFile(); + outFile.deleteOnExit(); + try { + BlobUploadFromFileOptions options = new BlobUploadFromFileOptions(sourceFile.getAbsolutePath()) + .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); + assertNotNull(client.uploadFromFileWithResponse(options).block().getValue().getETag(), + prefix + "Missing E-Tag on upload-from-file."); + client.downloadToFile(outFile.getPath(), true).block(); + assertTrue(compareFiles(sourceFile, outFile, 0, chosenPayloadSizeBytes), prefix); + } finally { + if (!sourceFile.delete()) { + sourceFile.deleteOnExit(); + } + if (!outFile.delete()) { + outFile.deleteOnExit(); + } + } + } catch (Exception e) { + throw new Exception("chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". " + e.getMessage(), e); + } } - @Test - public void uploadPagesRandomAlignedSizeRoundTripDataIntegrity() { - // Put Page allows at most 4 MiB per request; keep one uploadPages call within that limit. - int minPages = 1; - int maxPages = FOUR_MB_PAGE_ALIGNED / PAGE_BYTES; - int numPages = randomIntFromNamer(minPages, maxPages + 1); - int sizeBytes = numPages * PAGE_BYTES; - - BlobAsyncClient blobClient = createBlobAsyncClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - PageBlobAsyncClient client = blobClient.getPageBlobAsyncClient(); - client.create(sizeBytes).block(); - - byte[] randomData = getRandomByteArray(sizeBytes); - Flux data = Flux.just(ByteBuffer.wrap(randomData)); + // ---------- Deterministic parallel upload (async) ---------- - PageBlobUploadPagesOptions options - = new PageBlobUploadPagesOptions(new PageRange().setStart(0).setEnd(sizeBytes - 1), data) - .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); + @ParameterizedTest + @MethodSource("com.azure.storage.blob.BlobTestBase#fuzzyParallelUploadPutBlobReplayableCases") + public void fuzzyParallelUploadPutBlobReplayableRoundTrip(int payloadBytes, long segmentBytes, int maxConcurrency) + throws IOException { + assertParallelUploadFuzzyRoundTripAsync("putBlobReplay", payloadBytes, segmentBytes, maxConcurrency); + } - client.uploadPagesWithResponse(options).block(); + @LiveOnly // Staging-only cases: Put Block URLs include random IDs. + @ParameterizedTest + @MethodSource("com.azure.storage.blob.BlobTestBase#fuzzyParallelUploadSmallPayloadStagingCases") + public void fuzzyParallelUploadSmallPayloadRoundTripRequiresLiveStaging(int payloadBytes, long segmentBytes, + int maxConcurrency) throws IOException { + assertParallelUploadFuzzyRoundTripAsync("smallPayloadStaging", payloadBytes, segmentBytes, maxConcurrency); + } - byte[] downloaded = blobClient.downloadContent().block().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match page upload (random aligned size, size=" + sizeBytes + ")"); + @LiveOnly // payload > segment for every tuple; always staging/Put Block. + @ParameterizedTest + @MethodSource("com.azure.storage.blob.BlobTestBase#fuzzyParallelUploadSub4MiBCases") + public void fuzzyParallelUploadSubFourMiBBlobRoundTrip(int payloadBytes, long segmentBytes, int maxConcurrency) + throws IOException { + assertParallelUploadFuzzyRoundTripAsync("subFourMiB", payloadBytes, segmentBytes, maxConcurrency); } - // =========================================================================================== - // Large blob uploads (500 MiB–1 GiB) with parallel block staging and concurrency - // =========================================================================================== + @LiveOnly // Staging-only cases. + @ParameterizedTest + @MethodSource("com.azure.storage.blob.BlobTestBase#fuzzyParallelUploadFourMiBBoundaryStagingCases") + public void fuzzyParallelUploadFourMiBBoundaryRoundTripRequiresLiveStaging(int payloadBytes, long segmentBytes, + int maxConcurrency) throws IOException { + assertParallelUploadFuzzyRoundTripAsync("fourMiBBoundaryStaging", payloadBytes, segmentBytes, maxConcurrency); + } - @LiveOnly + @LiveOnly // Chunked uploads only. @ParameterizedTest - @EnumSource(value = ContentValidationAlgorithm.class, names = { "CRC64", "AUTO" }) - public void uploadFromFileLargeBlobChunkedWithConcurrency(ContentValidationAlgorithm algorithm) throws IOException { - int sizeBytes = (int) randomLongFromNamer(LARGE_UPLOAD_MIN_BYTES, LARGE_UPLOAD_MAX_BYTES + 1); - File sourceFile = getRandomFile(sizeBytes); - File outFile = Files.createTempFile("blob-cv-large-dl-async", ".bin").toFile(); - outFile.deleteOnExit(); + @MethodSource("com.azure.storage.blob.BlobTestBase#fuzzyParallelUploadMediumMultiPartCases") + public void fuzzyParallelUploadMediumMultiPartRoundTrip(int payloadBytes, long segmentBytes, int maxConcurrency) + throws IOException { + assertParallelUploadFuzzyRoundTripAsync("mediumMultiPart", payloadBytes, segmentBytes, maxConcurrency); + } - try { - BlobAsyncClient client = createBlobAsyncClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - ParallelTransferOptions parallelTransferOptions - = new ParallelTransferOptions().setBlockSizeLong(LARGE_UPLOAD_BLOCK_SIZE_BYTES) - .setMaxSingleUploadSizeLong(LARGE_UPLOAD_BLOCK_SIZE_BYTES) - .setMaxConcurrency(LARGE_UPLOAD_MAX_CONCURRENCY); + @LiveOnly // Large chunked uploads. + @ParameterizedTest + @MethodSource("com.azure.storage.blob.BlobTestBase#fuzzyParallelUploadLargeMultiPartCases") + public void fuzzyParallelUploadLargeMultiPartRoundTrip(int payloadBytes, long segmentBytes, int maxConcurrency) + throws IOException { + assertParallelUploadFuzzyRoundTripAsync("largeMultiPart", payloadBytes, segmentBytes, maxConcurrency); + } - BlobUploadFromFileOptions options = new BlobUploadFromFileOptions(sourceFile.getAbsolutePath()) - .setParallelTransferOptions(parallelTransferOptions) - .setContentValidationAlgorithm(algorithm); + private void assertParallelUploadFuzzyRoundTripAsync(String caseKind, int payloadBytes, long segmentBytes, + int maxConcurrency) throws IOException { + BlobAsyncClient client = createBlobAsyncClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - assertNotNull(client.uploadFromFileWithResponse(options).block().getValue().getETag()); - client.downloadToFile(outFile.getPath(), true).block(); - assertTrue(compareFiles(sourceFile, outFile, 0, sizeBytes), - "Downloaded file must match source (large chunked upload, size=" + sizeBytes + ")"); - } finally { - if (!sourceFile.delete()) { - sourceFile.deleteOnExit(); + ParallelTransferOptions parallelOptions = new ParallelTransferOptions().setBlockSizeLong(segmentBytes) + .setMaxSingleUploadSizeLong(segmentBytes) + .setMaxConcurrency(maxConcurrency); + + String assertionMessage = "Fuzzy parallel upload [" + caseKind + "] payloadBytes=" + payloadBytes + + ", segmentBytes=" + segmentBytes + ", maxConcurrency=" + maxConcurrency; + + // above this threshold the fuzzy parallel upload helpers stream from a temp source file + // to avoid materializing the full payload twice in heap. + if (payloadBytes >= 96 * Constants.MB) { + File sourceFile = getRandomFile(payloadBytes); + File outFile = Files.createTempFile("blob-cv-fuzzy-parallel-dl-async", ".bin").toFile(); + outFile.deleteOnExit(); + int readChunkSize = (int) Math.min(8L * Constants.MB, Math.max(64 * Constants.KB, segmentBytes)); + AsynchronousFileChannel channel + = AsynchronousFileChannel.open(sourceFile.toPath(), StandardOpenOption.READ); + try { + try { + Flux data = FluxUtil.readFile(channel, readChunkSize, 0, payloadBytes); + BlobParallelUploadOptions options + = new BlobParallelUploadOptions(data).setParallelTransferOptions(parallelOptions) + .setRequestConditions(new BlobRequestConditions()) + .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); + client.uploadWithResponse(options).block(); + } finally { + channel.close(); + } + client.downloadToFile(outFile.getPath(), true).block(); + assertTrue(compareFiles(sourceFile, outFile, 0, payloadBytes), assertionMessage); + } finally { + if (!sourceFile.delete()) { + sourceFile.deleteOnExit(); + } + if (!outFile.delete()) { + outFile.deleteOnExit(); + } } + } else { + byte[] randomData = getRandomByteArray(payloadBytes); + Flux data = Flux.just(ByteBuffer.wrap(randomData)); + BlobParallelUploadOptions options + = new BlobParallelUploadOptions(data).setParallelTransferOptions(parallelOptions) + .setRequestConditions(new BlobRequestConditions()) + .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); + + StepVerifier.create(client.uploadWithResponse(options).then(client.downloadContent())) + .assertNext(downloaded -> assertArrayEquals(randomData, downloaded.toBytes(), assertionMessage)) + .verifyComplete(); } } diff --git a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobContentValidationUploadTests.java b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobContentValidationUploadTests.java index 3f55695f88c4..f038d0dad772 100644 --- a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobContentValidationUploadTests.java +++ b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobContentValidationUploadTests.java @@ -31,19 +31,22 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.MethodSource; import java.io.ByteArrayInputStream; +import java.io.EOFException; import java.io.File; +import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; import java.nio.ByteBuffer; +import java.nio.channels.Channels; import java.nio.file.Files; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; import java.util.Base64; import java.util.Collections; import java.util.List; -import java.util.Random; import java.util.concurrent.CopyOnWriteArrayList; import static org.junit.jupiter.api.Assertions.assertArrayEquals; @@ -62,18 +65,24 @@ public class BlobContentValidationUploadTests extends BlobTestBase { /* single-shot uploads with length < 4MB use CRC64 header; >= 4MB use structured message. */ private static final int UNDER_4MB = 2 * Constants.MB; - /** Chunked uploadFromFile tests spanning 500 MiB–1 GiB (must use parallel block upload; max single Put Blob is lower). */ - private static final long LARGE_UPLOAD_MIN_BYTES = 500L * Constants.MB; - private static final long LARGE_UPLOAD_MAX_BYTES = 1L * Constants.GB; - private static final long LARGE_UPLOAD_BLOCK_SIZE_BYTES = 8L * Constants.MB; - private static final int LARGE_UPLOAD_MAX_CONCURRENCY = 8; + /** + * Live-only random payload band (256–500 MiB, inclusive upper bound via {@code randomLongFromNamer}+1) for + * {@code uploadWithResponse}, {@code uploadFromFileWithResponse}, and single-block {@code stageBlock}. + */ + private static final long LIVE_RANDOM_PARALLEL_PAYLOAD_MIN_BYTES_EXCLUSIVE = 256L * Constants.MB; + private static final long LIVE_RANDOM_PARALLEL_PAYLOAD_MAX_BYTES_INCLUSIVE = 500L * Constants.MB; + + /** + * Live-only random payload band for sequential append-block puts only. + * {@code Flux.concatMap} issues one append REST call per chunk in + * order (not parallel staging); use a smaller band than {@link #LIVE_RANDOM_PARALLEL_PAYLOAD_MIN_BYTES_EXCLUSIVE}. + */ + private static final long LIVE_RANDOM_SEQUENTIAL_APPEND_PAYLOAD_MIN_BYTES_EXCLUSIVE = 32L * Constants.MB; + private static final long LIVE_RANDOM_SEQUENTIAL_APPEND_PAYLOAD_MAX_BYTES_INCLUSIVE = 64L * Constants.MB; private static final String MD5_AND_CRC64_EXCLUSIVE_MESSAGE = "Only one form of transactional content validation may be used."; - private static final String UPLOAD_FROM_FILE_MD5_NOT_SUPPORTED_MESSAGE - = "ContentValidationAlgorithm.MD5 is not supported for uploadFromFile. Use CRC64 or AUTO instead."; - // =========================================================================================== // BlobClient.uploadWithResponse // =========================================================================================== @@ -980,199 +989,335 @@ public void uploadFromFileRoundTripDataIntegrity() throws IOException { } // =========================================================================================== - // Randomized payload sizes (exercises CRC64 header vs structured message lengths across runs) + // Live-only random payload bands. + // - 256–500 MiB: parallelUpload, uploadFromFile, stageBlock, block BlobOutputStream, SeekableByteChannel — + // parallel staging or single giant block / default transfer options as applicable. + // - 32–64 MiB (sequential append blocks only): appendBlobAppendBlocksLiveRandom… — one append REST call per + // chunk in order. // =========================================================================================== + @LiveOnly // This test is too large for the test proxy. @Test - public void uploadWithRandomSizeCrc64HeaderRoundTripDataIntegrity() { - int size = randomIntFromNamer(Constants.KB, EXACTLY_4MB); - - BlobClient client = createBlobClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - byte[] randomData = getRandomByteArray(size); - InputStream data = new ByteArrayInputStream(randomData); - - BlobParallelUploadOptions options = new BlobParallelUploadOptions(data) - .setParallelTransferOptions(new ParallelTransferOptions().setMaxSingleUploadSizeLong((long) size)) - .setRequestConditions(new BlobRequestConditions()) - .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - - client.uploadWithResponse(options, null, Context.NONE); - - byte[] downloaded = client.downloadContent().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match uploaded data (random size CRC64 header path, size=" + size + ")"); + public void parallelUploadLiveRandomRoundTripDataIntegrity() throws Exception { + int chosenPayloadSizeBytes = (int) randomLongFromNamer(LIVE_RANDOM_PARALLEL_PAYLOAD_MIN_BYTES_EXCLUSIVE + 1, + LIVE_RANDOM_PARALLEL_PAYLOAD_MAX_BYTES_INCLUSIVE + 1); + try { + String prefix = "chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". "; + BlobClient client = createBlobClientWithRequestSniffer(new CopyOnWriteArrayList<>()); + File sourceFile = getRandomFile(chosenPayloadSizeBytes); + File outFile = Files.createTempFile("blob-cv-live-par-dl", ".bin").toFile(); + outFile.deleteOnExit(); + try { + try (InputStream data = new FileInputStream(sourceFile)) { + BlobParallelUploadOptions options + = new BlobParallelUploadOptions(data).setRequestConditions(new BlobRequestConditions()) + .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); + client.uploadWithResponse(options, null, Context.NONE); + } + client.downloadToFile(outFile.getPath(), true); + assertTrue(compareFiles(sourceFile, outFile, 0, chosenPayloadSizeBytes), prefix); + } finally { + if (!sourceFile.delete()) { + sourceFile.deleteOnExit(); + } + if (!outFile.delete()) { + outFile.deleteOnExit(); + } + } + } catch (Exception e) { + throw new Exception("chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". " + e.getMessage(), e); + } } @LiveOnly // This test is too large for the test proxy. @Test - public void uploadWithRandomSizeStructuredMessageRoundTripDataIntegrity() { - int size = randomIntFromNamer(EXACTLY_4MB, 48 * Constants.MB + 1); - - BlobClient client = createBlobClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - byte[] randomData = getRandomByteArray(size); - InputStream data = new ByteArrayInputStream(randomData); - - BlobParallelUploadOptions options = new BlobParallelUploadOptions(data) - .setParallelTransferOptions(new ParallelTransferOptions().setMaxSingleUploadSizeLong((long) size)) - .setRequestConditions(new BlobRequestConditions()) - .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - - client.uploadWithResponse(options, null, Context.NONE); - - byte[] downloaded = client.downloadContent().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match uploaded data (random size structured message path, size=" + size + ")"); + public void stageBlockLiveRandomRoundTripDataIntegrity() throws Exception { + int chosenPayloadSizeBytes = (int) randomLongFromNamer(LIVE_RANDOM_PARALLEL_PAYLOAD_MIN_BYTES_EXCLUSIVE + 1, + LIVE_RANDOM_PARALLEL_PAYLOAD_MAX_BYTES_INCLUSIVE + 1); + try { + String prefix = "chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". "; + BlobClient blobClient = createBlobClientWithRequestSniffer(new CopyOnWriteArrayList<>()); + BlockBlobClient client = blobClient.getBlockBlobClient(); + String blockId = getBlockID(); + + File sourceFile = getRandomFile(chosenPayloadSizeBytes); + File outFile = Files.createTempFile("blob-cv-live-stage-dl", ".bin").toFile(); + outFile.deleteOnExit(); + try { + BinaryData binaryData = BinaryData.fromFile(sourceFile.toPath()); + BlockBlobStageBlockOptions stageOptions = new BlockBlobStageBlockOptions(blockId, binaryData) + .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); + client.stageBlockWithResponse(stageOptions, null, Context.NONE); + client.commitBlockList(Collections.singletonList(blockId)); + blobClient.downloadToFile(outFile.getPath(), true); + + assertTrue(compareFiles(sourceFile, outFile, 0, chosenPayloadSizeBytes), prefix); + } finally { + if (!sourceFile.delete()) { + sourceFile.deleteOnExit(); + } + if (!outFile.delete()) { + outFile.deleteOnExit(); + } + } + } catch (Exception e) { + throw new Exception("chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". " + e.getMessage(), e); + } } - @LiveOnly // Put Block URLs include random block IDs; not replayable with the test proxy. + @LiveOnly // This test is too large for the test proxy. @Test - public void uploadChunkedRandomSizesRoundTripDataIntegrity() { - Random rnd = newRandomFromNamer(); - long[] blockSizeChoices = { Constants.MB, 2L * Constants.MB, 4L * Constants.MB, 8L * Constants.MB }; - long blockSize = blockSizeChoices[rnd.nextInt(blockSizeChoices.length)]; - int minTotal = (int) Math.max(24L * Constants.MB, 2 * blockSize); - int totalSize = minTotal + rnd.nextInt(80 * Constants.MB + 1 - minTotal); - - BlobClient client = createBlobClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - byte[] randomData = getRandomByteArray(totalSize); - InputStream data = new ByteArrayInputStream(randomData); - - BlobParallelUploadOptions options = new BlobParallelUploadOptions(data) - .setParallelTransferOptions( - new ParallelTransferOptions().setBlockSizeLong(blockSize).setMaxSingleUploadSizeLong(blockSize)) - .setRequestConditions(new BlobRequestConditions()) - .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - - client.uploadWithResponse(options, null, Context.NONE); - - byte[] downloaded = client.downloadContent().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match uploaded data (random chunked path, total=" + totalSize + ", block=" + blockSize - + ")"); + public void appendBlobAppendBlocksLiveRandomRoundTripDataIntegrity() throws Exception { + int chosenPayloadSizeBytes + = (int) randomLongFromNamer(LIVE_RANDOM_SEQUENTIAL_APPEND_PAYLOAD_MIN_BYTES_EXCLUSIVE + 1, + LIVE_RANDOM_SEQUENTIAL_APPEND_PAYLOAD_MAX_BYTES_INCLUSIVE + 1); + try { + String prefix = "chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". "; + BlobClient blobClient = createBlobClientWithRequestSniffer(new CopyOnWriteArrayList<>()); + AppendBlobClient client = blobClient.getAppendBlobClient(); + File sourceFile = getRandomFile(chosenPayloadSizeBytes); + File outFile = Files.createTempFile("blob-cv-live-append-blocks-dl", ".bin").toFile(); + outFile.deleteOnExit(); + client.create(); + int maxAppendBlockBytes = client.getMaxAppendBlockBytes(); + try { + try (FileInputStream fis = new FileInputStream(sourceFile)) { + long remaining = chosenPayloadSizeBytes; + byte[] buf = new byte[maxAppendBlockBytes]; + while (remaining > 0) { + int chunk = (int) Math.min(maxAppendBlockBytes, remaining); + int totalRead = 0; + while (totalRead < chunk) { + int n = fis.read(buf, totalRead, chunk - totalRead); + if (n == -1) { + throw new EOFException( + prefix + "Unexpected EOF after " + totalRead + " bytes of chunk."); + } + totalRead += n; + } + ByteArrayInputStream chunkStream = new ByteArrayInputStream(buf, 0, chunk); + AppendBlobAppendBlockOptions appendOptions + = new AppendBlobAppendBlockOptions(chunkStream, chunk) + .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); + client.appendBlockWithResponse(appendOptions, null, Context.NONE); + remaining -= chunk; + } + } + blobClient.downloadToFile(outFile.getPath(), true); + assertTrue(compareFiles(sourceFile, outFile, 0, chosenPayloadSizeBytes), prefix); + } finally { + if (!sourceFile.delete()) { + sourceFile.deleteOnExit(); + } + if (!outFile.delete()) { + outFile.deleteOnExit(); + } + } + } catch (Exception e) { + throw new Exception("chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". " + e.getMessage(), e); + } } @LiveOnly // This test is too large for the test proxy. @Test - public void blockBlobSimpleUploadRandomSizeRoundTripDataIntegrity() { - int size = randomIntFromNamer(Constants.KB, 48 * Constants.MB + 1); - - BlobClient blobClient = createBlobClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - BlockBlobClient client = blobClient.getBlockBlobClient(); - byte[] randomData = getRandomByteArray(size); - BinaryData data = BinaryData.fromBytes(randomData); - - BlockBlobSimpleUploadOptions options - = new BlockBlobSimpleUploadOptions(data).setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - - client.uploadWithResponse(options, null, Context.NONE); - - byte[] downloaded = blobClient.downloadContent().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match uploaded data (random block blob simple upload, size=" + size + ")"); + public void uploadFromFileLiveRandomRoundTripDataIntegrity() throws Exception { + int chosenPayloadSizeBytes = (int) randomLongFromNamer(LIVE_RANDOM_PARALLEL_PAYLOAD_MIN_BYTES_EXCLUSIVE + 1, + LIVE_RANDOM_PARALLEL_PAYLOAD_MAX_BYTES_INCLUSIVE + 1); + try { + String prefix = "chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". "; + BlobClient client = createBlobClientWithRequestSniffer(new CopyOnWriteArrayList<>()); + File sourceFile = getRandomFile(chosenPayloadSizeBytes); + File outFile = Files.createTempFile("blob-cv-live-uploadfromfile-dl", ".bin").toFile(); + outFile.deleteOnExit(); + try { + BlobUploadFromFileOptions options = new BlobUploadFromFileOptions(sourceFile.getAbsolutePath()) + .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); + client.uploadFromFileWithResponse(options, null, Context.NONE); + client.downloadToFile(outFile.getPath(), true); + assertTrue(compareFiles(sourceFile, outFile, 0, chosenPayloadSizeBytes), prefix); + } finally { + if (!sourceFile.delete()) { + sourceFile.deleteOnExit(); + } + if (!outFile.delete()) { + outFile.deleteOnExit(); + } + } + } catch (Exception e) { + throw new Exception("chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". " + e.getMessage(), e); + } } @LiveOnly // This test is too large for the test proxy. @Test - public void stageBlockRandomSizeRoundTripDataIntegrity() { - int size = randomIntFromNamer(Constants.KB, 40 * Constants.MB + 1); - - BlobClient blobClient = createBlobClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - BlockBlobClient client = blobClient.getBlockBlobClient(); - String blockId = getBlockID(); - byte[] randomData = getRandomByteArray(size); - BinaryData data = BinaryData.fromBytes(randomData); - - BlockBlobStageBlockOptions options = new BlockBlobStageBlockOptions(blockId, data) - .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - - client.stageBlockWithResponse(options, null, Context.NONE); - client.commitBlockList(Collections.singletonList(blockId)); + public void blockBlobOutputStreamLiveRandomRoundTripDataIntegrity() throws Exception { + int chosenPayloadSizeBytes = (int) randomLongFromNamer(LIVE_RANDOM_PARALLEL_PAYLOAD_MIN_BYTES_EXCLUSIVE + 1, + LIVE_RANDOM_PARALLEL_PAYLOAD_MAX_BYTES_INCLUSIVE + 1); + try { + String prefix = "chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". "; + BlobClient blobClient = createBlobClientWithRequestSniffer(new CopyOnWriteArrayList<>()); + BlockBlobClient client = blobClient.getBlockBlobClient(); - byte[] downloaded = blobClient.downloadContent().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match staged block (random size, size=" + size + ")"); + // Explicit parallel-transfer tuning (8 MiB blocks × concurrency 8) on the stream ingest path. + ParallelTransferOptions parallelTransferOptions + = new ParallelTransferOptions().setBlockSizeLong(8L * Constants.MB) + .setMaxSingleUploadSizeLong(8L * Constants.MB) + .setMaxConcurrency(8); + + File sourceFile = getRandomFile(chosenPayloadSizeBytes); + File outFile = Files.createTempFile("blob-cv-live-block-os-dl", ".bin").toFile(); + outFile.deleteOnExit(); + try { + try (BlobOutputStream outputStream = client.getBlobOutputStream( + new BlockBlobOutputStreamOptions().setParallelTransferOptions(parallelTransferOptions) + .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64))) { + Files.copy(sourceFile.toPath(), outputStream); + } + + blobClient.downloadToFile(outFile.getPath(), true); + assertTrue(compareFiles(sourceFile, outFile, 0, chosenPayloadSizeBytes), prefix); + } finally { + if (!sourceFile.delete()) { + sourceFile.deleteOnExit(); + } + if (!outFile.delete()) { + outFile.deleteOnExit(); + } + } + } catch (Exception e) { + throw new Exception("chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". " + e.getMessage(), e); + } } @LiveOnly // This test is too large for the test proxy. @Test - public void appendBlockRandomSizeRoundTripDataIntegrity() { - int size = randomIntFromNamer(Constants.KB, 80 * Constants.MB + 1); - - BlobClient blobClient = createBlobClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - AppendBlobClient client = blobClient.getAppendBlobClient(); - client.create(); - - byte[] randomData = getRandomByteArray(size); - InputStream data = new ByteArrayInputStream(randomData); - - AppendBlobAppendBlockOptions options = new AppendBlobAppendBlockOptions(data, size) - .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); - - client.appendBlockWithResponse(options, null, Context.NONE); - - byte[] downloaded = blobClient.downloadContent().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match append block (random size, size=" + size + ")"); + public void seekableByteChannelWriteLiveRandomRoundTripDataIntegrity() throws Exception { + int chosenPayloadSizeBytes = (int) randomLongFromNamer(LIVE_RANDOM_PARALLEL_PAYLOAD_MIN_BYTES_EXCLUSIVE + 1, + LIVE_RANDOM_PARALLEL_PAYLOAD_MAX_BYTES_INCLUSIVE + 1); + try { + String prefix = "chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". "; + BlobClient blobClient = createBlobClientWithRequestSniffer(new CopyOnWriteArrayList<>()); + BlockBlobClient client = blobClient.getBlockBlobClient(); + File sourceFile = getRandomFile(chosenPayloadSizeBytes); + File outFile = Files.createTempFile("blob-cv-live-sbc-dl", ".bin").toFile(); + outFile.deleteOnExit(); + try { + try (java.nio.channels.SeekableByteChannel seekableByteChannel + = client.openSeekableByteChannelWrite(new BlockBlobSeekableByteChannelWriteOptions( + BlockBlobSeekableByteChannelWriteOptions.WriteMode.OVERWRITE) + .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64))) { + Files.copy(sourceFile.toPath(), Channels.newOutputStream(seekableByteChannel)); + } + + blobClient.downloadToFile(outFile.getPath(), true); + assertTrue(compareFiles(sourceFile, outFile, 0, chosenPayloadSizeBytes), prefix); + } finally { + if (!sourceFile.delete()) { + sourceFile.deleteOnExit(); + } + if (!outFile.delete()) { + outFile.deleteOnExit(); + } + } + } catch (Exception e) { + throw new Exception("chosenPayloadSizeBytes=" + chosenPayloadSizeBytes + ". " + e.getMessage(), e); + } } - @Test - public void uploadPagesRandomAlignedSizeRoundTripDataIntegrity() { - // Put Page allows at most 4 MiB per request; keep one uploadPages call within that limit. - int minPages = 1; - int maxPages = FOUR_MB_PAGE_ALIGNED / PAGE_BYTES; - int numPages = randomIntFromNamer(minPages, maxPages + 1); - int sizeBytes = numPages * PAGE_BYTES; - - BlobClient blobClient = createBlobClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - PageBlobClient client = blobClient.getPageBlobClient(); - client.create(sizeBytes); + // ---------- Deterministic parallel upload ---------- - byte[] randomData = getRandomByteArray(sizeBytes); - InputStream data = new ByteArrayInputStream(randomData); - - PageBlobUploadPagesOptions options - = new PageBlobUploadPagesOptions(new PageRange().setStart(0).setEnd(sizeBytes - 1), data) - .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); + @ParameterizedTest + @MethodSource("com.azure.storage.blob.BlobTestBase#fuzzyParallelUploadPutBlobReplayableCases") + public void fuzzyParallelUploadPutBlobReplayableRoundTrip(int payloadBytes, long segmentBytes, int maxConcurrency) + throws IOException { + assertParallelUploadFuzzyRoundTrip("putBlobReplay", payloadBytes, segmentBytes, maxConcurrency); + } - client.uploadPagesWithResponse(options, null, Context.NONE); + @LiveOnly // Staging-only cases: Put Block URLs include random IDs + @ParameterizedTest + @MethodSource("com.azure.storage.blob.BlobTestBase#fuzzyParallelUploadSmallPayloadStagingCases") + public void fuzzyParallelUploadSmallPayloadRoundTripRequiresLiveStaging(int payloadBytes, long segmentBytes, + int maxConcurrency) throws IOException { + assertParallelUploadFuzzyRoundTrip("smallPayloadStaging", payloadBytes, segmentBytes, maxConcurrency); + } - byte[] downloaded = blobClient.downloadContent().toBytes(); - assertArrayEquals(randomData, downloaded, - "Downloaded data must match page upload (random aligned size, size=" + sizeBytes + ")"); + @LiveOnly // payload > segment for every tuple; always staging/Put Block. + @ParameterizedTest + @MethodSource("com.azure.storage.blob.BlobTestBase#fuzzyParallelUploadSub4MiBCases") + public void fuzzyParallelUploadSubFourMiBBlobRoundTrip(int payloadBytes, long segmentBytes, int maxConcurrency) + throws IOException { + assertParallelUploadFuzzyRoundTrip("subFourMiB", payloadBytes, segmentBytes, maxConcurrency); } - // =========================================================================================== - // Large blob uploads (500 MiB–1 GiB) with parallel block staging and concurrency - // =========================================================================================== + @LiveOnly // Staging-only cases. + @ParameterizedTest + @MethodSource("com.azure.storage.blob.BlobTestBase#fuzzyParallelUploadFourMiBBoundaryStagingCases") + public void fuzzyParallelUploadFourMiBBoundaryRoundTripRequiresLiveStaging(int payloadBytes, long segmentBytes, + int maxConcurrency) throws IOException { + assertParallelUploadFuzzyRoundTrip("fourMiBBoundaryStaging", payloadBytes, segmentBytes, maxConcurrency); + } - @LiveOnly + @LiveOnly // payload > segment throughout; chunked upload. @ParameterizedTest - @EnumSource(value = ContentValidationAlgorithm.class, names = { "CRC64", "AUTO" }) - public void uploadFromFileLargeBlobChunkedWithConcurrency(ContentValidationAlgorithm algorithm) throws IOException { - int sizeBytes = (int) randomLongFromNamer(LARGE_UPLOAD_MIN_BYTES, LARGE_UPLOAD_MAX_BYTES + 1); - File sourceFile = getRandomFile(sizeBytes); - File outFile = Files.createTempFile("blob-cv-large-dl", ".bin").toFile(); - outFile.deleteOnExit(); + @MethodSource("com.azure.storage.blob.BlobTestBase#fuzzyParallelUploadMediumMultiPartCases") + public void fuzzyParallelUploadMediumMultiPartRoundTrip(int payloadBytes, long segmentBytes, int maxConcurrency) + throws IOException { + assertParallelUploadFuzzyRoundTrip("mediumMultiPart", payloadBytes, segmentBytes, maxConcurrency); + } - try { - BlobClient client = createBlobClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - ParallelTransferOptions parallelTransferOptions - = new ParallelTransferOptions().setBlockSizeLong(LARGE_UPLOAD_BLOCK_SIZE_BYTES) - .setMaxSingleUploadSizeLong(LARGE_UPLOAD_BLOCK_SIZE_BYTES) - .setMaxConcurrency(LARGE_UPLOAD_MAX_CONCURRENCY); + @LiveOnly // payload >> segment throughout; chunked upload / large payloads. + @ParameterizedTest + @MethodSource("com.azure.storage.blob.BlobTestBase#fuzzyParallelUploadLargeMultiPartCases") + public void fuzzyParallelUploadLargeMultiPartRoundTrip(int payloadBytes, long segmentBytes, int maxConcurrency) + throws IOException { + assertParallelUploadFuzzyRoundTrip("largeMultiPart", payloadBytes, segmentBytes, maxConcurrency); + } - BlobUploadFromFileOptions options = new BlobUploadFromFileOptions(sourceFile.getAbsolutePath()) - .setParallelTransferOptions(parallelTransferOptions) - .setContentValidationAlgorithm(algorithm); + private void assertParallelUploadFuzzyRoundTrip(String caseKind, int payloadBytes, long segmentBytes, + int maxConcurrency) throws IOException { + BlobClient client = createBlobClientWithRequestSniffer(new CopyOnWriteArrayList<>()); - assertNotNull(client.uploadFromFileWithResponse(options, null, Context.NONE).getValue().getETag()); - client.downloadToFile(outFile.getPath(), true); - assertTrue(compareFiles(sourceFile, outFile, 0, sizeBytes), - "Downloaded file must match source (large chunked upload, size=" + sizeBytes + ")"); - } finally { - if (!sourceFile.delete()) { - sourceFile.deleteOnExit(); + ParallelTransferOptions parallelOptions = new ParallelTransferOptions().setBlockSizeLong(segmentBytes) + .setMaxSingleUploadSizeLong(segmentBytes) + .setMaxConcurrency(maxConcurrency); + + String assertionMessage = "Fuzzy parallel upload [" + caseKind + "] payloadBytes=" + payloadBytes + + ", segmentBytes=" + segmentBytes + ", maxConcurrency=" + maxConcurrency; + + // above this threshold the fuzzy parallel upload helpers stream from a temp source file + // to avoid materializing the full payload twice in heap. + if (payloadBytes >= 96 * Constants.MB) { + File sourceFile = getRandomFile(payloadBytes); + File outFile = Files.createTempFile("blob-cv-fuzzy-parallel-dl", ".bin").toFile(); + outFile.deleteOnExit(); + try { + try (InputStream data = new FileInputStream(sourceFile)) { + BlobParallelUploadOptions options + = new BlobParallelUploadOptions(data).setParallelTransferOptions(parallelOptions) + .setRequestConditions(new BlobRequestConditions()) + .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); + client.uploadWithResponse(options, null, Context.NONE); + } + client.downloadToFile(outFile.getPath(), true); + assertTrue(compareFiles(sourceFile, outFile, 0, payloadBytes), assertionMessage); + } finally { + if (!sourceFile.delete()) { + sourceFile.deleteOnExit(); + } + if (!outFile.delete()) { + outFile.deleteOnExit(); + } } + } else { + byte[] randomData = getRandomByteArray(payloadBytes); + InputStream data = new ByteArrayInputStream(randomData); + BlobParallelUploadOptions options + = new BlobParallelUploadOptions(data).setParallelTransferOptions(parallelOptions) + .setRequestConditions(new BlobRequestConditions()) + .setContentValidationAlgorithm(ContentValidationAlgorithm.CRC64); + client.uploadWithResponse(options, null, Context.NONE); + byte[] downloaded = client.downloadContent().toBytes(); + assertArrayEquals(randomData, downloaded, assertionMessage); } } diff --git a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobTestBase.java b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobTestBase.java index 36c284cbf976..4804f82e7ee1 100644 --- a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobTestBase.java +++ b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobTestBase.java @@ -1572,6 +1572,131 @@ protected static long expectedStructuredMessageEncodedLengthChunked(int totalUne return sum; } + /** + * Every tuple keeps payloadBytes <= segmentBytes, so the parallel upload path issues a single Put Blob (no + * Put Block IDs), which replays under the test proxy unlike staging-heavy cases. + *

+ * Sizes are deliberately non-power-of-two (e.g. 7 * KB + 3) and use mixed segment ceilings (64 KiB + * through multi-MiB) to catch alignment and buffer edge cases; rows include the exact 4 MiB service boundary + * and several concurrency values (1–8) to exercise parallel request fan-out without live-only staging. + */ + protected static Stream fuzzyParallelUploadPutBlobReplayableCases() { + return Stream.of(Arguments.of(7 * Constants.KB + 3, 64L * Constants.KB, 1), + Arguments.of(7 * Constants.KB + 3, 128L * Constants.KB, 4), + Arguments.of(41 * Constants.KB + 17, 256L * Constants.KB, 1), + Arguments.of(41 * Constants.KB + 17, 256L * Constants.KB, 8), + Arguments.of(199 * Constants.KB + 5, 512L * Constants.KB, 2), + Arguments.of(512 * Constants.KB - 31, 1L * Constants.MB, 8), + Arguments.of(896 * Constants.KB + 101, 1L * Constants.MB, 6), + Arguments.of(4 * Constants.MB, 4L * Constants.MB, 1), + Arguments.of(4 * Constants.MB, 7L * Constants.MB + 919, 3)); + } + + /** + * payloadBytes > segmentBytes, so uploads still go through Put Block staging even though totals are only + * hundreds of KiB—too small for the proxy when block IDs vary per run. + *

+ * One row pairs a ~200 KiB payload with a 64 KiB segment and moderate concurrency; the other uses a + * ~512 KiB payload with a 1 KiB segment to force many tiny blocks (stress scheduling and per-block CRC64 + * framing) without the cost of the large multi-part grids. + */ + protected static Stream fuzzyParallelUploadSmallPayloadStagingCases() { + return Stream.of(Arguments.of(200 * Constants.KB, 64L * Constants.KB, 3), + Arguments.of(512 * Constants.KB - 31, 1L * Constants.KB, 1)); + } + + /** + * payloadBytes > segmentBytes and payloadBytes <= 4 * Constants.MB - 1 (the ceiling field), so the blob + * stays strictly under the 4 MiB transactional CRC64-header path while uploads remain + * chunked—live-only because of Put Block identity churn. + *

+ * Values mix MiB/KiB segment sizes with offsets (e.g. + 19, - 903) so part counts and last-block + * lengths are not powers of two; the last rows hug ceiling with awkward divisors in segmentBytes to + * stress remainder handling near the sub-4 MiB limit. + */ + protected static Stream fuzzyParallelUploadSub4MiBCases() { + final int ceiling = (4 * Constants.MB) - 1; + return Stream.of(Arguments.of(1 * Constants.MB + 1, 1L * Constants.MB, 1), + Arguments.of(1 * Constants.MB + 1, 2L * Constants.KB, 8), + Arguments.of((5 * Constants.MB) / 4 + 19, 256L * Constants.KB, 4), + Arguments.of(2 * Constants.MB - 903, 1L * Constants.MB, 2), + Arguments.of(2 * Constants.MB + 33, 1L * Constants.KB, 1), + Arguments.of(2 * Constants.MB + 33, 1L * Constants.MB, 8), + Arguments.of((11 * Constants.MB) / 4 - 17, 512L * Constants.KB, 6), + Arguments.of(3 * Constants.MB - 777, 2L * Constants.MB, 8), + Arguments.of(3 * Constants.MB - 1, 1L * Constants.MB, 1), Arguments.of(ceiling - 511, 1L * Constants.MB, 4), + Arguments.of(ceiling - 511, 1L * Constants.MB + 511, 2), + Arguments.of(ceiling, (long) (ceiling / 7 + 17), 3), Arguments.of(ceiling, (long) (ceiling / 2 + 1), 8)); + } + + /** + * Centers on 4 * Constants.MB - 1, exactly 4 * Constants.MB, and just above 4 MiB, with segment + * sizes spanning KiB through multi-MiB—exercising the SDK/service boundary where single-shot vs block staging and + * CRC64 header vs structured-message rules flip, while keeping deterministic Put Blob coverage in the replayable + * supplier above. + *

+ * Includes near-boundary payloads (e.g. -8192, +31, +8191 from 4 MiB) so neither total size nor last segment + * length aligns to typical buffer multiples. + */ + protected static Stream fuzzyParallelUploadFourMiBBoundaryStagingCases() { + final int minus = (4 * Constants.MB) - 1; + final int plus = (4 * Constants.MB) + 1; + return Stream.of(Arguments.of(minus, 1L * Constants.MB, 1), Arguments.of(minus, 512L * Constants.KB, 6), + Arguments.of(minus, 2L * Constants.MB, 8), Arguments.of((4 * Constants.MB) - 8192, 1L * Constants.KB, 4), + Arguments.of(4 * Constants.MB, (long) (4 * Constants.MB / 2), 8), + Arguments.of(4 * Constants.MB, 256L * Constants.KB, 2), Arguments.of(plus, 1L * Constants.MB, 1), + Arguments.of(plus, 2L * Constants.MB, 8), Arguments.of(plus, 1L * Constants.KB, 7), + Arguments.of((4 * Constants.MB) + 31, 511L * Constants.KB + 409, 4), + Arguments.of((4 * Constants.MB) + 8191, 3L * Constants.MB - 413, 6)); + } + + /** + * All rows keep payloadBytes > segmentBytes with totals roughly 6–80 MiB—large enough for meaningful parallel + * block fan-out and structured-message segments, but cheaper than {@link #fuzzyParallelUploadLargeMultiPartCases}. + *

+ * Block sizes step through common service limits (1–8 MiB, half-MiB tail values); concurrency 1–8 pairs with + * imbalanced payloads (e.g. 701, 333) to flush merge/retry edge cases. + */ + protected static Stream fuzzyParallelUploadMediumMultiPartCases() { + return Stream.of(Arguments.of(6 * Constants.MB + 701, Constants.MB, 1), + Arguments.of(6 * Constants.MB + 701, 3L * Constants.MB + 271, 4), + Arguments.of(9 * Constants.MB + 333, 2L * Constants.MB, 1), + Arguments.of(9 * Constants.MB + 333, 3L * Constants.MB + 199, 8), + Arguments.of(12 * Constants.MB + 901, 4L * Constants.MB + 901, 2), + Arguments.of(14 * Constants.MB, 500L * Constants.KB + 13, 6), + Arguments.of(18 * Constants.MB - 4021, 5L * Constants.MB - 701, 3), + Arguments.of(24 * Constants.MB, 8L * Constants.MB, 8), + Arguments.of(28 * Constants.MB + 56789, 7L * Constants.MB + 13, 2), + Arguments.of(31 * Constants.MB, 1024L * Constants.KB + 17, 4), + Arguments.of(40 * Constants.MB + 12345, 7L * Constants.MB + 13, 3), + Arguments.of(48 * Constants.MB - 777, 5L * Constants.MB + 809L * Constants.KB, 6), + Arguments.of(56 * Constants.MB + 19, 9L * Constants.MB + 4096, 8), + Arguments.of(72 * Constants.MB, 4L * Constants.MB + 65536, 8), + Arguments.of(80 * Constants.MB + 321, 13L * Constants.MB - 3073, 1)); + } + + /** + * Stresses high block counts and long-running parallel uploads (~96–320 MiB payloads) with service-realistic segment + * sizes (8–61 MiB class) and heavy concurrency. + *

+ * The final rows use named near-256/288/320 MiB totals with irregular byte tails to keep total bytes and + * block remainders off common multiples while still bounding runtime for Live-only CI. + */ + protected static Stream fuzzyParallelUploadLargeMultiPartCases() { + final int payload257MiBPlus = (int) (257L * Constants.MB + 18881); + final int payload288MiBPlus = (int) (288L * Constants.MB + 7777); + final int payload320MiBPlus = (int) (320L * Constants.MB + 1999); + return Stream.of(Arguments.of(96 * Constants.MB + 17, 8L * Constants.MB + 511, 2), + Arguments.of(112 * Constants.MB, 15L * Constants.MB + 4096, 8), + Arguments.of(128 * Constants.MB + 45673, 17L * Constants.MB - 11264 + 173, 4), + Arguments.of(160 * Constants.MB + 12345, 12L * Constants.MB + 8192, 8), + Arguments.of(192 * Constants.MB + 9876, 31L * Constants.MB - 513, 8), + Arguments.of(224 * Constants.MB, 23L * Constants.MB + 524288, 8), + Arguments.of(payload257MiBPlus, 61L * Constants.MB + 23L * Constants.KB, 6), + Arguments.of(payload288MiBPlus, 36L * Constants.MB + 513, 8), + Arguments.of(payload320MiBPlus, 16L * Constants.MB + 511, 8)); + } + private static boolean hasValidStructuredContentLengthHeader(HttpHeaders headers) { String structuredContentLength = headers.getValue("x-ms-structured-content-length"); if (CoreUtils.isNullOrEmpty(structuredContentLength)