Enhance Blob Test Reliability

eng/pipelines/client.yml

@@ -12,7 +12,7 @@ resources:
       endpoint: azure
 
 variables:
-  DefaultOptions: '--batch-mode -Dmaven.wagon.http.pool=false --settings eng/settings.xml'
+  DefaultOptions: '--batch-mode --fail-at-end -Dmaven.wagon.http.pool=false --settings eng/settings.xml'
   LoggingOptions: '-Dorg.slf4j.simpleLogger.defaultLogLevel=error -Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=warn'
 
 jobs:

eng/pipelines/templates/jobs/archetype-sdk-tests.yml

@@ -10,7 +10,7 @@ parameters:
   TestName: LiveTest
   TimeoutInMinutes: 60
   TestStepMavenInputs:
-    options: '--batch-mode --fail-at-end -Dmaven.wagon.http.pool=false -Dsurefire.rerunFailingTestsCount=3 --settings eng/settings.xml'
+    options: '--batch-mode -Dmaven.wagon.http.pool=false -Dsurefire.rerunFailingTestsCount=3 --settings eng/settings.xml'
     mavenOptions: '-Xmx3072m -Dorg.slf4j.simpleLogger.defaultLogLevel=error -Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=warn'
     javaHomeOption: 'JDKVersion'
     jdkVersionOption: '1.8'

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobAsyncClientBase.java

@@ -74,8 +74,8 @@
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
-import static com.azure.core.util.FluxUtil.monoError;
 import static com.azure.core.util.FluxUtil.fluxError;
+import static com.azure.core.util.FluxUtil.monoError;
 import static com.azure.core.util.FluxUtil.withContext;
 import static java.lang.StrictMath.toIntExact;
 
