Storage - Fix Flaky Stress Tests

sdk/storage/azure-storage-blob-stress/pom.xml

@@ -57,12 +57,12 @@
     <dependency>
       <groupId>io.opentelemetry.instrumentation</groupId>
       <artifactId>opentelemetry-runtime-telemetry-java8</artifactId>
-      <version>2.24.0-alpha</version> <!-- {x-version-update;io.opentelemetry.instrumentation:opentelemetry-runtime-telemetry-java8;external_dependency} -->
+      <version>2.15.0-alpha</version> <!-- {x-version-update;io.opentelemetry.instrumentation:opentelemetry-runtime-telemetry-java8;external_dependency} -->
     </dependency>
     <dependency>
       <groupId>io.opentelemetry.instrumentation</groupId>
       <artifactId>opentelemetry-logback-appender-1.0</artifactId>
-      <version>2.24.0-alpha</version> <!-- {x-version-update;io.opentelemetry.instrumentation:opentelemetry-logback-appender-1.0;external_dependency} -->
+      <version>2.15.0-alpha</version> <!-- {x-version-update;io.opentelemetry.instrumentation:opentelemetry-logback-appender-1.0;external_dependency} -->
     </dependency>
 
     <dependency>

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/AppendBlobOutputStream.java

@@ -70,7 +70,7 @@ public Mono<Void> setupAsync() {
 
     @Override
     public Mono<Void> cleanupAsync() {
-        return asyncNoFaultClient.delete()
+        return asyncNoFaultClient.deleteIfExists()
             .then(super.cleanupAsync());
     }
 }

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/AppendBlock.java

@@ -62,8 +62,9 @@ public Mono<Void> setupAsync() {
 
     @Override
     public Mono<Void> cleanupAsync() {
-        return asyncNoFaultClient.getAppendBlobAsyncClient().delete()
-            .then(tempSetupBlobClient.delete())
+        return asyncNoFaultClient.getAppendBlobAsyncClient().deleteIfExists()
+            .onErrorResume(e -> Mono.empty())
+            .then(tempSetupBlobClient.deleteIfExists())
             .then(super.cleanupAsync());
     }
 }

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/BlobScenarioBase.java

@@ -6,6 +6,7 @@
 import com.azure.core.http.policy.HttpLogDetailLevel;
 import com.azure.core.http.policy.HttpLogOptions;
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.identity.DefaultAzureCredential;
 import com.azure.identity.DefaultAzureCredentialBuilder;
 import com.azure.perf.test.core.PerfStressTest;
@@ -14,17 +15,21 @@
 import com.azure.storage.blob.BlobServiceAsyncClient;
 import com.azure.storage.blob.BlobServiceClient;
 import com.azure.storage.blob.BlobServiceClientBuilder;
+import com.azure.storage.stress.ContentMismatchException;
 import com.azure.storage.stress.TelemetryHelper;
 import com.azure.storage.stress.FaultInjectionProbabilities;
 import com.azure.storage.stress.FaultInjectingHttpPolicy;
 import com.azure.storage.stress.StorageStressOptions;
+import reactor.core.Exceptions;
 import reactor.core.publisher.Mono;
 
 import java.time.Instant;
 import java.util.UUID;
