[Storage] Fix flaky etag tests. (Azure#23656)

kasobol-msft · web-flow · commit 6bd4054d57ef · 2021-08-18T16:44:06.000-07:00
* fix these tests... * update comment. * in case retries kick in in the setup. * name stress jobs. * Revert "name stress jobs." This reverts commit 9c3bb8f.
diff --git a/sdk/storage/azure-storage-blob-cryptography/src/test/java/com/azure/storage/blob/specialized/cryptography/EncyptedBlockBlobAPITest.groovy b/sdk/storage/azure-storage-blob-cryptography/src/test/java/com/azure/storage/blob/specialized/cryptography/EncyptedBlockBlobAPITest.groovy
@@ -8,6 +8,7 @@ import com.azure.core.http.HttpPipelinePosition
 import com.azure.core.http.HttpResponse
 import com.azure.core.http.policy.HttpPipelinePolicy
 import com.azure.identity.DefaultAzureCredentialBuilder
+import com.azure.storage.blob.BlobClientBuilder
 import com.azure.storage.blob.BlobContainerClient
 import com.azure.storage.blob.BlobServiceClientBuilder
 import com.azure.storage.blob.BlobUrlParts
@@ -51,6 +52,7 @@ import java.nio.file.Files
 import java.nio.file.OpenOption
 import java.nio.file.StandardOpenOption
 import java.time.Duration
+import java.util.concurrent.atomic.AtomicInteger
 
 class EncyptedBlockBlobAPITest extends APISpec {
 
@@ -1241,9 +1243,29 @@ class EncyptedBlockBlobAPITest extends APISpec {
         def outFile = new File(namer.getResourcePrefix())
         Files.deleteIfExists(file.toPath())
 
+        def counter = new AtomicInteger()
+
         expect:
-        def bac = getEncryptedClientBuilder(fakeKey, null, env.primaryAccount.credential,
+        def bacUploading = getEncryptedClientBuilder(fakeKey, null, env.primaryAccount.credential,
+            ebc.getBlobUrl().toString())
+            .buildEncryptedBlobAsyncClient()
+
+        def bacDownloading = getEncryptedClientBuilder(fakeKey, null, env.primaryAccount.credential,
             ebc.getBlobUrl().toString())
+            .addPolicy({ context, next ->
+                return next.process()
+                    .flatMap({ r ->
+                        if (counter.incrementAndGet() == 1) {
+                            /*
+                             * When the download begins trigger an upload to overwrite the downloading blob
+                             * so that the download is able to get an ETag before it is changed.
+                             */
+                            return bacUploading.upload(data.defaultFlux, null, true)
+                                .thenReturn(r)
+                        }
+                        return Mono.just(r)
+                    })
+            })
             .buildEncryptedBlobAsyncClient()
 
         /*
@@ -1261,12 +1283,7 @@ class EncyptedBlockBlobAPITest extends APISpec {
          */
         Hooks.onErrorDropped({ ignored -> /* do nothing with it */ })
 
-        /*
-         * When the download begins trigger an upload to overwrite the downloading blob after waiting 500 milliseconds
-         * so that the download is able to get an ETag before it is changed.
-         */
-        StepVerifier.create(bac.downloadToFileWithResponse(outFile.toPath().toString(), null, options, null, null, false)
-            .doOnSubscribe({ bac.upload(data.defaultFlux, null, true).delaySubscription(Duration.ofMillis(500)).subscribe() }))
+        StepVerifier.create(bacDownloading.downloadToFileWithResponse(outFile.toPath().toString(), null, options, null, null, false))
             .verifyErrorSatisfies({
                 /*
                  * If an operation is running on multiple threads and multiple return an exception Reactor will combine
diff --git a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/APISpec.groovy b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/APISpec.groovy
@@ -3,7 +3,6 @@
 
 package com.azure.storage.blob
 
-import com.azure.core.credential.TokenRequestContext
 import com.azure.core.http.HttpHeaders
 import com.azure.core.http.HttpMethod
 import com.azure.core.http.HttpPipelineCallContext
@@ -17,7 +16,9 @@ import com.azure.core.test.TestMode
 import com.azure.core.util.CoreUtils
 import com.azure.core.util.FluxUtil
 import com.azure.identity.EnvironmentCredentialBuilder
+import com.azure.storage.blob.models.BlobErrorCode
 import com.azure.storage.blob.models.BlobProperties
+import com.azure.storage.blob.models.BlobStorageException
 import com.azure.storage.blob.models.CopyStatusType
 import com.azure.storage.blob.models.LeaseStateType
 import com.azure.storage.blob.models.ListBlobContainersOptions
@@ -32,7 +33,6 @@ import com.azure.storage.common.implementation.Constants
 import com.azure.storage.common.policy.RequestRetryOptions
 import com.azure.storage.common.test.shared.StorageSpec
 import com.azure.storage.common.test.shared.TestAccount
-import com.azure.storage.common.test.shared.policy.MockDownloadHttpResponse
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import spock.lang.Timeout
@@ -106,7 +106,7 @@ class APISpec extends StorageSpec {
         containerName = generateContainerName()
         cc = primaryBlobServiceClient.getBlobContainerClient(containerName)
         ccAsync = primaryBlobServiceAsyncClient.getBlobContainerAsyncClient(containerName)
-        cc.create()
+        ignoreErrors({ cc.create() }, BlobErrorCode.CONTAINER_ALREADY_EXISTS)
     }
 
     def cleanup() {
@@ -125,7 +125,7 @@ class APISpec extends StorageSpec {
                     createLeaseClient(containerClient).breakLeaseWithResponse(new BlobBreakLeaseOptions().setBreakPeriod(Duration.ofSeconds(0)), null, null)
                 }
 
-                containerClient.delete()
+                ignoreErrors({ containerClient.delete() }, BlobErrorCode.CONTAINER_NOT_FOUND)
             }
         }
     }
@@ -672,6 +672,16 @@ class APISpec extends StorageSpec {
         }
     }
 
+    def ignoreErrors(Closure closure, BlobErrorCode... errors) {
+        try {
+            closure.call()
+        } catch (BlobStorageException ex) {
+            if (!errors.contains(ex.errorCode)) {
+                throw ex
+            }
+        }
+    }
+
     /**
      * Injects one retry-able IOException failure per url.
      */
diff --git a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobAPITest.groovy b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobAPITest.groovy
@@ -44,14 +44,14 @@ import com.azure.storage.blob.specialized.BlobClientBase
 import com.azure.storage.blob.specialized.SpecializedBlobClientBuilder
 import com.azure.storage.common.Utility
 import com.azure.storage.common.implementation.Constants
-import com.azure.storage.common.test.shared.TestHttpClientType
 import com.azure.storage.common.test.shared.extensions.LiveOnly
 import com.azure.storage.common.test.shared.extensions.PlaybackOnly
 import com.azure.storage.common.test.shared.extensions.RequiredServiceVersion
 import com.azure.storage.common.test.shared.policy.MockFailureResponsePolicy
 import com.azure.storage.common.test.shared.policy.MockRetryRangeResponsePolicy
 import reactor.core.Exceptions
 import reactor.core.publisher.Hooks
+import reactor.core.publisher.Mono
 import reactor.test.StepVerifier
 import spock.lang.IgnoreIf
 import spock.lang.Unroll
@@ -66,6 +66,7 @@ import java.nio.file.StandardOpenOption
 import java.security.MessageDigest
 import java.time.Duration
 import java.time.OffsetDateTime
+import java.util.concurrent.atomic.AtomicInteger
 
 class BlobAPITest extends APISpec {
     BlobClient bc
@@ -1142,11 +1143,32 @@ class BlobAPITest extends APISpec {
         bc.uploadFromFile(file.toPath().toString(), true)
         def outFile = new File(namer.getResourcePrefix())
         Files.deleteIfExists(file.toPath())
+        def counter = new AtomicInteger()
 
         expect:
-        def bac = instrument(new BlobClientBuilder()
-            .pipeline(bc.getHttpPipeline())
-            .endpoint(bc.getBlobUrl()))
+        def bacUploading = instrument(new BlobClientBuilder()
+            .endpoint(bc.getBlobUrl())
+            .credential(env.primaryAccount.credential))
+            .buildAsyncClient()
+            .getBlockBlobAsyncClient()
+
+        def bacDownloading = instrument(new BlobClientBuilder()
+            .addPolicy({ context, next ->
+                return next.process()
+                    .flatMap({ r ->
+                        if (counter.incrementAndGet() == 1) {
+                            /*
+                             * When the download begins trigger an upload to overwrite the downloading blob
+                             * so that the download is able to get an ETag before it is changed.
+                             */
+                            return bacUploading.upload(data.defaultFlux, data.defaultDataSize, true)
+                            .thenReturn(r)
+                        }
+                        return Mono.just(r)
+                    })
+            })
+            .endpoint(bc.getBlobUrl())
+            .credential(env.primaryAccount.credential))
             .buildAsyncClient()
             .getBlockBlobAsyncClient()
 
@@ -1165,12 +1187,7 @@ class BlobAPITest extends APISpec {
          */
         Hooks.onErrorDropped({ ignored -> /* do nothing with it */ })
 
-        /*
-         * When the download begins trigger an upload to overwrite the downloading blob after waiting 500 milliseconds
-         * so that the download is able to get an ETag before it is changed.
-         */
-        StepVerifier.create(bac.downloadToFileWithResponse(outFile.toPath().toString(), null, options, null, null, false)
-            .doOnSubscribe({ bac.upload(data.defaultFlux, data.defaultDataSize, true).delaySubscription(Duration.ofMillis(500)).subscribe() }))
+        StepVerifier.create(bacDownloading.downloadToFileWithResponse(outFile.toPath().toString(), null, options, null, null, false))
             .verifyErrorSatisfies({
                 /*
                  * If an operation is running on multiple threads and multiple return an exception Reactor will combine
diff --git a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/specialized/BlockBlobAPITest.groovy b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/specialized/BlockBlobAPITest.groovy
@@ -64,10 +64,10 @@ class BlockBlobAPITest extends APISpec {
         blobName = generateBlobName()
         blobClient = cc.getBlobClient(blobName)
         blockBlobClient = blobClient.getBlockBlobClient()
-        blockBlobClient.upload(data.defaultInputStream, data.defaultDataSize)
+        blockBlobClient.upload(data.defaultInputStream, data.defaultDataSize, true)
         blobAsyncClient = ccAsync.getBlobAsyncClient(generateBlobName())
         blockBlobAsyncClient = blobAsyncClient.getBlockBlobAsyncClient()
-        blockBlobAsyncClient.upload(data.defaultFlux, data.defaultDataSize).block()
+        blockBlobAsyncClient.upload(data.defaultFlux, data.defaultDataSize, true).block()
     }
 
     def "Stage block"() {
diff --git a/sdk/storage/azure-storage-file-datalake/src/test/java/com/azure/storage/file/datalake/FileAPITest.groovy b/sdk/storage/azure-storage-file-datalake/src/test/java/com/azure/storage/file/datalake/FileAPITest.groovy
@@ -11,12 +11,10 @@ import com.azure.storage.blob.models.BlobStorageException
 import com.azure.storage.common.ParallelTransferOptions
 import com.azure.storage.common.ProgressReceiver
 import com.azure.storage.common.implementation.Constants
-import com.azure.storage.common.test.shared.TestHttpClientType
 import com.azure.storage.common.test.shared.extensions.LiveOnly
 import com.azure.storage.common.test.shared.extensions.RequiredServiceVersion
 import com.azure.storage.common.test.shared.policy.MockFailureResponsePolicy
 import com.azure.storage.common.test.shared.policy.MockRetryRangeResponsePolicy
-import com.azure.storage.file.datalake.models.DownloadRetryOptions
 import com.azure.storage.file.datalake.models.AccessTier
 import com.azure.storage.file.datalake.models.DataLakeRequestConditions
 import com.azure.storage.file.datalake.models.DataLakeStorageException
@@ -46,10 +44,9 @@ import com.azure.storage.file.datalake.options.FileScheduleDeletionOptions
 import reactor.core.Exceptions
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Hooks
+import reactor.core.publisher.Mono
 import reactor.test.StepVerifier
-import spock.lang.Ignore
 import spock.lang.IgnoreIf
-import spock.lang.Requires
 import spock.lang.Retry
 import spock.lang.Unroll
 
@@ -63,6 +60,7 @@ import java.security.MessageDigest
 import java.time.Duration
 import java.time.OffsetDateTime
 import java.time.temporal.ChronoUnit
+import java.util.concurrent.atomic.AtomicInteger
 import java.util.function.Consumer
 
 class FileAPITest extends APISpec {
@@ -1502,18 +1500,38 @@ class FileAPITest extends APISpec {
     }
 
     @LiveOnly
-    @Ignore("failing in ci")
     def "Download file etag lock"() {
         setup:
         def file = getRandomFile(Constants.MB)
         fc.uploadFromFile(file.toPath().toString(), true)
         def outFile = new File(namer.getResourcePrefix())
         Files.deleteIfExists(file.toPath())
+        def counter = new AtomicInteger()
 
         expect:
-        def fac = new DataLakePathClientBuilder()
-            .pipeline(fc.getHttpPipeline())
+
+        def facUploading = instrument(new DataLakePathClientBuilder()
+            .endpoint(fc.getPathUrl())
+            .credential(env.dataLakeAccount.credential))
+            .buildFileAsyncClient()
+
+        def facDownloading = instrument(new DataLakePathClientBuilder()
+            .addPolicy({ context, next ->
+                return next.process()
+                    .flatMap({ r ->
+                        if (counter.incrementAndGet() == 1) {
+                            /*
+                             * When the download begins trigger an upload to overwrite the downloading blob
+                             * so that the download is able to get an ETag before it is changed.
+                             */
+                            return facUploading.upload(data.defaultFlux, null, true)
+                                .thenReturn(r)
+                        }
+                        return Mono.just(r)
+                    })
+            })
             .endpoint(fc.getPathUrl())
+            .credential(env.dataLakeAccount.credential))
             .buildFileAsyncClient()
 
         /*
@@ -1531,12 +1549,7 @@ class FileAPITest extends APISpec {
          */
         Hooks.onErrorDropped({ ignored -> /* do nothing with it */ })
 
-        /*
-         * When the download begins trigger an upload to overwrite the downloading blob after waiting 500 milliseconds
-         * so that the download is able to get an ETag before it is changed.
-         */
-        StepVerifier.create(fac.readToFileWithResponse(outFile.toPath().toString(), null, options, null, null, false, null)
-            .doOnSubscribe({ fac.upload(data.defaultFlux, null, true).delaySubscription(Duration.ofMillis(500)).subscribe() }))
+        StepVerifier.create(facDownloading.readToFileWithResponse(outFile.toPath().toString(), null, options, null, null, false, null))
             .verifyErrorSatisfies({
                 /*
                  * If an operation is running on multiple threads and multiple return an exception Reactor will combine
