Fix BlobBatch Not Emitting Responses (Azure#23304)

Fix BlobBatch Not Emitting Responses

Author: alzimmermsft

sdk/storage/azure-storage-blob-batch/src/main/java/com/azure/storage/blob/batch/BlobBatchHelper.java

@@ -51,6 +51,16 @@ class BlobBatchHelper {
     private static final Pattern APPLICATION_HTTP_PATTERN = Pattern
         .compile("application\\/http", Pattern.CASE_INSENSITIVE);
 
+    /*
+     * The following patterns were previously used in 'String.split' calls. 'String.split' internally compiles the
+     * String pattern into a Pattern regex if it is larger than one character, or two if-and-only-if it is an escaped
+     * single character. Compiling these patterns here will help reduce the number of regex compilations, greatly
+     * improving performance.
+     */
+    private static final Pattern HTTP_NEWLINE_PATTERN = Pattern.compile(HTTP_NEWLINE);
+    private static final Pattern HTTP_DOUBLE_NEWLINE_PATTERN = Pattern.compile(HTTP_NEWLINE + HTTP_NEWLINE_PATTERN);
+    private static final Pattern HTTP_HEADER_SPLIT_PATTERN = Pattern.compile(":\\s*");
+
     // This method connects the batch response values to the individual batch operations based on their Content-Id
     static Mono<SimpleResponse<Void>> mapBatchResponse(BlobBatchOperationInfo batchOperationInfo,
         Response<Flux<ByteBuffer>> rawResponse, boolean throwOnAnyFailure, ClientLogger logger) {
@@ -70,19 +80,29 @@ static Mono<SimpleResponse<Void>> mapBatchResponse(BlobBatchOperationInfo batchO
         String boundary = boundaryPieces[1];
 
         return FluxUtil.collectBytesInByteBufferStream(rawResponse.getValue())
-            .flatMap(byteArrayBody -> Mono.fromRunnable(() -> {
+            /*
+             * This has been changed from using 'Mono.fromRunnable' to 'Mono.create' to resolve an issue where iterating
+             * the responses resulted in 0 responses being returned. The reason that this occurred is that
+             * 'Mono.fromRunnable' only returns an 'onComplete' trigger with no response value, resulting on all
+             * downstream operators being skipped as there is no 'onNext' trigger. Effectively, the request and response
+             * worked correctly but emitting the responses failed.
+             *
+             * This change has an additional benefit in that the MonoSink used in 'Mono.create' now allows for 'onError'
+             * emissions to occur instead of escaping the reactive stream boundaries when throwing an exception.
+             */
+            .flatMap(byteArrayBody -> Mono.create(sink -> {
                 String body = new String(byteArrayBody, StandardCharsets.UTF_8);
                 List<BlobStorageException> exceptions = new ArrayList<>();
 
                 String[] subResponses = body.split("--" + boundary);
                 if (subResponses.length == 3 && batchOperationInfo.getOperationCount() != 1) {
-                    String[] exceptionSections = subResponses[1].split(HTTP_NEWLINE + HTTP_NEWLINE);
+                    String[] exceptionSections = HTTP_DOUBLE_NEWLINE_PATTERN.split(subResponses[1]);
                     int statusCode = getStatusCode(exceptionSections[1], logger);
                     HttpHeaders headers = getHttpHeaders(exceptionSections[1]);
 
-                    throw logger.logExceptionAsError(new BlobStorageException(
+                    sink.error(logger.logExceptionAsError(new BlobStorageException(
                         headers.getValue(Constants.HeaderConstants.ERROR_CODE),
-                        createHttpResponse(rawResponse.getRequest(), statusCode, headers, body), body));
+                        createHttpResponse(rawResponse.getRequest(), statusCode, headers, body), body)));
                 }
 
                 // Split the batch response body into batch operation responses.
@@ -93,7 +113,7 @@ static Mono<SimpleResponse<Void>> mapBatchResponse(BlobBatchOperationInfo batchO
                     }
 
                     // The batch operation response will be delimited by two new lines.
-                    String[] subResponseSections = subResponse.split(HTTP_NEWLINE + HTTP_NEWLINE);
+                    String[] subResponseSections = HTTP_DOUBLE_NEWLINE_PATTERN.split(subResponse);
 
                     // The first section will contain batching metadata.
                     BlobBatchOperationResponse<?> batchOperationResponse =
@@ -111,11 +131,11 @@ static Mono<SimpleResponse<Void>> mapBatchResponse(BlobBatchOperationInfo batchO
                 }
 
                 if (throwOnAnyFailure && exceptions.size() != 0) {
-                    throw logger.logExceptionAsError(new BlobBatchStorageException("Batch had operation failures.",
-                        createHttpResponse(rawResponse), exceptions));
+                    sink.error(logger.logExceptionAsError(new BlobBatchStorageException("Batch had operation failures.",
+                        createHttpResponse(rawResponse), exceptions)));
                 }
 
-                new SimpleResponse<>(rawResponse, null);
+                sink.success(new SimpleResponse<>(rawResponse, null));
             }));
     }
 
@@ -146,16 +166,16 @@ private static int getStatusCode(String responseMetadata, ClientLogger logger) {
     private static HttpHeaders getHttpHeaders(String responseMetadata) {
         HttpHeaders headers = new HttpHeaders();
 
-        for (String line : responseMetadata.split(HTTP_NEWLINE)) {
+        for (String line : HTTP_NEWLINE_PATTERN.split(responseMetadata)) {
             if (CoreUtils.isNullOrEmpty(line) || (line.startsWith("HTTP") && !line.contains(":"))) {
                 continue;
             }
 
-            String[] headerPieces = line.split(":\\s*", 2);
+            String[] headerPieces = HTTP_HEADER_SPLIT_PATTERN.split(line, 2);
             if (headerPieces.length == 1) {
-                headers.put(headerPieces[0], null);
+                headers.set(headerPieces[0], (String) null);
             } else {
-                headers.put(headerPieces[0], headerPieces[1]);
+                headers.set(headerPieces[0], headerPieces[1]);
             }
         }
 
@@ -165,7 +185,7 @@ private static HttpHeaders getHttpHeaders(String responseMetadata) {
     private static void setBodyOrAddException(BlobBatchOperationResponse<?> batchOperationResponse,
         String responseBody, List<BlobStorageException> exceptions, ClientLogger logger) {
         /*
-         * Currently no batching operations will return a success body, they will only return a body on an exception.
+         * Currently, no batching operations will return a success body, they will only return a body on an exception.
          * For now this will only construct the exception and throw if it should throw on an error.
          */
         BlobStorageException exception = new BlobStorageException(responseBody,

sdk/storage/azure-storage-blob-batch/src/test/java/com/azure/storage/blob/batch/BatchAPITest.groovy

@@ -1,30 +1,29 @@
 package com.azure.storage.blob.batch
 
-import com.azure.core.http.HttpPipelineBuilder
-import com.azure.core.http.policy.HttpPipelinePolicy
+
 import com.azure.core.http.rest.Response
 import com.azure.core.test.TestMode
 import com.azure.core.util.Context
-import com.azure.storage.blob.BlobServiceAsyncClient
 import com.azure.storage.blob.BlobServiceVersion
 import com.azure.storage.blob.batch.options.BlobBatchSetBlobAccessTierOptions
 import com.azure.storage.blob.models.AccessTier
 import com.azure.storage.blob.models.BlobStorageException
 import com.azure.storage.blob.models.DeleteSnapshotsOptionType
 import com.azure.storage.blob.models.RehydratePriority
 import com.azure.storage.blob.sas.BlobContainerSasPermission
-import com.azure.storage.blob.sas.BlobSasPermission
 import com.azure.storage.blob.sas.BlobServiceSasSignatureValues
 import com.azure.storage.common.sas.AccountSasPermission
 import com.azure.storage.common.sas.AccountSasResourceType
 import com.azure.storage.common.sas.AccountSasService
 import com.azure.storage.common.sas.AccountSasSignatureValues
-import com.azure.storage.common.sas.SasIpRange
 import com.azure.storage.common.sas.SasProtocol
 import com.azure.storage.common.test.shared.extensions.RequiredServiceVersion
+import reactor.test.StepVerifier
 import spock.lang.Unroll
 
 import java.nio.charset.StandardCharsets
+import java.time.Duration
+import java.util.stream.Collectors
 
 class BatchAPITest extends APISpec {
     /*
@@ -33,6 +32,7 @@ class BatchAPITest extends APISpec {
      * as these requests will be properly associated to the response by their `Content-ID` but this causes issues in
      * playback as we are using a static response that cannot handle changes in operation order.
      */
+
     static def assertExpectedOrException(Response<?> response, int expectedStatusCode) {
         try {
             def statusCode = response.getStatusCode()
@@ -45,10 +45,13 @@ class BatchAPITest extends APISpec {
     }
 
     BlobBatchClient batchClient
+    BlobBatchAsyncClient batchAsyncClient
     BlobBatchClient oauthBatchClient
 
     def setup() {
-        batchClient = new BlobBatchClientBuilder(primaryBlobServiceClient).buildClient()
+        def blobBatchClientBuilder = new BlobBatchClientBuilder(primaryBlobServiceAsyncClient)
+        batchClient = blobBatchClientBuilder.buildClient()
+        batchAsyncClient = blobBatchClientBuilder.buildAsyncClient()
         oauthBatchClient = new BlobBatchClientBuilder(getOAuthServiceClient()).buildClient()
     }
 
@@ -189,9 +192,39 @@ class BatchAPITest extends APISpec {
         primaryBlobServiceClient.deleteBlobContainer(containerName)
 
         where:
-        leaseId         | tags
-        garbageLeaseID  | null
-        null            | "\"notfoo\" = 'notbar'"
+        leaseId        | tags
+        garbageLeaseID | null
+        null           | "\"notfoo\" = 'notbar'"
+    }
+
+    // Ensures errors in the batch using BlobBatchAsyncClient are emitted as onError and are not thrown.
+    @Unroll
+    def "Set tier AC fail async"() {
+        setup:
+        def containerName = generateContainerName()
+        def blobName1 = generateBlobName()
+        def batch = batchAsyncClient.getBlobBatch()
+        def containerClient = primaryBlobServiceClient.createBlobContainer(containerName)
+        def blobClient1 = containerClient.getBlobClient(blobName1)
+        blobClient1.getBlockBlobClient().upload(data.defaultInputStream, data.defaultDataSize)
+
+        when:
+        batch.setBlobAccessTier(new BlobBatchSetBlobAccessTierOptions(blobClient1.getBlobUrl(), AccessTier.HOT)
+            .setLeaseId(leaseId).setTagsConditions(tags))
+        def request = batchAsyncClient.submitBatch(batch)
+
+        then:
+        StepVerifier.create(request)
+            .expectError(BlobBatchStorageException.class)
+            .verify(Duration.ofSeconds(30))
+
+        cleanup:
+        primaryBlobServiceClient.deleteBlobContainer(containerName)
+
+        where:
+        leaseId        | tags
+        garbageLeaseID | null
+        null           | "\"notfoo\" = 'notbar'"
     }
 
     def "Set tier some succeed throw on any error"() {
@@ -227,6 +260,41 @@ class BatchAPITest extends APISpec {
         primaryBlobServiceClient.deleteBlobContainer(containerName)
     }
 
+    def "Set tier some succeed throw on any error async"() {
+        setup:
+        def containerName = generateContainerName()
+        def blobName1 = generateBlobName()
+        def blobName2 = generateBlobName()
+        def batch = batchAsyncClient.getBlobBatch()
+        def containerClient = primaryBlobServiceClient.createBlobContainer(containerName)
+        containerClient.getBlobClient(blobName1).getBlockBlobClient().upload(data.defaultInputStream, data.defaultDataSize)
+
+        when:
+        def response1 = batch.setBlobAccessTier(containerName, blobName1, AccessTier.HOT)
+        def response2 = batch.setBlobAccessTier(containerName, blobName2, AccessTier.COOL)
+        def request = batchAsyncClient.submitBatch(batch)
+
+        then:
+        StepVerifier.create(request)
+            .expectError(BlobBatchStorageException.class)
+            .verify(Duration.ofSeconds(30))
+
+        // In PLAYBACK check responses in an order invariant fashion.
+        if (env.testMode == TestMode.PLAYBACK) {
+            assert (assertExpectedOrException(response1, 200) + assertExpectedOrException(response2, 200)) == 1
+        } else {
+            assert response1.getStatusCode() == 200
+            try {
+                response2.getStatusCode()
+            } catch (def exception) {
+                assert exception instanceof BlobStorageException
+            }
+        }
+
+        cleanup:
+        primaryBlobServiceClient.deleteBlobContainer(containerName)
+    }
+
     def "Set tier some succeed do not throw on any error"() {
         setup:
         def containerName = generateContainerName()
@@ -507,7 +575,9 @@ class BatchAPITest extends APISpec {
         def responses = batchClient.deleteBlobs(blobUrls, DeleteSnapshotsOptionType.INCLUDE)
 
         then:
-        for (def response : responses) {
+        def responseList = responses.stream().collect(Collectors.toList())
+        assert responseList.size() == 10
+        for (def response : responseList) {
             assert response.getStatusCode() == 202
         }
 
@@ -530,7 +600,9 @@ class BatchAPITest extends APISpec {
         def responses = batchClient.setBlobsAccessTier(blobUrls, AccessTier.HOT)
 
         then:
-        for (def response : responses) {
+        def responseList = responses.stream().collect(Collectors.toList())
+        assert responseList.size() == 10
+        for (def response : responseList) {
             assert response.getStatusCode() == 200
         }
 
@@ -553,7 +625,9 @@ class BatchAPITest extends APISpec {
         def responses = batchClient.setBlobsAccessTier(blobUrls, AccessTier.HOT)
 
         then:
-        for (def response : responses) {
+        def responseList = responses.stream().collect(Collectors.toList())
+        assert responseList.size() == 1
+        for (def response : responseList) {
             assert response.getStatusCode() == 200
         }
 
@@ -579,7 +653,9 @@ class BatchAPITest extends APISpec {
         def responses = batchClient.setBlobsAccessTier(blobUrls, AccessTier.HOT)
 
         then:
-        for (def response : responses) {
+        def responseList = responses.stream().collect(Collectors.toList())
+        assert responseList.size() == 1
+        for (def response : responseList) {
             assert response.getStatusCode() == 200
         }
 
@@ -601,7 +677,8 @@ class BatchAPITest extends APISpec {
         batchClient.deleteBlobs(blobUrls, DeleteSnapshotsOptionType.INCLUDE).iterator().next()
 
         then:
-        thrown(BlobStorageException)
+        def ex = thrown(RuntimeException)
+        assert ex instanceof BlobStorageException || ex.getCause() instanceof BlobStorageException
 
         cleanup:
         primaryBlobServiceClient.deleteBlobContainer(containerName)
@@ -711,7 +788,7 @@ class BatchAPITest extends APISpec {
         def sas = primaryBlobServiceClient.generateAccountSas(sasValues)
 
         def batchClient = new BlobBatchClientBuilder(getServiceClient(sas, primaryBlobServiceClient.getAccountUrl()))
-                    .buildClient()
+            .buildClient()
 
         def batch = batchClient.getBlobBatch()