diff --git a/asterixdb/asterix-app/pom.xml b/asterixdb/asterix-app/pom.xml
index e9d7412..3f75c32 100644
--- a/asterixdb/asterix-app/pom.xml
+++ b/asterixdb/asterix-app/pom.xml
@@ -998,6 +998,17 @@
       <groupId>software.amazon.awssdk</groupId>
       <artifactId>auth</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.adobe.testing</groupId>
+      <artifactId>s3mock</artifactId>
+      <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>org.springframework.boot</groupId>
+          <artifactId>spring-boot-starter-logging</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
     <!-- Mock for AWS S3 -->
     <dependency>
       <groupId>io.findify</groupId>
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/LocalCloudUtil.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/LocalCloudUtil.java
index b50d2f2..95fdafc 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/LocalCloudUtil.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/LocalCloudUtil.java
@@ -29,10 +29,13 @@
 
 import io.findify.s3mock.S3Mock;
 import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
+import software.amazon.awssdk.core.sync.RequestBody;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.S3ClientBuilder;
 import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 
 public class LocalCloudUtil {
 
@@ -41,6 +44,7 @@
     private static final int MOCK_SERVER_PORT = 8001;
     public static final String MOCK_SERVER_HOSTNAME = "http://127.0.0.1:" + MOCK_SERVER_PORT;
     public static final String CLOUD_STORAGE_BUCKET = "cloud-storage-container";
+    public static final String STORAGE_DUMMY_FILE = "storage/dummy.txt";
     public static final String MOCK_SERVER_REGION = "us-west-2";
     private static final String MOCK_FILE_BACKEND = joinPath("target", "s3mock");
     private static S3Mock s3MockServer;
@@ -84,6 +88,14 @@
         client.createBucket(CreateBucketRequest.builder().bucket(CLOUD_STORAGE_BUCKET).build());
         LOGGER.info("Created bucket {} for cloud storage", CLOUD_STORAGE_BUCKET);
 
+        // create a storage container and delete stuff inside it, just to create a directory.
+        PutObjectRequest putObjectRequest =
+                PutObjectRequest.builder().bucket(CLOUD_STORAGE_BUCKET).key(STORAGE_DUMMY_FILE).build();
+
+        client.putObject(putObjectRequest, RequestBody.empty());
+        // delete dummy file to retain storage directory.
+        client.deleteObject(DeleteObjectRequest.builder().bucket(CLOUD_STORAGE_BUCKET).key(STORAGE_DUMMY_FILE).build());
+
         // added for convenience since some non-external-based tests include an external collection test on this bucket
         if (createPlaygroundContainer) {
             client.createBucket(CreateBucketRequest.builder().bucket("playground").build());
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/LocalCloudUtilAdobeMock.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/LocalCloudUtilAdobeMock.java
new file mode 100644
index 0000000..5f6b492
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/LocalCloudUtilAdobeMock.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.api.common;
+
+import static org.apache.asterix.api.common.LocalCloudUtil.MOCK_SERVER_HOSTNAME;
+import static org.apache.asterix.api.common.LocalCloudUtil.MOCK_SERVER_REGION;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import com.adobe.testing.s3mock.S3MockApplication;
+
+import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3ClientBuilder;
+import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+
+public class LocalCloudUtilAdobeMock {
+
+    private static final Logger LOGGER = LogManager.getLogger();
+
+    private static final int MOCK_SERVER_PORT = 8001;
+    private static final int MOCK_SERVER_PORT_HTTPS = 8002;
+    public static final String CLOUD_STORAGE_BUCKET = "cloud-storage-container";
+    private static S3MockApplication s3Mock;
+
+    private LocalCloudUtilAdobeMock() {
+        throw new AssertionError("Do not instantiate");
+    }
+
+    public static void main(String[] args) {
+        String cleanStartString = System.getProperty("cleanup.start", "true");
+        boolean cleanStart = Boolean.parseBoolean(cleanStartString);
+        // Change to 'true' if you want to delete "s3mock" folder on start
+        startS3CloudEnvironment(cleanStart);
+    }
+
+    public static S3MockApplication startS3CloudEnvironment(boolean cleanStart) {
+        return startS3CloudEnvironment(cleanStart, false);
+    }
+
+    public static S3MockApplication startS3CloudEnvironment(boolean cleanStart, boolean createPlaygroundContainer) {
+        // Starting S3 mock server to be used instead of real S3 server
+        LOGGER.info("Starting S3 mock server");
+
+        Map<String, Object> properties = new HashMap<>();
+        properties.put(S3MockApplication.PROP_HTTP_PORT, MOCK_SERVER_PORT);
+        properties.put(S3MockApplication.PROP_HTTPS_PORT, MOCK_SERVER_PORT_HTTPS);
+        properties.put(S3MockApplication.PROP_SILENT, false);
+        shutdownSilently();
+        s3Mock = S3MockApplication.start(properties);
+
+        LOGGER.info("S3 mock server started successfully");
+
+        S3ClientBuilder builder = S3Client.builder();
+        URI endpoint = URI.create(MOCK_SERVER_HOSTNAME); // endpoint pointing to S3 mock server
+        builder.region(Region.of(MOCK_SERVER_REGION)).credentialsProvider(AnonymousCredentialsProvider.create())
+                .endpointOverride(endpoint);
+        S3Client client = builder.build();
+        client.createBucket(CreateBucketRequest.builder().bucket(CLOUD_STORAGE_BUCKET).build());
+        LOGGER.info("Created bucket {} for cloud storage", CLOUD_STORAGE_BUCKET);
+
+        if (createPlaygroundContainer) {
+            client.createBucket(CreateBucketRequest.builder().bucket("playground").build());
+            LOGGER.info("Created bucket {}", "playground");
+        }
+        client.close();
+        return s3Mock;
+    }
+
+    public static void shutdownSilently() {
+        if (s3Mock != null) {
+            try {
+                LOGGER.info("test cleanup, stopping S3 mock server");
+                s3Mock.stop();
+                LOGGER.info("test cleanup, stopped S3 mock server");
+            } catch (Exception ex) {
+                // do nothing
+            }
+        }
+    }
+}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/atomic_statements/AtomicMetadataTransactionWithoutWALTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/atomic_statements/AtomicMetadataTransactionWithoutWALTest.java
index 0d915de..b38d82b 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/atomic_statements/AtomicMetadataTransactionWithoutWALTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/atomic_statements/AtomicMetadataTransactionWithoutWALTest.java
@@ -26,7 +26,7 @@
 import java.util.Random;
 
 import org.apache.asterix.api.common.AsterixHyracksIntegrationUtil;
-import org.apache.asterix.api.common.LocalCloudUtil;
+import org.apache.asterix.api.common.LocalCloudUtilAdobeMock;
 import org.apache.asterix.common.TestDataUtil;
 import org.apache.asterix.common.utils.Servlets;
 import org.apache.asterix.test.common.TestExecutor;
@@ -55,7 +55,7 @@
     @Before
     public void setUp() throws Exception {
         boolean cleanStart = Boolean.getBoolean("cleanup.start");
-        LocalCloudUtil.startS3CloudEnvironment(cleanStart);
+        LocalCloudUtilAdobeMock.startS3CloudEnvironment(cleanStart);
         integrationUtil.setGracefulShutdown(false);
         integrationUtil.init(true, CONFIG_FILE);
     }
@@ -63,6 +63,7 @@
     @After
     public void tearDown() throws Exception {
         integrationUtil.deinit(true);
+        LocalCloudUtilAdobeMock.shutdownSilently();
     }
 
     private void createDatasets() throws Exception {
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/atomic_statements/AtomicStatementsCancellationTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/atomic_statements/AtomicStatementsCancellationTest.java
index 1cb87d8..02e9fd0 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/atomic_statements/AtomicStatementsCancellationTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/atomic_statements/AtomicStatementsCancellationTest.java
@@ -36,7 +36,7 @@
 import java.util.concurrent.Future;
 
 import org.apache.asterix.api.common.AsterixHyracksIntegrationUtil;
-import org.apache.asterix.api.common.LocalCloudUtil;
+import org.apache.asterix.api.common.LocalCloudUtilAdobeMock;
 import org.apache.asterix.common.TestDataUtil;
 import org.apache.asterix.common.utils.Servlets;
 import org.apache.asterix.test.common.TestExecutor;
@@ -67,7 +67,7 @@
     @Before
     public void setUp() throws Exception {
         boolean cleanStart = true;
-        LocalCloudUtil.startS3CloudEnvironment(cleanStart);
+        LocalCloudUtilAdobeMock.startS3CloudEnvironment(cleanStart);
         integrationUtil.setGracefulShutdown(true);
         integrationUtil.init(true, TEST_CONFIG_FILE_PATH);
         createDatasets();
@@ -76,6 +76,7 @@
     @After
     public void tearDown() throws Exception {
         integrationUtil.deinit(true);
+        LocalCloudUtilAdobeMock.shutdownSilently();
     }
 
     private void createDatasets() throws Exception {
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageAzTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageAzTest.java
index 508810d..2db2ef5 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageAzTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageAzTest.java
@@ -26,6 +26,7 @@
 import java.util.Objects;
 import java.util.Random;
 
+import org.apache.asterix.api.common.LocalCloudUtilAdobeMock;
 import org.apache.asterix.common.config.GlobalConfig;
 import org.apache.asterix.test.common.TestExecutor;
 import org.apache.asterix.test.runtime.LangExecutionUtil;
@@ -70,6 +71,7 @@
 
     @BeforeClass
     public static void setUp() throws Exception {
+        LocalCloudUtilAdobeMock.startS3CloudEnvironment(true, true);
         String endpointString = "http://127.0.0.1:15055/devstoreaccount1/" + CLOUD_STORAGE_BUCKET;
         final String accKey =
                 "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==";
@@ -93,6 +95,7 @@
     @AfterClass
     public static void tearDown() throws Exception {
         LangExecutionUtil.tearDown();
+        LocalCloudUtilAdobeMock.shutdownSilently();
     }
 
     @Parameters(name = "CloudStorageAzBlobTest {index}: {0}")
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageGCSTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageGCSTest.java
index 6ac4a5d..65b5adf 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageGCSTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageGCSTest.java
@@ -27,7 +27,7 @@
 import java.util.Objects;
 import java.util.Random;
 
-import org.apache.asterix.api.common.LocalCloudUtil;
+import org.apache.asterix.api.common.LocalCloudUtilAdobeMock;
 import org.apache.asterix.common.config.GlobalConfig;
 import org.apache.asterix.test.common.TestExecutor;
 import org.apache.asterix.test.runtime.LangExecutionUtil;
@@ -77,7 +77,7 @@
 
     @BeforeClass
     public static void setUp() throws Exception {
-        LocalCloudUtil.startS3CloudEnvironment(true, true);
+        LocalCloudUtilAdobeMock.startS3CloudEnvironment(true, true);
         Storage storage = StorageOptions.newBuilder().setHost(MOCK_SERVER_HOSTNAME)
                 .setCredentials(NoCredentials.getInstance()).setProjectId(MOCK_SERVER_PROJECT_ID).build().getService();
         cleanup(storage);
@@ -93,6 +93,7 @@
     @AfterClass
     public static void tearDown() throws Exception {
         LangExecutionUtil.tearDown();
+        LocalCloudUtilAdobeMock.shutdownSilently();
     }
 
     @Parameters(name = "CloudStorageGCSTest {index}: {0}")
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageTest.java
index 78f4e55..498f060 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageTest.java
@@ -22,7 +22,7 @@
 import java.util.Collection;
 import java.util.List;
 
-import org.apache.asterix.api.common.LocalCloudUtil;
+import org.apache.asterix.api.common.LocalCloudUtilAdobeMock;
 import org.apache.asterix.common.config.GlobalConfig;
 import org.apache.asterix.test.common.TestExecutor;
 import org.apache.asterix.test.runtime.LangExecutionUtil;
@@ -74,7 +74,7 @@
 
     @BeforeClass
     public static void setUp() throws Exception {
-        LocalCloudUtil.startS3CloudEnvironment(true);
+        LocalCloudUtilAdobeMock.startS3CloudEnvironment(true);
         TestExecutor testExecutor = new TestExecutor(DELTA_RESULT_PATH);
         testExecutor.executorId = "cloud";
         testExecutor.stripSubstring = "//DB:";
@@ -94,6 +94,7 @@
     @AfterClass
     public static void tearDown() throws Exception {
         LangExecutionUtil.tearDown();
+        LocalCloudUtilAdobeMock.shutdownSilently();
     }
 
     @Parameters(name = "CloudStorageTest {index}: {0}")
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudClient.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudClient.java
index b208714..fd82944 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudClient.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudClient.java
@@ -124,7 +124,7 @@
      * @param bucket bucket
      * @param paths  paths of all objects to be deleted
      */
-    void deleteObjects(String bucket, Collection<String> paths);
+    void deleteObjects(String bucket, Collection<String> paths) throws HyracksDataException;
 
     /**
      * Returns the size of the object at the specified path
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/UnstableCloudClient.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/UnstableCloudClient.java
index 4e1c0f7..28fa53e 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/UnstableCloudClient.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/UnstableCloudClient.java
@@ -98,7 +98,7 @@
     }
 
     @Override
-    public void deleteObjects(String bucket, Collection<String> paths) {
+    public void deleteObjects(String bucket, Collection<String> paths) throws HyracksDataException {
         cloudClient.deleteObjects(bucket, paths);
     }
 
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3CloudClient.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3CloudClient.java
index 319b713..d279643 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3CloudClient.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3CloudClient.java
@@ -46,11 +46,15 @@
 import org.apache.asterix.cloud.clients.profiler.CountRequestProfilerLimiter;
 import org.apache.asterix.cloud.clients.profiler.IRequestProfilerLimiter;
 import org.apache.asterix.cloud.clients.profiler.RequestLimiterNoOpProfiler;
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.common.exceptions.RuntimeDataException;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.io.FileReference;
 import org.apache.hyracks.api.util.IoUtil;
 import org.apache.hyracks.control.nc.io.IOManager;
 import org.apache.hyracks.util.annotations.ThreadSafe;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -65,16 +69,19 @@
 import software.amazon.awssdk.services.s3.model.CopyObjectRequest;
 import software.amazon.awssdk.services.s3.model.Delete;
 import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest;
+import software.amazon.awssdk.services.s3.model.DeleteObjectsResponse;
 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
 import software.amazon.awssdk.services.s3.model.GetObjectResponse;
 import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
 import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
 import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import software.amazon.awssdk.services.s3.model.S3Error;
 import software.amazon.awssdk.services.s3.model.S3Object;
 
 @ThreadSafe
 public final class S3CloudClient implements ICloudClient {
+    private static final Logger LOGGER = LogManager.getLogger();
     private final S3ClientConfig config;
     private final S3Client s3Client;
     private final ICloudGuardian guardian;
@@ -216,7 +223,7 @@
     }
 
     @Override
-    public void deleteObjects(String bucket, Collection<String> paths) {
+    public void deleteObjects(String bucket, Collection<String> paths) throws HyracksDataException {
         if (paths.isEmpty()) {
             return;
         }
@@ -234,7 +241,16 @@
 
             Delete delete = Delete.builder().objects(objectIdentifiers).build();
             DeleteObjectsRequest deleteReq = DeleteObjectsRequest.builder().bucket(bucket).delete(delete).build();
-            s3Client.deleteObjects(deleteReq);
+            DeleteObjectsResponse deleteObjectsResponse = s3Client.deleteObjects(deleteReq);
+            if (deleteObjectsResponse.hasErrors()) {
+                List<S3Error> deleteErrors = deleteObjectsResponse.errors();
+                for (S3Error s3Error : deleteErrors) {
+                    LOGGER.warn("Failed to delete object: {}, code: {}, message: {}", s3Error.key(), s3Error.code(),
+                            s3Error.message());
+                }
+                throw new RuntimeDataException(ErrorCode.CLOUD_IO_FAILURE, "DELETE", deleteErrors.get(0).key(),
+                        paths.toString());
+            }
             profiler.objectDelete();
         }
     }
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/azure/blobstorage/AzBlobStorageCloudClient.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/azure/blobstorage/AzBlobStorageCloudClient.java
index b9f9421..02da6ae 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/azure/blobstorage/AzBlobStorageCloudClient.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/azure/blobstorage/AzBlobStorageCloudClient.java
@@ -48,6 +48,8 @@
 import org.apache.asterix.cloud.clients.profiler.CountRequestProfilerLimiter;
 import org.apache.asterix.cloud.clients.profiler.IRequestProfilerLimiter;
 import org.apache.asterix.cloud.clients.profiler.RequestLimiterNoOpProfiler;
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.common.exceptions.RuntimeDataException;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.io.FileReference;
 import org.apache.hyracks.control.nc.io.IOManager;
@@ -55,6 +57,7 @@
 import org.apache.logging.log4j.Logger;
 
 import com.azure.core.http.rest.PagedIterable;
+import com.azure.core.http.rest.Response;
 import com.azure.core.util.BinaryData;
 import com.azure.storage.blob.BlobClient;
 import com.azure.storage.blob.BlobContainerClient;
@@ -246,7 +249,7 @@
     }
 
     @Override
-    public void deleteObjects(String bucket, Collection<String> paths) {
+    public void deleteObjects(String bucket, Collection<String> paths) throws HyracksDataException {
         if (paths.isEmpty())
             return;
         Set<BlobItem> blobsToDelete = getBlobsMatchingThesePaths(paths);
@@ -255,7 +258,17 @@
             return;
         Collection<List<String>> batchedBlobURLs = getBatchedBlobURLs(blobURLs);
         for (List<String> batch : batchedBlobURLs) {
-            blobBatchClient.deleteBlobs(batch, null).stream().count();
+            PagedIterable<Response<Void>> responses = blobBatchClient.deleteBlobs(batch, null);
+            Iterator<String> deletePathIter = paths.iterator();
+            String deletedPath = null;
+            try {
+                for (Response<Void> response : responses) {
+                    deletedPath = deletePathIter.next();
+                    response.getStatusCode();
+                }
+            } catch (BlobStorageException e) {
+                throw new RuntimeDataException(ErrorCode.CLOUD_IO_FAILURE, e, "DELETE", deletedPath, paths.toString());
+            }
         }
     }
 
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSCloudClient.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSCloudClient.java
index 62ca4ec..2ef34b0 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSCloudClient.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSCloudClient.java
@@ -41,17 +41,22 @@
 import org.apache.asterix.cloud.clients.profiler.CountRequestProfilerLimiter;
 import org.apache.asterix.cloud.clients.profiler.IRequestProfilerLimiter;
 import org.apache.asterix.cloud.clients.profiler.RequestLimiterNoOpProfiler;
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.common.exceptions.RuntimeDataException;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.io.FileReference;
 import org.apache.hyracks.api.util.CleanupUtils;
 import org.apache.hyracks.api.util.IoUtil;
 import org.apache.hyracks.control.nc.io.IOManager;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.api.gax.paging.Page;
+import com.google.cloud.BaseServiceException;
 import com.google.cloud.ReadChannel;
 import com.google.cloud.storage.Blob;
 import com.google.cloud.storage.BlobId;
@@ -60,10 +65,12 @@
 import com.google.cloud.storage.Storage.BlobListOption;
 import com.google.cloud.storage.Storage.CopyRequest;
 import com.google.cloud.storage.StorageBatch;
+import com.google.cloud.storage.StorageBatchResult;
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageOptions;
 
 public class GCSCloudClient implements ICloudClient {
+    private static final Logger LOGGER = LogManager.getLogger();
     private final Storage gcsClient;
     private final GCSClientConfig config;
     private final ICloudGuardian guardian;
@@ -193,11 +200,12 @@
     }
 
     @Override
-    public void deleteObjects(String bucket, Collection<String> paths) {
+    public void deleteObjects(String bucket, Collection<String> paths) throws HyracksDataException {
         if (paths.isEmpty()) {
             return;
         }
 
+        List<StorageBatchResult<Boolean>> deleteResponses = new ArrayList<>();
         StorageBatch batchRequest;
         Iterator<String> pathIter = paths.iterator();
         while (pathIter.hasNext()) {
@@ -205,10 +213,24 @@
             for (int i = 0; pathIter.hasNext() && i < DELETE_BATCH_SIZE; i++) {
                 BlobId blobId = BlobId.of(bucket, config.getPrefix() + pathIter.next());
                 guardian.checkWriteAccess(bucket, blobId.getName());
-                batchRequest.delete(blobId);
+                deleteResponses.add(batchRequest.delete(blobId));
             }
 
             batchRequest.submit();
+            Iterator<String> deletePathIter = paths.iterator();
+            for (StorageBatchResult<Boolean> deleteResponse : deleteResponses) {
+                String deletedPath = deletePathIter.next();
+                try {
+                    boolean deleted = deleteResponse.get();
+                    if (!deleted) {
+                        LOGGER.warn("File {} already deleted while deleting {}", deletedPath, paths);
+                    }
+                } catch (BaseServiceException e) {
+                    LOGGER.warn("Failed to delete object {} while deleting {}", deletedPath, paths, e);
+                    throw new RuntimeDataException(ErrorCode.CLOUD_IO_FAILURE, e, "DELETE", deletedPath,
+                            paths.toString());
+                }
+            }
             profilerLimiter.objectDelete();
         }
     }
@@ -287,4 +309,4 @@
     private String stripCloudPrefix(String objectName) {
         return objectName.substring(config.getPrefix().length());
     }
-}
\ No newline at end of file
+}
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
index e131f8a..afc43e2 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
@@ -99,6 +99,7 @@
     INVALID_KEY_TYPE(68),
     FAILED_TO_READ_KEY(69),
     AVRO_SUPPORTED_TYPE_WITH_OPTION(70),
+    CLOUD_IO_FAILURE(71),
     UNSUPPORTED_JRE(100),
 
     EXTERNAL_UDF_RESULT_TYPE_ERROR(200),
diff --git a/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties b/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
index 16dcda5..4a1629d 100644
--- a/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
+++ b/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
@@ -105,6 +105,7 @@
 68 = Invalid key type. Expected '%1$s', found '%2$s'.
 69 = Failed to read key. Reason: %1$s.
 70 = Avro type '%1$s' is not supported by default. To enable type conversion, recreate the external dataset with the option '%2$s' enabled
+71 = Cloud I/O operation '%1$s' failed during deletion of file '%2$s' in context of files '%3$s'
 
 100 = Unsupported JRE: %1$s
 
diff --git a/asterixdb/pom.xml b/asterixdb/pom.xml
index 16ec9ca..351c14e 100644
--- a/asterixdb/pom.xml
+++ b/asterixdb/pom.xml
@@ -1660,6 +1660,19 @@
         <artifactId>aws-crt</artifactId>
         <version>${awsjavasdk.crt.version}</version>
       </dependency>
+      <!-- Mock for Adobe AWS S3 -->
+      <dependency>
+        <groupId>com.adobe.testing</groupId>
+        <artifactId>s3mock</artifactId>
+        <version>2.17.0</version>
+        <scope>test</scope>
+        <exclusions>
+          <exclusion>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-logging</artifactId>
+          </exclusion>
+        </exclusions>
+      </dependency>
       <!-- Mock for AWS S3 -->
       <dependency>
         <groupId>io.findify</groupId>