+import java.time.Duration;
 
 public abstract class BlobScenarioBase<TOptions extends StorageStressOptions> extends PerfStressTest<TOptions> {
     private static final String CONTAINER_NAME = "stress-" + UUID.randomUUID();
+    private static final ClientLogger LOGGER = new ClientLogger(BlobScenarioBase.class);
     protected final TelemetryHelper telemetryHelper = new TelemetryHelper(this.getClass());
     private final BlobContainerClient syncContainerClient;
     private final BlobContainerAsyncClient asyncContainerClient;
@@ -72,10 +77,76 @@ public Mono<Void> globalSetupAsync() {
     @Override
     public Mono<Void> globalCleanupAsync() {
         telemetryHelper.recordEnd(startTime);
-        return asyncNoFaultContainerClient.deleteIfExists()
+        return cleanupContainerWithRetry()
+            .onErrorResume(error -> {
+                // Log cleanup failure but don't fail the overall test
+                LOGGER.atWarning()
+                    .addKeyValue("error", error.getMessage())
+                    .log("Container cleanup failed");
+
+                return Mono.empty();
+            })
             .then(super.globalCleanupAsync());
     }
 
+    private static final int DELETE_TIMEOUT_SECONDS = 30;
+    private static final int BLOB_CLEANUP_TIMEOUT_SECONDS = 60;
+    private static final int MAX_RETRY_ATTEMPTS = 3;
+
+    private Mono<Void> cleanupContainerWithRetry() {
+        return tryDeleteContainer()
+            .onErrorResume(error -> fallbackCleanup());
+    }
+
+    private Mono<Void> tryDeleteContainer() {
+        return asyncNoFaultContainerClient.deleteIfExists()
+            .then()
+            .timeout(Duration.ofSeconds(DELETE_TIMEOUT_SECONDS))
+            .retry(MAX_RETRY_ATTEMPTS);
+    }
+
+    private Mono<Void> fallbackCleanup() {
+        return deleteAllBlobsInContainer()
+            .then(tryDeleteContainerOnce())
+            .onErrorResume(this::logCleanupFailure);
+    }
+
+    private Mono<Void> tryDeleteContainerOnce() {
+        return asyncNoFaultContainerClient.deleteIfExists()
+            .then()
+            .timeout(Duration.ofSeconds(DELETE_TIMEOUT_SECONDS));
+    }
+
+    private Mono<Void> logCleanupFailure(Throwable error) {
+        LOGGER.atWarning()
+            .addKeyValue("error", error.getMessage())
+            .log("Final container cleanup failed after retries");
+        return Mono.empty();
+    }
+
+    /**
+     * Deletes all blobs in the container sequentially to avoid throttling.
+     */
+    private Mono<Void> deleteAllBlobsInContainer() {
+        return asyncNoFaultContainerClient.listBlobs()
+            .concatMap(this::deleteBlobQuietly)
+            .then()
+            .timeout(Duration.ofSeconds(BLOB_CLEANUP_TIMEOUT_SECONDS))
+            .onErrorResume(error -> {
+                LOGGER.atWarning()
+                    .addKeyValue("error", error.getMessage())
+                    .log("Blob cleanup partially failed");
+                return Mono.empty();
+            });
+    }
+
+    private Mono<Void> deleteBlobQuietly(com.azure.storage.blob.models.BlobItem blobItem) {
+        return asyncNoFaultContainerClient.getBlobAsyncClient(blobItem.getName())
+            .deleteIfExists()
+            .onErrorResume(e -> Mono.empty())
+            .then();
+    }
+
     @SuppressWarnings("try")
     @Override
     public void run() {
@@ -85,8 +156,18 @@ public void run() {
     @SuppressWarnings("try")
     @Override
     public Mono<Void> runAsync() {
-        return telemetryHelper.instrumentRunAsync(ctx -> runInternalAsync(ctx))
-            .onErrorResume(e -> Mono.empty());
+        return telemetryHelper.instrumentRunAsync(ctx ->
+            runInternalAsync(ctx)
+                .retryWhen(reactor.util.retry.Retry.max(3)
+                    .filter(e -> !(Exceptions.unwrap(e)
+                        instanceof ContentMismatchException)))
+                .doOnError(e -> {
+                    // Log the error for debugging but let legitimate failures propagate
+                    LOGGER.atError()
+                        .addKeyValue("error", e.getMessage())
+                        .addKeyValue("errorType", e.getClass().getSimpleName())
+                        .log("Test operation failed after retries");
+                }));
     }
 
     protected abstract void runInternal(Context context) throws Exception;

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/BlockBlobOutputStream.java

@@ -70,7 +70,7 @@ public Mono<Void> setupAsync() {
 
     @Override
     public Mono<Void> cleanupAsync() {
-        return asyncNoFaultClient.delete()
+        return asyncNoFaultClient.deleteIfExists()
             .then(super.cleanupAsync());
     }
 

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/BlockBlobUpload.java

@@ -56,7 +56,7 @@ public Mono<Void> setupAsync() {
 
     @Override
     public Mono<Void> cleanupAsync() {
-        return asyncNoFaultClient.delete()
+        return asyncNoFaultClient.deleteIfExists()
             .then(super.cleanupAsync());
     }
 }

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/CommitBlockList.java

@@ -80,7 +80,7 @@ public Mono<Void> setupAsync() {
 
     @Override
     public Mono<Void> cleanupAsync() {
-        return asyncNoFaultClient.delete()
+        return asyncNoFaultClient.deleteIfExists()
             .then(super.cleanupAsync());
     }
 }

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/DownloadToFile.java

@@ -66,7 +66,7 @@ protected Mono<Void> runInternalAsync(Context span) {
 
     private static void deleteFile(Path path) {
         try {
-            path.toFile().delete();
+            Files.deleteIfExists(path);
         } catch (Throwable e) {
             LOGGER.atError()
                 .addKeyValue("path", path)

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/PageBlobOutputStream.java

@@ -93,8 +93,9 @@ public Mono<Void> setupAsync() {
 
     @Override
     public Mono<Void> cleanupAsync() {
-        return asyncNoFaultClient.getPageBlobAsyncClient().delete()
-            .then(tempSetupPageBlobClient.delete())
+        return asyncNoFaultClient.getPageBlobAsyncClient().deleteIfExists()
+            .onErrorResume(e -> Mono.empty())
+            .then(tempSetupPageBlobClient.deleteIfExists())
             .then(super.cleanupAsync());
     }
 }

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/PageBlobScenarioBase.java

@@ -6,6 +6,7 @@
 import com.azure.core.http.policy.HttpLogDetailLevel;
 import com.azure.core.http.policy.HttpLogOptions;
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.identity.DefaultAzureCredential;
 import com.azure.identity.DefaultAzureCredentialBuilder;
 import com.azure.perf.test.core.PerfStressTest;
@@ -14,18 +15,21 @@
 import com.azure.storage.blob.BlobServiceAsyncClient;
 import com.azure.storage.blob.BlobServiceClient;
 import com.azure.storage.blob.BlobServiceClientBuilder;
+import com.azure.storage.stress.ContentMismatchException;
 import com.azure.storage.stress.TelemetryHelper;
 import com.azure.storage.stress.FaultInjectingHttpPolicy;
 import com.azure.storage.stress.FaultInjectionProbabilities;
 import com.azure.storage.stress.StorageStressOptions;
+import reactor.core.Exceptions;
 import reactor.core.publisher.Mono;
 
+import java.time.Duration;
 import java.time.Instant;
-import java.util.Objects;
 import java.util.UUID;
 
 public abstract class PageBlobScenarioBase<TOptions extends StorageStressOptions> extends PerfStressTest<TOptions> {
     private static final String CONTAINER_NAME = "stress-" + UUID.randomUUID();
+    private static final ClientLogger LOGGER = new ClientLogger(PageBlobScenarioBase.class);
     protected final TelemetryHelper telemetryHelper = new TelemetryHelper(this.getClass());
     private final BlobServiceClient noFaultServiceClient;
     private final BlobContainerClient syncContainerClient;
@@ -72,10 +76,69 @@ public Mono<Void> globalSetupAsync() {
     @Override
     public Mono<Void> globalCleanupAsync() {
         telemetryHelper.recordEnd(startTime);
-        return asyncNoFaultContainerClient.deleteIfExists()
+        return cleanupContainerWithRetry()
+            .onErrorResume(error -> {
+                // Log cleanup failure but don't fail the overall test
+                LOGGER.atWarning()
+                    .addKeyValue("error", error.getMessage())
+                    .log("Container cleanup failed");
+
+                return Mono.empty();
+            })
             .then(super.globalCleanupAsync());
     }
 
+    /**
+     * Enhanced cleanup with timeout and retry logic to ensure containers are properly destroyed.
+     */
+    private Mono<Void> cleanupContainerWithRetry() {
+        return asyncNoFaultContainerClient.deleteIfExists()
+            .then()  // Convert Mono<Boolean> to Mono<Void>
+            .timeout(Duration.ofSeconds(30))
+            .retry(3)
+            .onErrorResume(error -> {
+                // If container deletion fails, try to delete all blobs first then retry container deletion
+                return deleteAllBlobsInContainer()
+                    .then(asyncNoFaultContainerClient.deleteIfExists())
+                    .then()  // Convert Mono<Boolean> to Mono<Void>
+                    .timeout(Duration.ofSeconds(30))
+                    .onErrorResume(finalError -> {
+                        // Log the error but don't fail the test
+                        LOGGER.atWarning()
+                            .addKeyValue("error", finalError.getMessage())
+                            .log("Final container cleanup failed after retries");
+                        return Mono.empty();
+                    });
+            });
+    }
+
+    /**
+     * Delete all blobs in the container to help with cleanup.
+     */
+    private Mono<Void> deleteAllBlobsInContainer() {
+        return asyncNoFaultContainerClient.listBlobs()
+            .concatMap(blobItem ->
+                asyncNoFaultContainerClient.getBlobAsyncClient(blobItem.getName())
+                    .deleteIfExists()
+                    .onErrorResume(error -> {
+                        // Log but continue - failures for individual blobs should not stop cleanup
+                        LOGGER.atWarning()
+                            .addKeyValue("blobName", blobItem.getName())
+                            .addKeyValue("error", error.getMessage())
+                            .log("Failed to delete blob during cleanup");
+                        return Mono.empty();
+                    }))
+            .then()
+            .timeout(Duration.ofSeconds(60))
+            .onErrorResume(error -> {
+                // Log but continue - some blobs might have been deleted
+                LOGGER.atWarning()
+                    .addKeyValue("error", error.getMessage())
+                    .log("Blob cleanup partially failed");
+                return Mono.empty();
+            });
+    }
+
     @SuppressWarnings("try")
     @Override
     public void run() {
@@ -85,8 +148,14 @@ public void run() {
     @SuppressWarnings("try")
     @Override
     public Mono<Void> runAsync() {
-        return telemetryHelper.instrumentRunAsync(ctx -> runInternalAsync(ctx))
-            .onErrorResume(e -> Mono.empty());
+        return telemetryHelper.instrumentRunAsync(ctx ->
+        runInternalAsync(ctx)
+            .retryWhen(reactor.util.retry.Retry.max(3)
+                .filter(e -> !(Exceptions.unwrap(e)
+                    instanceof ContentMismatchException))))
+            .doOnError(e -> LOGGER.atError()
+                .addKeyValue("error", e.getMessage())
+                .log("Test operation failed after retries"));
     }
 
     protected abstract void runInternal(Context context) throws Exception;

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/StageBlock.java

@@ -79,7 +79,7 @@ public Mono<Void> setupAsync() {
 
     @Override
     public Mono<Void> cleanupAsync() {
-        return asyncNoFaultClient.delete()
+        return asyncNoFaultClient.deleteIfExists()
             .then(super.cleanupAsync());
     }
 

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/Upload.java

@@ -61,7 +61,7 @@ public Mono<Void> setupAsync() {
 
     @Override
     public Mono<Void> cleanupAsync() {
-        return asyncNoFaultClient.delete()
+        return asyncNoFaultClient.deleteIfExists()
             .then(super.cleanupAsync());
     }
 }

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/UploadFromFile.java

@@ -107,7 +107,7 @@ private Path getTempPath(String prefix) {
 
     private static void deleteFile(Path path) {
         try {
-            path.toFile().delete();
+            Files.deleteIfExists(path);
         } catch (Throwable e) {
             LOGGER.atError()
                 .addKeyValue("path", path)

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/UploadPages.java

@@ -69,8 +69,9 @@ public Mono<Void> setupAsync() {
 
     @Override
     public Mono<Void> cleanupAsync() {
-        return asyncNoFaultClient.getPageBlobAsyncClient().delete()
-            .then(tempSetupPageBlobClient.delete())
+        return asyncNoFaultClient.getPageBlobAsyncClient().deleteIfExists()
+            .onErrorResume(e -> Mono.empty())
+            .then(tempSetupPageBlobClient.deleteIfExists())
             .then(super.cleanupAsync());
     }
 }

sdk/storage/azure-storage-file-datalake-stress/pom.xml

@@ -57,12 +57,12 @@
     <dependency>
       <groupId>io.opentelemetry.instrumentation</groupId>
       <artifactId>opentelemetry-runtime-telemetry-java8</artifactId>
-      <version>2.24.0-alpha</version> <!-- {x-version-update;io.opentelemetry.instrumentation:opentelemetry-runtime-telemetry-java8;external_dependency} -->
+      <version>2.15.0-alpha</version> <!-- {x-version-update;io.opentelemetry.instrumentation:opentelemetry-runtime-telemetry-java8;external_dependency} -->
     </dependency>
     <dependency>
       <groupId>io.opentelemetry.instrumentation</groupId>
       <artifactId>opentelemetry-logback-appender-1.0</artifactId>
-      <version>2.24.0-alpha</version> <!-- {x-version-update;io.opentelemetry.instrumentation:opentelemetry-logback-appender-1.0;external_dependency} -->
+      <version>2.15.0-alpha</version> <!-- {x-version-update;io.opentelemetry.instrumentation:opentelemetry-logback-appender-1.0;external_dependency} -->
     </dependency>
 
     <dependency>