@@ -322,19 +322,19 @@ public PollerFlux<BlobCopyInfo, Void> beginCopy(String sourceUrl, Duration pollI
      * @param metadata Metadata to associate with the destination blob.
      * @param tier {@link AccessTier} for the destination blob.
      * @param priority {@link RehydratePriority} for rehydrating the blob.
-     * @param sourceModifiedRequestConditions {@link RequestConditions} against the source. Standard HTTP
-     * Access conditions related to the modification of data. ETag and LastModifiedTime are used to construct
-     * conditions related to when the blob was changed relative to the given request. The request will fail if the
-     * specified condition is not satisfied.
+     * @param sourceModifiedRequestConditions {@link RequestConditions} against the source. Standard HTTP Access
+     * conditions related to the modification of data. ETag and LastModifiedTime are used to construct conditions
+     * related to when the blob was changed relative to the given request. The request will fail if the specified
+     * condition is not satisfied.
      * @param destRequestConditions {@link BlobRequestConditions} against the destination.
      * @param pollInterval Duration between each poll for the copy status. If none is specified, a default of one second
      * is used.
      * @return A {@link PollerFlux} that polls the blob copy operation until it has completed, has failed, or has been
      * cancelled.
      */
     public PollerFlux<BlobCopyInfo, Void> beginCopy(String sourceUrl, Map<String, String> metadata, AccessTier tier,
-                                        RehydratePriority priority, RequestConditions sourceModifiedRequestConditions,
-                                        BlobRequestConditions destRequestConditions, Duration pollInterval) {
+        RehydratePriority priority, RequestConditions sourceModifiedRequestConditions,
+        BlobRequestConditions destRequestConditions, Duration pollInterval) {
 
         final Duration interval = pollInterval != null ? pollInterval : Duration.ofSeconds(1);
         final RequestConditions sourceModifiedCondition = sourceModifiedRequestConditions == null
@@ -367,9 +367,9 @@ public PollerFlux<BlobCopyInfo, Void> beginCopy(String sourceUrl, Map<String, St
                 }
             },
             (pollingContext, firstResponse) -> {
-                if (firstResponse == null ||  firstResponse.getValue() == null) {
+                if (firstResponse == null || firstResponse.getValue() == null) {
                     return Mono.error(logger.logExceptionAsError(
-                            new IllegalArgumentException("Cannot cancel a poll response that never started.")));
+                        new IllegalArgumentException("Cannot cancel a poll response that never started.")));
                 }
                 final String copyIdentifier = firstResponse.getValue().getCopyId();
 
@@ -385,8 +385,8 @@ public PollerFlux<BlobCopyInfo, Void> beginCopy(String sourceUrl, Map<String, St
     }
 
     private Mono<BlobCopyInfo> onStart(String sourceUrl, Map<String, String> metadata, AccessTier tier,
-            RehydratePriority priority, RequestConditions sourceModifiedRequestConditions,
-            BlobRequestConditions destinationRequestConditions) {
+        RehydratePriority priority, RequestConditions sourceModifiedRequestConditions,
+        BlobRequestConditions destinationRequestConditions) {
         URL url;
         try {
             url = new URL(sourceUrl);
@@ -420,7 +420,7 @@ private Mono<PollResponse<BlobCopyInfo>> onPoll(PollResponse<BlobCopyInfo> pollR
         if (lastInfo == null) {
             logger.warning("BlobCopyInfo does not exist. Activation operation failed.");
             return Mono.just(new PollResponse<>(
-                    LongRunningOperationStatus.fromString("COPY_START_FAILED", true), null));
+                LongRunningOperationStatus.fromString("COPY_START_FAILED", true), null));
         }
 
         return getProperties().map(response -> {
@@ -462,11 +462,11 @@ private Mono<PollResponse<BlobCopyInfo>> onPoll(PollResponse<BlobCopyInfo> pollR
      * <p>For more information, see the
      * <a href="https://docs.microsoft.com/en-us/rest/api/storageservices/abort-copy-blob">Azure Docs</a></p>
      *
+     * @param copyId The id of the copy operation to abort.
+     * @return A reactive response signalling completion.
      * @see #copyFromUrl(String)
      * @see #beginCopy(String, Duration)
      * @see #beginCopy(String, Map, AccessTier, RehydratePriority, RequestConditions, BlobRequestConditions, Duration)
-     * @param copyId The id of the copy operation to abort.
-     * @return A reactive response signalling completion.
      */
     public Mono<Void> abortCopyFromUrl(String copyId) {
         try {
@@ -486,12 +486,12 @@ public Mono<Void> abortCopyFromUrl(String copyId) {
      * <p>For more information, see the
      * <a href="https://docs.microsoft.com/en-us/rest/api/storageservices/abort-copy-blob">Azure Docs</a></p>
      *
-     * @see #copyFromUrl(String)
-     * @see #beginCopy(String, Duration)
-     * @see #beginCopy(String, Map, AccessTier, RehydratePriority, RequestConditions, BlobRequestConditions, Duration)
      * @param copyId The id of the copy operation to abort.
      * @param leaseId The lease ID the active lease on the blob must match.
      * @return A reactive response signalling completion.
+     * @see #copyFromUrl(String)
+     * @see #beginCopy(String, Duration)
+     * @see #beginCopy(String, Map, AccessTier, RehydratePriority, RequestConditions, BlobRequestConditions, Duration)
      */
     public Mono<Response<Void>> abortCopyFromUrlWithResponse(String copyId, String leaseId) {
         try {
@@ -549,8 +549,8 @@ public Mono<String> copyFromUrl(String copySource) {
      * @return A reactive response containing the copy ID for the long running operation.
      */
     public Mono<Response<String>> copyFromUrlWithResponse(String copySource, Map<String, String> metadata,
-            AccessTier tier, RequestConditions sourceModifiedRequestConditions,
-            BlobRequestConditions destRequestConditions) {
+        AccessTier tier, RequestConditions sourceModifiedRequestConditions,
+        BlobRequestConditions destRequestConditions) {
         try {
             return withContext(context -> copyFromUrlWithResponse(copySource, metadata, tier,
                 sourceModifiedRequestConditions, destRequestConditions, context));
@@ -560,8 +560,8 @@ public Mono<Response<String>> copyFromUrlWithResponse(String copySource, Map<Str
     }
 
     Mono<Response<String>> copyFromUrlWithResponse(String copySource, Map<String, String> metadata, AccessTier tier,
-            RequestConditions sourceModifiedRequestConditions, BlobRequestConditions destRequestConditions,
-            Context context) {
+        RequestConditions sourceModifiedRequestConditions, BlobRequestConditions destRequestConditions,
+        Context context) {
         sourceModifiedRequestConditions = sourceModifiedRequestConditions == null
             ? new RequestConditions() : sourceModifiedRequestConditions;
         destRequestConditions = destRequestConditions == null ? new BlobRequestConditions() : destRequestConditions;
@@ -764,44 +764,42 @@ private Mono<Response<BlobProperties>> downloadToFileImpl(AsynchronousFileChanne
          * Downloads the first chunk and gets the size of the data and etag if not specified by the user.
          */
         return getSetupMono(finalRange, finalParallelTransferOptions, downloadRetryOptions, requestConditions,
-           rangeGetContentMd5, context)
-        .flatMap(setupTuple3 -> {
-            long newCount = setupTuple3.getT1();
-            BlobRequestConditions finalConditions = setupTuple3.getT2();
-
-            int numChunks = calculateNumBlocks(newCount, finalParallelTransferOptions.getBlockSize());
-
-            // In case it is an empty blob, this ensures we still actually perform a download operation.
-            numChunks = numChunks == 0 ? 1 : numChunks;
-
-            BlobDownloadAsyncResponse initialResponse = setupTuple3.getT3();
-            return Flux.range(0, numChunks)
-                .flatMap(chunkNum -> {
-                    // The first chunk was retrieved during setup.
-                    if (chunkNum == 0) {
-                        return writeBodyToFile(initialResponse, file, 0, finalParallelTransferOptions, progressLock,
-                            totalProgress);
-                    }
-
-                    // Calculate whether we need a full chunk or something smaller because we are at the end.
-                    long chunkSizeActual = Math.min(finalParallelTransferOptions.getBlockSize(),
-                        newCount - (chunkNum.longValue() * finalParallelTransferOptions.getBlockSize().longValue()));
-                    BlobRange chunkRange = new BlobRange(
-                        finalRange.getOffset()
-                            + (chunkNum.longValue() * finalParallelTransferOptions.getBlockSize().longValue()),
-                        chunkSizeActual);
-
-                    // Make the download call.
-                    return this.downloadWithResponse(chunkRange, downloadRetryOptions, finalConditions,
-                        rangeGetContentMd5, null)
-                        .subscribeOn(Schedulers.elastic())
-                        .flatMap(response ->
-                            writeBodyToFile(response, file, chunkNum, finalParallelTransferOptions, progressLock,
-                                totalProgress));
-                })
-                // Only the first download call returns a value.
-                .then(Mono.just(buildBlobPropertiesResponse(initialResponse)));
-        });
+            rangeGetContentMd5, context)
+            .flatMap(setupTuple3 -> {
+                long newCount = setupTuple3.getT1();
+                BlobRequestConditions finalConditions = setupTuple3.getT2();
+
+                int numChunks = calculateNumBlocks(newCount, finalParallelTransferOptions.getBlockSize());
+
+                // In case it is an empty blob, this ensures we still actually perform a download operation.
+                numChunks = numChunks == 0 ? 1 : numChunks;
+
+                BlobDownloadAsyncResponse initialResponse = setupTuple3.getT3();
+                return Flux.range(0, numChunks)
+                    .flatMap(chunkNum -> {
+                        // The first chunk was retrieved during setup.
+                        if (chunkNum == 0) {
+                            return writeBodyToFile(initialResponse, file, 0, finalParallelTransferOptions, progressLock,
+                                totalProgress);
+                        }
+
+                        // Calculate whether we need a full chunk or something smaller because we are at the end.
+                        long modifier = chunkNum.longValue() * finalParallelTransferOptions.getBlockSize();
+                        long chunkSizeActual = Math.min(finalParallelTransferOptions.getBlockSize(),
+                            newCount - modifier);
+                        BlobRange chunkRange = new BlobRange(finalRange.getOffset() + modifier, chunkSizeActual);
+
+                        // Make the download call.
+                        return this.downloadWithResponse(chunkRange, downloadRetryOptions, finalConditions,
+                            rangeGetContentMd5, null)
+                            .subscribeOn(Schedulers.elastic())
+                            .flatMap(response ->
+                                writeBodyToFile(response, file, chunkNum, finalParallelTransferOptions, progressLock,
+                                    totalProgress));
+                    })
+                    // Only the first download call returns a value.
+                    .then(Mono.just(buildBlobPropertiesResponse(initialResponse)));
+            });
     }
 
     private int calculateNumBlocks(long dataSize, long blockLength) {
@@ -852,10 +850,10 @@ private Mono<Tuple3<Long, BlobRequestConditions, BlobDownloadAsyncResponse>> get
             })
             .onErrorResume(BlobStorageException.class, blobStorageException -> {
                 /*
-                In the case of an empty blob, we still want to report success and give back valid headers. Attempting a
-                range download on an empty blob will return an InvalidRange error code and a Content-Range header of the
-                format "bytes * /0". We need to double check that the total size is zero in the case that the customer
-                has attempted an invalid range on a non-zero length blob.
+                 * In the case of an empty blob, we still want to report success and give back valid headers.
+                 * Attempting a range download on an empty blob will return an InvalidRange error code and a
+                 * Content-Range header of the format "bytes * /0". We need to double check that the total size is zero
+                 * in the case that the customer has attempted an invalid range on a non-zero length blob.
                  */
                 if (blobStorageException.getErrorCode() == BlobErrorCode.INVALID_RANGE
                     && extractTotalBlobLength(blobStorageException.getResponse()
@@ -876,6 +874,7 @@ && extractTotalBlobLength(blobStorageException.getResponse()
                             return Mono.zip(Mono.just(0L), Mono.just(requestConditions), Mono.just(response));
                         });
                 }
+
                 return Mono.error(blobStorageException);
             });
     }
@@ -940,7 +939,7 @@ private void downloadToFileCleanup(AsynchronousFileChannel channel, String fileP
         try {
             channel.close();
             if (!signalType.equals(SignalType.ON_COMPLETE)) {
-                Files.delete(Paths.get(filePath));
+                Files.deleteIfExists(Paths.get(filePath));
                 logger.verbose("Downloading to file failed. Cleaning up resources.");
             }
         } catch (IOException e) {