diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
index 1d6a0d8..18e24ab 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
@@ -35,6 +35,7 @@
 import org.apache.asterix.app.result.ResultReader;
 import org.apache.asterix.cloud.CloudConfigurator;
 import org.apache.asterix.cloud.LocalPartitionBootstrapper;
+import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.common.api.IConfigValidator;
 import org.apache.asterix.common.api.IConfigValidatorFactory;
 import org.apache.asterix.common.api.ICoordinationService;
@@ -221,7 +222,8 @@
         IDiskCachedPageAllocator pageAllocator;
         IBufferCacheReadContext defaultContext;
         if (isCloudDeployment()) {
-            cloudConfigurator = CloudConfigurator.of(cloudProperties, ioManager, namespacePathResolver);
+            cloudConfigurator = CloudConfigurator.of(cloudProperties, ioManager, namespacePathResolver,
+                    getCloudGuardian(cloudProperties));
             persistenceIOManager = cloudConfigurator.getCloudIoManager();
             partitionBootstrapper = cloudConfigurator.getPartitionBootstrapper();
             lockNotifier = cloudConfigurator.getLockNotifier();
@@ -361,6 +363,10 @@
         diskWriteRateLimiterProvider = new DiskWriteRateLimiterProvider();
     }
 
+    protected ICloudGuardian getCloudGuardian(CloudProperties cloudProperties) {
+        return ICloudGuardian.NoOpCloudGuardian.INSTANCE;
+    }
+
     @Override
     public boolean isShuttingdown() {
         return isShuttingdown;
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
index cb8b8e5..b8cd51a 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
@@ -62,6 +62,7 @@
 import org.apache.asterix.app.replication.NcLifecycleCoordinator;
 import org.apache.asterix.app.result.JobResultCallback;
 import org.apache.asterix.cloud.CloudConfigurator;
+import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.common.api.AsterixThreadFactory;
 import org.apache.asterix.common.api.IConfigValidatorFactory;
 import org.apache.asterix.common.api.INamespacePathResolver;
@@ -186,7 +187,8 @@
         CloudProperties cloudProperties = null;
         if (cloudDeployment) {
             cloudProperties = new CloudProperties(PropertiesAccessor.getInstance(ccServiceCtx.getAppConfig()));
-            ioManager = CloudConfigurator.createIOManager(ioManager, cloudProperties, namespacePathResolver);
+            ioManager = CloudConfigurator.createIOManager(ioManager, cloudProperties, namespacePathResolver,
+                    getCloudGuardian(cloudProperties));
         }
         IGlobalTxManager globalTxManager = createGlobalTxManager(ioManager);
         appCtx = createApplicationContext(null, globalRecoveryManager, lifecycleCoordinator, Receptionist::new,
@@ -216,6 +218,10 @@
         jobCapacityController = new JobCapacityController(controllerService.getResourceManager());
     }
 
+    protected ICloudGuardian getCloudGuardian(CloudProperties cloudProperties) {
+        return ICloudGuardian.NoOpCloudGuardian.INSTANCE;
+    }
+
     private Map<String, String> parseCredentialMap(String credPath) {
         File credentialFile = new File(credPath);
         Map<String, String> storedCredentials = new HashMap<>();
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/AbstractCloudIOManager.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/AbstractCloudIOManager.java
index 9df26f2..2bafade 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/AbstractCloudIOManager.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/AbstractCloudIOManager.java
@@ -36,6 +36,7 @@
 import org.apache.asterix.cloud.clients.CloudClientProvider;
 import org.apache.asterix.cloud.clients.CloudFile;
 import org.apache.asterix.cloud.clients.ICloudClient;
+import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.cloud.clients.ICloudWriter;
 import org.apache.asterix.cloud.util.CloudFileUtil;
 import org.apache.asterix.common.api.INamespacePathResolver;
@@ -60,6 +61,7 @@
 public abstract class AbstractCloudIOManager extends IOManager implements IPartitionBootstrapper, ICloudIOManager {
     private static final Logger LOGGER = LogManager.getLogger();
     protected final ICloudClient cloudClient;
+    protected final ICloudGuardian guardian;
     protected final IWriteBufferProvider writeBufferProvider;
     protected final String bucket;
     protected final Set<Integer> partitions;
@@ -68,12 +70,13 @@
     protected final INamespacePathResolver nsPathResolver;
 
     public AbstractCloudIOManager(IOManager ioManager, CloudProperties cloudProperties,
-            INamespacePathResolver nsPathResolver) throws HyracksDataException {
+            INamespacePathResolver nsPathResolver, ICloudGuardian guardian) throws HyracksDataException {
         super(ioManager.getIODevices(), ioManager.getDeviceComputer(), ioManager.getIOParallelism(),
                 ioManager.getQueueSize());
         this.nsPathResolver = nsPathResolver;
         this.bucket = cloudProperties.getStorageBucket();
-        cloudClient = CloudClientProvider.getClient(cloudProperties);
+        cloudClient = CloudClientProvider.getClient(cloudProperties, guardian);
+        this.guardian = guardian;
         int numOfThreads = getIODevices().size() * getIOParallelism();
         writeBufferProvider = new WriteBufferProvider(numOfThreads, cloudClient.getWriteBufferSize());
         partitions = new HashSet<>();
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/CloudConfigurator.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/CloudConfigurator.java
index a0c1fbb..0f9a4de 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/CloudConfigurator.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/CloudConfigurator.java
@@ -22,6 +22,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
+import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.common.api.INamespacePathResolver;
 import org.apache.asterix.common.cloud.CloudCachePolicy;
 import org.apache.asterix.common.cloud.IPartitionBootstrapper;
@@ -65,11 +66,11 @@
     private final long diskCacheMonitoringInterval;
 
     private CloudConfigurator(CloudProperties cloudProperties, IIOManager ioManager,
-            INamespacePathResolver nsPathResolver) throws HyracksDataException {
+            INamespacePathResolver nsPathResolver, ICloudGuardian guardian) throws HyracksDataException {
         this.cloudProperties = cloudProperties;
         localIoManager = (IOManager) ioManager;
         diskCacheManagerRequired = cloudProperties.getCloudCachePolicy() == CloudCachePolicy.SELECTIVE;
-        cloudIOManager = createIOManager(ioManager, cloudProperties, nsPathResolver);
+        cloudIOManager = createIOManager(ioManager, cloudProperties, nsPathResolver, guardian);
         physicalDrive = createPhysicalDrive(diskCacheManagerRequired, cloudProperties, ioManager);
         lockNotifier = createLockNotifier(diskCacheManagerRequired);
         pageAllocator = createPageAllocator(diskCacheManagerRequired);
@@ -126,20 +127,20 @@
     }
 
     public static CloudConfigurator of(CloudProperties cloudProperties, IIOManager ioManager,
-            INamespacePathResolver nsPathResolver) throws HyracksDataException {
-        return new CloudConfigurator(cloudProperties, ioManager, nsPathResolver);
+            INamespacePathResolver nsPathResolver, ICloudGuardian cloudGuardian) throws HyracksDataException {
+        return new CloudConfigurator(cloudProperties, ioManager, nsPathResolver, cloudGuardian);
     }
 
     public static AbstractCloudIOManager createIOManager(IIOManager ioManager, CloudProperties cloudProperties,
-            INamespacePathResolver nsPathResolver) throws HyracksDataException {
+            INamespacePathResolver nsPathResolver, ICloudGuardian guardian) throws HyracksDataException {
         IOManager localIoManager = (IOManager) ioManager;
         CloudCachePolicy policy = cloudProperties.getCloudCachePolicy();
         if (policy == CloudCachePolicy.EAGER) {
-            return new EagerCloudIOManager(localIoManager, cloudProperties, nsPathResolver);
+            return new EagerCloudIOManager(localIoManager, cloudProperties, nsPathResolver, guardian);
         }
 
         boolean selective = policy == CloudCachePolicy.SELECTIVE;
-        return new LazyCloudIOManager(localIoManager, cloudProperties, nsPathResolver, selective);
+        return new LazyCloudIOManager(localIoManager, cloudProperties, nsPathResolver, selective, guardian);
     }
 
     private static IPhysicalDrive createPhysicalDrive(boolean diskCacheManagerRequired, CloudProperties cloudProperties,
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/CloudManagerProvider.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/CloudManagerProvider.java
index fb61d7a..b49f3ce 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/CloudManagerProvider.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/CloudManagerProvider.java
@@ -18,6 +18,7 @@
  */
 package org.apache.asterix.cloud;
 
+import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.common.api.INamespacePathResolver;
 import org.apache.asterix.common.cloud.CloudCachePolicy;
 import org.apache.asterix.common.cloud.IPartitionBootstrapper;
@@ -31,13 +32,13 @@
     }
 
     public static IIOManager createIOManager(CloudProperties cloudProperties, IIOManager ioManager,
-            INamespacePathResolver nsPathResolver) throws HyracksDataException {
+            INamespacePathResolver nsPathResolver, ICloudGuardian guardian) throws HyracksDataException {
         IOManager localIoManager = (IOManager) ioManager;
         if (cloudProperties.getCloudCachePolicy() == CloudCachePolicy.LAZY) {
-            return new LazyCloudIOManager(localIoManager, cloudProperties, nsPathResolver, false);
+            return new LazyCloudIOManager(localIoManager, cloudProperties, nsPathResolver, false, guardian);
         }
 
-        return new EagerCloudIOManager(localIoManager, cloudProperties, nsPathResolver);
+        return new EagerCloudIOManager(localIoManager, cloudProperties, nsPathResolver, guardian);
     }
 
     public static IPartitionBootstrapper getCloudPartitionBootstrapper(IIOManager ioManager) {
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/EagerCloudIOManager.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/EagerCloudIOManager.java
index c993cd3..764d436 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/EagerCloudIOManager.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/EagerCloudIOManager.java
@@ -25,6 +25,7 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.cloud.clients.IParallelDownloader;
 import org.apache.asterix.common.api.INamespacePathResolver;
 import org.apache.asterix.common.config.CloudProperties;
@@ -47,8 +48,8 @@
     private static final Logger LOGGER = LogManager.getLogger();
 
     public EagerCloudIOManager(IOManager ioManager, CloudProperties cloudProperties,
-            INamespacePathResolver nsPathResolver) throws HyracksDataException {
-        super(ioManager, cloudProperties, nsPathResolver);
+            INamespacePathResolver nsPathResolver, ICloudGuardian guardian) throws HyracksDataException {
+        super(ioManager, cloudProperties, nsPathResolver, guardian);
     }
 
     /*
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/LazyCloudIOManager.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/LazyCloudIOManager.java
index afe0878..72f3446 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/LazyCloudIOManager.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/LazyCloudIOManager.java
@@ -33,6 +33,7 @@
 
 import org.apache.asterix.cloud.bulk.DeleteBulkCloudOperation;
 import org.apache.asterix.cloud.clients.CloudFile;
+import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.cloud.clients.IParallelDownloader;
 import org.apache.asterix.cloud.lazy.ParallelCacher;
 import org.apache.asterix.cloud.lazy.accessor.ILazyAccessor;
@@ -68,8 +69,9 @@
     private ILazyAccessor accessor;
 
     public LazyCloudIOManager(IOManager ioManager, CloudProperties cloudProperties,
-            INamespacePathResolver nsPathResolver, boolean selective) throws HyracksDataException {
-        super(ioManager, cloudProperties, nsPathResolver);
+            INamespacePathResolver nsPathResolver, boolean selective, ICloudGuardian guardian)
+            throws HyracksDataException {
+        super(ioManager, cloudProperties, nsPathResolver, guardian);
         accessor = new InitialCloudAccessor(cloudClient, bucket, localIoManager);
         puncher = HolePuncherProvider.get(this, cloudProperties, writeBufferProvider);
         if (selective) {
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/CloudClientProvider.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/CloudClientProvider.java
index d5f508d..b0a1e0c 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/CloudClientProvider.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/CloudClientProvider.java
@@ -26,21 +26,22 @@
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 
 public class CloudClientProvider {
-    private static final String S3 = "s3";
-    private static final String GCS = "gs";
+    public static final String S3 = "s3";
+    public static final String GCS = "gs";
 
     private CloudClientProvider() {
         throw new AssertionError("do not instantiate");
     }
 
-    public static ICloudClient getClient(CloudProperties cloudProperties) throws HyracksDataException {
+    public static ICloudClient getClient(CloudProperties cloudProperties, ICloudGuardian guardian)
+            throws HyracksDataException {
         String storageScheme = cloudProperties.getStorageScheme();
         if (S3.equalsIgnoreCase(storageScheme)) {
             S3ClientConfig config = S3ClientConfig.of(cloudProperties);
-            return new S3CloudClient(config);
+            return new S3CloudClient(config, guardian);
         } else if (GCS.equalsIgnoreCase(storageScheme)) {
             GCSClientConfig config = GCSClientConfig.of(cloudProperties);
-            return new GCSCloudClient(config);
+            return new GCSCloudClient(config, guardian);
         }
         throw new IllegalStateException("unsupported cloud storage scheme: " + storageScheme);
     }
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudGuardian.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudGuardian.java
new file mode 100644
index 0000000..2125cdd
--- /dev/null
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudGuardian.java
@@ -0,0 +1,72 @@
+/*
+ * 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.cloud.clients;
+
+/**
+ * Interface containing methods to perform IO operation on the Cloud Storage
+ */
+public interface ICloudGuardian {
+    /**
+     * Ensure we have authorization to perform isolated writes on the bucket. Isolated means that the writes do not
+     * interfere with any other node, at least at present. In the event that the isolated writes can become interfering
+     * in the future (i.e. committed), the {@link #checkWriteAccess(String, String)} method should be invoked just
+     * prior.
+     */
+    void checkIsolatedWriteAccess(String bucket, String path);
+
+    /**
+     * Ensure we have authorization to perform writes on the bucket. These writes are not isolated in that they are
+     * visibly side-effecting immediately.
+     */
+    void checkWriteAccess(String bucket, String path);
+
+    /**
+     * Ensure we have authorization to perform reads on the bucket.
+     */
+    void checkReadAccess(String bucket, String path);
+
+    void setCloudClient(ICloudClient cloudClient);
+
+    class NoOpCloudGuardian implements ICloudGuardian {
+
+        public static final NoOpCloudGuardian INSTANCE = new NoOpCloudGuardian();
+
+        private NoOpCloudGuardian() {
+        }
+
+        @Override
+        public void checkIsolatedWriteAccess(String bucket, String path) {
+            // no-op
+        }
+
+        @Override
+        public void checkWriteAccess(String bucket, String path) {
+            // no-op
+        }
+
+        @Override
+        public void checkReadAccess(String bucket, String path) {
+            // no-op
+        }
+
+        @Override
+        public void setCloudClient(ICloudClient cloudClient) {
+        }
+    }
+}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3BufferedWriter.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3BufferedWriter.java
index a6579c2..93be80c 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3BufferedWriter.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3BufferedWriter.java
@@ -24,6 +24,7 @@
 import java.util.concurrent.TimeUnit;
 
 import org.apache.asterix.cloud.clients.ICloudBufferedWriter;
+import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.cloud.clients.profiler.IRequestProfiler;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.logging.log4j.LogManager;
@@ -45,6 +46,7 @@
     private static final Logger LOGGER = LogManager.getLogger();
     private final S3Client s3Client;
     private final IRequestProfiler profiler;
+    private final ICloudGuardian guardian;
     private final String bucket;
     private final String path;
     private final List<CompletedPart> partQueue;
@@ -52,9 +54,11 @@
     private String uploadId;
     private int partNumber;
 
-    public S3BufferedWriter(S3Client s3client, IRequestProfiler profiler, String bucket, String path) {
+    public S3BufferedWriter(S3Client s3client, IRequestProfiler profiler, ICloudGuardian guardian, String bucket,
+            String path) {
         this.s3Client = s3client;
         this.profiler = profiler;
+        this.guardian = guardian;
         this.bucket = bucket;
         this.path = path;
         partQueue = new ArrayList<>();
@@ -62,6 +66,7 @@
 
     @Override
     public int upload(InputStream stream, int length) {
+        guardian.checkIsolatedWriteAccess(bucket, path);
         profiler.objectMultipartUpload();
         setUploadId();
         UploadPartRequest upReq =
@@ -123,6 +128,7 @@
     }
 
     private void completeMultipartUpload(CompleteMultipartUploadRequest request) throws HyracksDataException {
+        guardian.checkWriteAccess(bucket, path);
         profiler.objectMultipartUpload();
         try {
             s3Client.completeMultipartUpload(request);
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 38a6801..f395362 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
@@ -40,6 +40,7 @@
 import org.apache.asterix.cloud.clients.CloudFile;
 import org.apache.asterix.cloud.clients.ICloudBufferedWriter;
 import org.apache.asterix.cloud.clients.ICloudClient;
+import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.cloud.clients.ICloudWriter;
 import org.apache.asterix.cloud.clients.IParallelDownloader;
 import org.apache.asterix.cloud.clients.profiler.CountRequestProfiler;
@@ -76,21 +77,24 @@
 public final class S3CloudClient implements ICloudClient {
     private final S3ClientConfig config;
     private final S3Client s3Client;
+    private final ICloudGuardian guardian;
     private final IRequestProfiler profiler;
 
-    public S3CloudClient(S3ClientConfig config) {
-        this(config, buildClient(config));
+    public S3CloudClient(S3ClientConfig config, ICloudGuardian guardian) {
+        this(config, buildClient(config), guardian);
     }
 
-    public S3CloudClient(S3ClientConfig config, S3Client s3Client) {
+    public S3CloudClient(S3ClientConfig config, S3Client s3Client, ICloudGuardian guardian) {
         this.config = config;
         this.s3Client = s3Client;
+        this.guardian = guardian;
         long profilerInterval = config.getProfilerLogInterval();
         if (profilerInterval > 0) {
             profiler = new CountRequestProfiler(profilerInterval);
         } else {
             profiler = NoOpRequestProfiler.INSTANCE;
         }
+        guardian.setCloudClient(this);
     }
 
     @Override
@@ -100,12 +104,13 @@
 
     @Override
     public ICloudWriter createWriter(String bucket, String path, IWriteBufferProvider bufferProvider) {
-        ICloudBufferedWriter bufferedWriter = new S3BufferedWriter(s3Client, profiler, bucket, path);
+        ICloudBufferedWriter bufferedWriter = new S3BufferedWriter(s3Client, profiler, guardian, bucket, path);
         return new CloudResettableInputStream(bufferedWriter, bufferProvider);
     }
 
     @Override
     public Set<CloudFile> listObjects(String bucket, String path, FilenameFilter filter) {
+        guardian.checkReadAccess(bucket, path);
         profiler.objectsList();
         path = config.isLocalS3Provider() ? encodeURI(path) : path;
         return filterAndGet(listS3Objects(s3Client, bucket, path), filter);
@@ -113,6 +118,7 @@
 
     @Override
     public int read(String bucket, String path, long offset, ByteBuffer buffer) throws HyracksDataException {
+        guardian.checkReadAccess(bucket, path);
         profiler.objectGet();
         long readTo = offset + buffer.remaining();
         GetObjectRequest rangeGetObjectRequest =
@@ -140,6 +146,7 @@
 
     @Override
     public byte[] readAllBytes(String bucket, String path) throws HyracksDataException {
+        guardian.checkReadAccess(bucket, path);
         profiler.objectGet();
         GetObjectRequest getReq = GetObjectRequest.builder().bucket(bucket).key(path).build();
 
@@ -154,6 +161,7 @@
 
     @Override
     public InputStream getObjectStream(String bucket, String path, long offset, long length) {
+        guardian.checkReadAccess(bucket, path);
         profiler.objectGet();
         long readTo = offset + length;
         GetObjectRequest getReq =
@@ -168,6 +176,7 @@
 
     @Override
     public void write(String bucket, String path, byte[] data) {
+        guardian.checkWriteAccess(bucket, path);
         profiler.objectWrite();
         PutObjectRequest putReq = PutObjectRequest.builder().bucket(bucket).key(path).build();
         s3Client.putObject(putReq, RequestBody.fromBytes(data));
@@ -175,11 +184,13 @@
 
     @Override
     public void copy(String bucket, String srcPath, FileReference destPath) {
+        guardian.checkReadAccess(bucket, srcPath);
         srcPath = config.isLocalS3Provider() ? encodeURI(srcPath) : srcPath;
         List<S3Object> objects = listS3Objects(s3Client, bucket, srcPath);
 
         profiler.objectsList();
         for (S3Object object : objects) {
+            guardian.checkWriteAccess(bucket, destPath.getRelativePath());
             profiler.objectCopy();
             String srcKey = object.key();
             String destKey = destPath.getChildPath(IoUtil.getFileNameFromPath(srcKey));
@@ -201,7 +212,9 @@
         while (pathIter.hasNext()) {
             objectIdentifiers.clear();
             for (int i = 0; pathIter.hasNext() && i < DELETE_BATCH_SIZE; i++) {
-                objectIdentifiers.add(builder.key(pathIter.next()).build());
+                String path = pathIter.next();
+                guardian.checkWriteAccess(bucket, path);
+                objectIdentifiers.add(builder.key(path).build());
             }
 
             Delete delete = Delete.builder().objects(objectIdentifiers).build();
@@ -213,6 +226,7 @@
 
     @Override
     public long getObjectSize(String bucket, String path) throws HyracksDataException {
+        guardian.checkReadAccess(bucket, path);
         profiler.objectGet();
         try {
             return s3Client.headObject(HeadObjectRequest.builder().bucket(bucket).key(path).build()).contentLength();
@@ -225,6 +239,7 @@
 
     @Override
     public boolean exists(String bucket, String path) throws HyracksDataException {
+        guardian.checkReadAccess(bucket, path);
         profiler.objectGet();
         try {
             s3Client.headObject(HeadObjectRequest.builder().bucket(bucket).key(path).build());
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 9bcd386..accf3c9 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
@@ -35,6 +35,7 @@
 import org.apache.asterix.cloud.IWriteBufferProvider;
 import org.apache.asterix.cloud.clients.CloudFile;
 import org.apache.asterix.cloud.clients.ICloudClient;
+import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.cloud.clients.ICloudWriter;
 import org.apache.asterix.cloud.clients.IParallelDownloader;
 import org.apache.asterix.cloud.clients.profiler.CountRequestProfiler;
@@ -64,21 +65,24 @@
 public class GCSCloudClient implements ICloudClient {
     private final Storage gcsClient;
     private final GCSClientConfig config;
+    private final ICloudGuardian guardian;
     private final IRequestProfiler profiler;
 
-    public GCSCloudClient(GCSClientConfig config, Storage gcsClient) {
+    public GCSCloudClient(GCSClientConfig config, Storage gcsClient, ICloudGuardian guardian) {
         this.gcsClient = gcsClient;
         this.config = config;
+        this.guardian = guardian;
         long profilerInterval = config.getProfilerLogInterval();
         if (profilerInterval > 0) {
             profiler = new CountRequestProfiler(profilerInterval);
         } else {
             profiler = NoOpRequestProfiler.INSTANCE;
         }
+        guardian.setCloudClient(this);
     }
 
-    public GCSCloudClient(GCSClientConfig config) throws HyracksDataException {
-        this(config, buildClient(config));
+    public GCSCloudClient(GCSClientConfig config, ICloudGuardian guardian) throws HyracksDataException {
+        this(config, buildClient(config), guardian);
     }
 
     @Override
@@ -93,6 +97,7 @@
 
     @Override
     public Set<CloudFile> listObjects(String bucket, String path, FilenameFilter filter) {
+        guardian.checkReadAccess(bucket, path);
         profiler.objectsList();
         Page<Blob> blobs =
                 gcsClient.list(bucket, BlobListOption.prefix(path), BlobListOption.fields(Storage.BlobField.SIZE));
@@ -151,6 +156,7 @@
 
     @Override
     public void write(String bucket, String path, byte[] data) {
+        guardian.checkWriteAccess(bucket, path);
         profiler.objectWrite();
         BlobInfo blobInfo = BlobInfo.newBuilder(bucket, path).build();
         gcsClient.create(blobInfo, data);
@@ -165,6 +171,7 @@
             BlobId source = blob.getBlobId();
             String targetName = destPath.getChildPath(IoUtil.getFileNameFromPath(source.getName()));
             BlobId target = BlobId.of(bucket, targetName);
+            guardian.checkWriteAccess(bucket, targetName);
             CopyRequest copyReq = CopyRequest.newBuilder().setSource(source).setTarget(target).build();
             gcsClient.copy(copyReq);
         }
@@ -182,6 +189,7 @@
             batchRequest = gcsClient.batch();
             for (int i = 0; pathIter.hasNext() && i < DELETE_BATCH_SIZE; i++) {
                 BlobId blobId = BlobId.of(bucket, pathIter.next());
+                guardian.checkWriteAccess(bucket, blobId.getName());
                 batchRequest.delete(blobId);
             }
 
@@ -192,6 +200,7 @@
 
     @Override
     public long getObjectSize(String bucket, String path) {
+        guardian.checkReadAccess(bucket, path);
         profiler.objectGet();
         Blob blob = gcsClient.get(bucket, path, Storage.BlobGetOption.fields(Storage.BlobField.SIZE));
         if (blob == null) {
@@ -202,6 +211,7 @@
 
     @Override
     public boolean exists(String bucket, String path) {
+        guardian.checkReadAccess(bucket, path);
         profiler.objectGet();
         Blob blob = gcsClient.get(bucket, path, Storage.BlobGetOption.fields(Storage.BlobField.values()));
         return blob != null && blob.exists();
@@ -209,6 +219,7 @@
 
     @Override
     public boolean isEmptyPrefix(String bucket, String path) {
+        guardian.checkReadAccess(bucket, path);
         profiler.objectsList();
         Page<Blob> blobs = gcsClient.list(bucket, BlobListOption.prefix(path));
         return !blobs.hasNextPage();
@@ -222,6 +233,8 @@
 
     @Override
     public JsonNode listAsJson(ObjectMapper objectMapper, String bucket) {
+        guardian.checkReadAccess(bucket, "/");
+        profiler.objectsList();
         Page<Blob> blobs = gcsClient.list(bucket, BlobListOption.fields(Storage.BlobField.SIZE));
         ArrayNode objectsInfo = objectMapper.createArrayNode();
 
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/GCSExternalFileWriterFactory.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/GCSExternalFileWriterFactory.java
index 3245869..9e9c003 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/GCSExternalFileWriterFactory.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/GCSExternalFileWriterFactory.java
@@ -21,6 +21,7 @@
 import java.io.IOException;
 
 import org.apache.asterix.cloud.clients.ICloudClient;
+import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.cloud.clients.google.gcs.GCSClientConfig;
 import org.apache.asterix.cloud.clients.google.gcs.GCSCloudClient;
 import org.apache.asterix.common.exceptions.CompilationException;
@@ -62,7 +63,8 @@
     @Override
     ICloudClient createCloudClient() throws CompilationException {
         GCSClientConfig config = GCSClientConfig.of(configuration);
-        return new GCSCloudClient(config, GCSUtils.buildClient(configuration));
+        return new GCSCloudClient(config, GCSUtils.buildClient(configuration),
+                ICloudGuardian.NoOpCloudGuardian.INSTANCE);
     }
 
     @Override
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/S3ExternalFileWriterFactory.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/S3ExternalFileWriterFactory.java
index 96aa929..dcaf488 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/S3ExternalFileWriterFactory.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/S3ExternalFileWriterFactory.java
@@ -21,6 +21,7 @@
 import java.io.IOException;
 
 import org.apache.asterix.cloud.clients.ICloudClient;
+import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.cloud.clients.aws.s3.S3ClientConfig;
 import org.apache.asterix.cloud.clients.aws.s3.S3CloudClient;
 import org.apache.asterix.common.exceptions.CompilationException;
@@ -62,7 +63,8 @@
     @Override
     ICloudClient createCloudClient() throws CompilationException {
         S3ClientConfig config = S3ClientConfig.of(configuration);
-        return new S3CloudClient(config, S3Utils.buildAwsS3Client(configuration));
+        return new S3CloudClient(config, S3Utils.buildAwsS3Client(configuration),
+                ICloudGuardian.NoOpCloudGuardian.INSTANCE);
     }
 
     @Override
diff --git a/asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/gcs/LSMGCSTest.java b/asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/gcs/LSMGCSTest.java
index 86bb5ad..3c62cce 100644
--- a/asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/gcs/LSMGCSTest.java
+++ b/asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/gcs/LSMGCSTest.java
@@ -19,6 +19,7 @@
 package org.apache.asterix.cloud.gcs;
 
 import org.apache.asterix.cloud.AbstractLSMTest;
+import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.cloud.clients.google.gcs.GCSClientConfig;
 import org.apache.asterix.cloud.clients.google.gcs.GCSCloudClient;
 import org.junit.AfterClass;
@@ -48,7 +49,7 @@
                 .setLocation(MOCK_SERVER_REGION).build());
         LOGGER.info("Client created successfully");
         GCSClientConfig config = new GCSClientConfig(MOCK_SERVER_REGION, MOCK_SERVER_HOSTNAME, "", true, 0);
-        CLOUD_CLIENT = new GCSCloudClient(config);
+        CLOUD_CLIENT = new GCSCloudClient(config, ICloudGuardian.NoOpCloudGuardian.INSTANCE);
     }
 
     private static void cleanup() {
diff --git a/asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/s3/LSMS3Test.java b/asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/s3/LSMS3Test.java
index c785e57..01d3422 100644
--- a/asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/s3/LSMS3Test.java
+++ b/asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/s3/LSMS3Test.java
@@ -21,6 +21,7 @@
 import java.net.URI;
 
 import org.apache.asterix.cloud.AbstractLSMTest;
+import org.apache.asterix.cloud.clients.ICloudGuardian;
 import org.apache.asterix.cloud.clients.aws.s3.S3ClientConfig;
 import org.apache.asterix.cloud.clients.aws.s3.S3CloudClient;
 import org.junit.AfterClass;
@@ -65,7 +66,7 @@
         client.createBucket(CreateBucketRequest.builder().bucket(PLAYGROUND_CONTAINER).build());
         LOGGER.info("Client created successfully");
         S3ClientConfig config = new S3ClientConfig(MOCK_SERVER_REGION, MOCK_SERVER_HOSTNAME, "", true, 0);
-        CLOUD_CLIENT = new S3CloudClient(config);
+        CLOUD_CLIENT = new S3CloudClient(config, ICloudGuardian.NoOpCloudGuardian.INSTANCE);
     }
 
     private static void cleanup() {
diff --git a/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/Span.java b/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/Span.java
index 1a40360..26d7f1c 100644
--- a/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/Span.java
+++ b/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/Span.java
@@ -93,7 +93,6 @@
 
     private Span(long span, TimeUnit unit) {
         spanNanos = unit.toNanos(span);
-        reset();
     }
 
     public void reset() {
@@ -109,6 +108,12 @@
     }
 
     public static Span start(long span, TimeUnit unit) {
+        Span s = new Span(span, unit);
+        s.reset();
+        return s;
+    }
+
+    public static Span init(long span, TimeUnit unit) {
         return new Span(span, unit);
     }