HDDS-2244. Use new ReadWrite lock in OzoneManager. (#1589)

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/lock/OzoneManagerLock.java

@@ -21,6 +21,7 @@ package org.apache.hadoop.ozone.om.lock;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.function.Consumer;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -75,6 +76,9 @@ public class OzoneManagerLock {
   private static final Logger LOG =
       LoggerFactory.getLogger(OzoneManagerLock.class);
 
+  private static final String READ_LOCK = "read";
+  private static final String WRITE_LOCK = "write";
+
   private final LockManager<String> manager;
   private final ThreadLocal<Short> lockSet = ThreadLocal.withInitial(
       () -> Short.valueOf((short)0));
@@ -105,15 +109,66 @@ public class OzoneManagerLock {
    * should be bucket name. For remaining all resource only one param should
    * be passed.
    */
+  @Deprecated
   public boolean acquireLock(Resource resource, String... resources) {
     String resourceName = generateResourceName(resource, resources);
+    return lock(resource, resourceName, manager::writeLock, WRITE_LOCK);
+  }
+
+  /**
+   * Acquire read lock on resource.
+   *
+   * For S3_BUCKET_LOCK, VOLUME_LOCK, BUCKET_LOCK type resource, same
+   * thread acquiring lock again is allowed.
+   *
+   * For USER_LOCK, PREFIX_LOCK, S3_SECRET_LOCK type resource, same thread
+   * acquiring lock again is not allowed.
+   *
+   * Special Note for USER_LOCK: Single thread can acquire single user lock/
+   * multi user lock. But not both at the same time.
+   * @param resource - Type of the resource.
+   * @param resources - Resource names on which user want to acquire lock.
+   * For Resource type BUCKET_LOCK, first param should be volume, second param
+   * should be bucket name. For remaining all resource only one param should
+   * be passed.
+   */
+  public boolean acquireReadLock(Resource resource, String... resources) {
+    String resourceName = generateResourceName(resource, resources);
+    return lock(resource, resourceName, manager::readLock, READ_LOCK);
+  }
+
+
+  /**
+   * Acquire write lock on resource.
+   *
+   * For S3_BUCKET_LOCK, VOLUME_LOCK, BUCKET_LOCK type resource, same
+   * thread acquiring lock again is allowed.
+   *
+   * For USER_LOCK, PREFIX_LOCK, S3_SECRET_LOCK type resource, same thread
+   * acquiring lock again is not allowed.
+   *
+   * Special Note for USER_LOCK: Single thread can acquire single user lock/
+   * multi user lock. But not both at the same time.
+   * @param resource - Type of the resource.
+   * @param resources - Resource names on which user want to acquire lock.
+   * For Resource type BUCKET_LOCK, first param should be volume, second param
+   * should be bucket name. For remaining all resource only one param should
+   * be passed.
+   */
+  public boolean acquireWriteLock(Resource resource, String... resources) {
+    String resourceName = generateResourceName(resource, resources);
+    return lock(resource, resourceName, manager::writeLock, WRITE_LOCK);
+  }
+
+  private boolean lock(Resource resource, String resourceName,
+      Consumer<String> lockFn, String lockType) {
     if (!resource.canLock(lockSet.get())) {
       String errorMessage = getErrorMessage(resource);
       LOG.error(errorMessage);
       throw new RuntimeException(errorMessage);
     } else {
-      manager.lock(resourceName);
-      LOG.debug("Acquired {} lock on resource {}", resource.name,
+      lockFn.accept(resourceName);
+      LOG.debug("Acquired {} {} lock on resource {}", lockType, resource.name,
           resourceName);
       lockSet.set(resource.setLock(lockSet.get()));
       return true;
@@ -197,19 +252,19 @@ public class OzoneManagerLock {
 
       if (compare == 0) {
         // both users are equal.
-        manager.lock(firstUser);
+        manager.writeLock(firstUser);
       } else {
-        manager.lock(firstUser);
+        manager.writeLock(firstUser);
         try {
-          manager.lock(secondUser);
+          manager.writeLock(secondUser);
         } catch (Exception ex) {
           // We got an exception acquiring 2nd user lock. Release already
           // acquired user lock, and throw exception to the user.
-          manager.unlock(firstUser);
+          manager.writeUnlock(firstUser);
           throw ex;
         }
       }
-      LOG.debug("Acquired {} lock on resource {} and {}", resource.name,
+      LOG.debug("Acquired Write {} lock on resource {} and {}", resource.name,
           firstUser, secondUser);
       lockSet.set(resource.setLock(lockSet.get()));
       return true;
@@ -240,35 +295,66 @@ public class OzoneManagerLock {
 
     if (compare == 0) {
       // both users are equal.
-      manager.unlock(firstUser);
+      manager.writeUnlock(firstUser);
     } else {
-      manager.unlock(firstUser);
-      manager.unlock(secondUser);
+      manager.writeUnlock(firstUser);
+      manager.writeUnlock(secondUser);
     }
-    LOG.debug("Release {} lock on resource {} and {}", resource.name,
+    LOG.debug("Release Write {} lock on resource {} and {}", resource.name,
         firstUser, secondUser);
     lockSet.set(resource.clearLock(lockSet.get()));
   }
 
   /**
-   * Release lock on resource.
+   * Release write lock on resource.
+   * @param resource - Type of the resource.
+   * @param resources - Resource names on which user want to acquire lock.
+   * For Resource type BUCKET_LOCK, first param should be volume, second param
+   * should be bucket name. For remaining all resource only one param should
+   * be passed.
+   */
+  public void releaseWriteLock(Resource resource, String... resources) {
+    String resourceName = generateResourceName(resource, resources);
+    unlock(resource, resourceName, manager::writeUnlock, WRITE_LOCK);
+  }
+
+  /**
+   * Release read lock on resource.
+   * @param resource - Type of the resource.
+   * @param resources - Resource names on which user want to acquire lock.
+   * For Resource type BUCKET_LOCK, first param should be volume, second param
+   * should be bucket name. For remaining all resource only one param should
+   * be passed.
+   */
+  public void releaseReadLock(Resource resource, String... resources) {
+    String resourceName = generateResourceName(resource, resources);
+    unlock(resource, resourceName, manager::readUnlock, READ_LOCK);
+  }
+
+  /**
+   * Release write lock on resource.
    * @param resource - Type of the resource.
    * @param resources - Resource names on which user want to acquire lock.
    * For Resource type BUCKET_LOCK, first param should be volume, second param
    * should be bucket name. For remaining all resource only one param should
    * be passed.
    */
+  @Deprecated
   public void releaseLock(Resource resource, String... resources) {
     String resourceName = generateResourceName(resource, resources);
+    unlock(resource, resourceName, manager::writeUnlock, WRITE_LOCK);
+  }
+
+  private void unlock(Resource resource, String resourceName,
+      Consumer<String> lockFn, String lockType) {
     // TODO: Not checking release of higher order level lock happened while
     // releasing lower order level lock, as for that we need counter for
     // locks, as some locks support acquiring lock again.
-    manager.unlock(resourceName);
+    lockFn.accept(resourceName);
     // clear lock
-    LOG.debug("Release {}, lock on resource {}", resource.name,
-        resource.name, resourceName);
+    LOG.debug("Release {} {}, lock on resource {}", lockType, resource.name,
+        resourceName);
     lockSet.set(resource.clearLock(lockSet.get()));
-
   }
 
   /**

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/BucketManagerImpl.java

@@ -219,7 +219,8 @@ public class BucketManagerImpl implements BucketManager {
       throws IOException {
     Preconditions.checkNotNull(volumeName);
     Preconditions.checkNotNull(bucketName);
-    metadataManager.getLock().acquireLock(BUCKET_LOCK, volumeName, bucketName);
+    metadataManager.getLock().acquireReadLock(BUCKET_LOCK, volumeName,
+        bucketName);
     try {
       String bucketKey = metadataManager.getBucketKey(volumeName, bucketName);
       OmBucketInfo value = metadataManager.getBucketTable().get(bucketKey);
@@ -237,7 +238,7 @@ public class BucketManagerImpl implements BucketManager {
       }
       throw ex;
     } finally {
-      metadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+      metadataManager.getLock().releaseReadLock(BUCKET_LOCK, volumeName,
           bucketName);
     }
   }
@@ -520,7 +521,7 @@ public class BucketManagerImpl implements BucketManager {
     }
     String volume = obj.getVolumeName();
     String bucket = obj.getBucketName();
-    metadataManager.getLock().acquireLock(BUCKET_LOCK, volume, bucket);
+    metadataManager.getLock().acquireReadLock(BUCKET_LOCK, volume, bucket);
     try {
       String dbBucketKey = metadataManager.getBucketKey(volume, bucket);
       OmBucketInfo bucketInfo =
@@ -538,7 +539,7 @@ public class BucketManagerImpl implements BucketManager {
       }
       throw ex;
     } finally {
-      metadataManager.getLock().releaseLock(BUCKET_LOCK, volume, bucket);
+      metadataManager.getLock().releaseReadLock(BUCKET_LOCK, volume, bucket);
     }
   }
 
@@ -557,7 +558,7 @@ public class BucketManagerImpl implements BucketManager {
 
     String volume = ozObject.getVolumeName();
     String bucket = ozObject.getBucketName();
-    metadataManager.getLock().acquireLock(BUCKET_LOCK, volume, bucket);
+    metadataManager.getLock().acquireReadLock(BUCKET_LOCK, volume, bucket);
     try {
       String dbBucketKey = metadataManager.getBucketKey(volume, bucket);
       OmBucketInfo bucketInfo =
@@ -581,7 +582,7 @@ public class BucketManagerImpl implements BucketManager {
       throw new OMException("Check access operation failed for " +
           "bucket:" + bucket, ex, INTERNAL_ERROR);
     } finally {
-      metadataManager.getLock().releaseLock(BUCKET_LOCK, volume, bucket); 
+      metadataManager.getLock().releaseReadLock(BUCKET_LOCK, volume, bucket);
     }
   }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java

@@ -622,7 +622,8 @@ public class KeyManagerImpl implements KeyManager {
     String volumeName = args.getVolumeName();
     String bucketName = args.getBucketName();
     String keyName = args.getKeyName();
-    metadataManager.getLock().acquireLock(BUCKET_LOCK, volumeName, bucketName);
+    metadataManager.getLock().acquireReadLock(BUCKET_LOCK, volumeName,
+        bucketName);
     try {
       String keyBytes = metadataManager.getOzoneKey(
           volumeName, bucketName, keyName);
@@ -682,7 +683,7 @@ public class KeyManagerImpl implements KeyManager {
       throw new OMException(ex.getMessage(),
           KEY_NOT_FOUND);
     } finally {
-      metadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+      metadataManager.getLock().releaseReadLock(BUCKET_LOCK, volumeName,
           bucketName);
     }
   }
@@ -1311,7 +1312,8 @@ public class KeyManagerImpl implements KeyManager {
     Preconditions.checkNotNull(volumeName);
     Preconditions.checkNotNull(bucketName);
 
-    metadataManager.getLock().acquireLock(BUCKET_LOCK, volumeName, bucketName);
+    metadataManager.getLock().acquireReadLock(BUCKET_LOCK, volumeName,
+        bucketName);
     try {
 
       List<String> multipartUploadKeys =
@@ -1354,7 +1356,7 @@ public class KeyManagerImpl implements KeyManager {
       throw new OMException(ex.getMessage(), ResultCodes
           .LIST_MULTIPART_UPLOAD_PARTS_FAILED);
     } finally {
-      metadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+      metadataManager.getLock().releaseReadLock(BUCKET_LOCK, volumeName,
           bucketName);
     }
   }
@@ -1370,7 +1372,8 @@ public class KeyManagerImpl implements KeyManager {
     boolean isTruncated = false;
     int nextPartNumberMarker = 0;
 
-    metadataManager.getLock().acquireLock(BUCKET_LOCK, volumeName, bucketName);
+    metadataManager.getLock().acquireReadLock(BUCKET_LOCK, volumeName,
+        bucketName);
     try {
       String multipartKey = metadataManager.getMultipartKey(volumeName,
           bucketName, keyName, uploadID);
@@ -1457,7 +1460,7 @@ public class KeyManagerImpl implements KeyManager {
       throw new OMException(ex.getMessage(), ResultCodes
               .LIST_MULTIPART_UPLOAD_PARTS_FAILED);
     } finally {
-      metadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+      metadataManager.getLock().releaseReadLock(BUCKET_LOCK, volumeName,
           bucketName);
     }
   }
@@ -1603,7 +1606,7 @@ public class KeyManagerImpl implements KeyManager {
     String bucket = obj.getBucketName();
     String keyName = obj.getKeyName();
 
-    metadataManager.getLock().acquireLock(BUCKET_LOCK, volume, bucket);
+    metadataManager.getLock().acquireReadLock(BUCKET_LOCK, volume, bucket);
     try {
       validateBucket(volume, bucket);
       String objectKey = metadataManager.getOzoneKey(volume, bucket, keyName);
@@ -1620,7 +1623,7 @@ public class KeyManagerImpl implements KeyManager {
       }
       throw ex;
     } finally {
-      metadataManager.getLock().releaseLock(BUCKET_LOCK, volume, bucket);
+      metadataManager.getLock().releaseReadLock(BUCKET_LOCK, volume, bucket);
     }
   }
 
@@ -1648,7 +1651,7 @@ public class KeyManagerImpl implements KeyManager {
         .setKeyName(keyName)
         .build();
 
-    metadataManager.getLock().acquireLock(BUCKET_LOCK, volume, bucket);
+    metadataManager.getLock().acquireReadLock(BUCKET_LOCK, volume, bucket);
     try {
       validateBucket(volume, bucket);
       OmKeyInfo keyInfo = null;
@@ -1687,7 +1690,7 @@ public class KeyManagerImpl implements KeyManager {
       throw new OMException("Check access operation failed for " +
           "key:" + keyName, ex, INTERNAL_ERROR);
     } finally {
-      metadataManager.getLock().releaseLock(BUCKET_LOCK, volume, bucket);
+      metadataManager.getLock().releaseReadLock(BUCKET_LOCK, volume, bucket);
     }
   }
 
@@ -1732,7 +1735,8 @@ public class KeyManagerImpl implements KeyManager {
     String bucketName = args.getBucketName();
     String keyName = args.getKeyName();
 
-    metadataManager.getLock().acquireLock(BUCKET_LOCK, volumeName, bucketName);
+    metadataManager.getLock().acquireReadLock(BUCKET_LOCK, volumeName,
+        bucketName);
     try {
       // Check if this is the root of the filesystem.
       if (keyName.length() == 0) {
@@ -1770,7 +1774,7 @@ public class KeyManagerImpl implements KeyManager {
           volumeName + " bucket: " + bucketName + " key: " + keyName,
           FILE_NOT_FOUND);
     } finally {
-      metadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+      metadataManager.getLock().releaseReadLock(BUCKET_LOCK, volumeName,
           bucketName);
     }
   }
@@ -1915,7 +1919,8 @@ public class KeyManagerImpl implements KeyManager {
     String bucketName = args.getBucketName();
     String keyName = args.getKeyName();
 
-    metadataManager.getLock().acquireLock(BUCKET_LOCK, volumeName, bucketName);
+    metadataManager.getLock().acquireReadLock(BUCKET_LOCK, volumeName,
+        bucketName);
     try {
       OzoneFileStatus fileStatus = getFileStatus(args);
       if (fileStatus.isFile()) {
@@ -1926,7 +1931,7 @@ public class KeyManagerImpl implements KeyManager {
       }
       //if key is not of type file or if key is not found we throw an exception
     } finally {
-      metadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+      metadataManager.getLock().releaseReadLock(BUCKET_LOCK, volumeName,
           bucketName);
     }
 
@@ -1953,7 +1958,8 @@ public class KeyManagerImpl implements KeyManager {
     String keyName = args.getKeyName();
 
     List<OzoneFileStatus> fileStatusList = new ArrayList<>();
-    metadataManager.getLock().acquireLock(BUCKET_LOCK, volumeName, bucketName);
+    metadataManager.getLock().acquireReadLock(BUCKET_LOCK, volumeName,
+        bucketName);
     try {
       if (Strings.isNullOrEmpty(startKey)) {
         OzoneFileStatus fileStatus = getFileStatus(args);
@@ -2015,7 +2021,7 @@ public class KeyManagerImpl implements KeyManager {
         }
       }
     } finally {
-      metadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+      metadataManager.getLock().releaseReadLock(BUCKET_LOCK, volumeName,
           bucketName);
     }
     return fileStatusList;

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/VolumeManagerImpl.java

@@ -319,7 +319,7 @@ public class VolumeManagerImpl implements VolumeManager {
   @Override
   public OmVolumeArgs getVolumeInfo(String volume) throws IOException {
     Preconditions.checkNotNull(volume);
-    metadataManager.getLock().acquireLock(VOLUME_LOCK, volume);
+    metadataManager.getLock().acquireReadLock(VOLUME_LOCK, volume);
     try {
       String dbVolumeKey = metadataManager.getVolumeKey(volume);
       OmVolumeArgs volumeArgs =
@@ -337,7 +337,7 @@ public class VolumeManagerImpl implements VolumeManager {
       }
       throw ex;
     } finally {
-      metadataManager.getLock().releaseLock(VOLUME_LOCK, volume);
+      metadataManager.getLock().releaseReadLock(VOLUME_LOCK, volume);
     }
   }
 
@@ -423,7 +423,7 @@ public class VolumeManagerImpl implements VolumeManager {
       throws IOException {
     Preconditions.checkNotNull(volume);
     Preconditions.checkNotNull(userAcl);
-    metadataManager.getLock().acquireLock(VOLUME_LOCK, volume);
+    metadataManager.getLock().acquireReadLock(VOLUME_LOCK, volume);
     try {
       String dbVolumeKey = metadataManager.getVolumeKey(volume);
       OmVolumeArgs volumeArgs =
@@ -443,7 +443,7 @@ public class VolumeManagerImpl implements VolumeManager {
       }
       throw ex;
     } finally {
-      metadataManager.getLock().releaseLock(VOLUME_LOCK, volume);
+      metadataManager.getLock().releaseReadLock(VOLUME_LOCK, volume);
     }
   }
 
@@ -634,7 +634,7 @@ public class VolumeManagerImpl implements VolumeManager {
           "VolumeManager. OzoneObj type:" + obj.getResourceType());
     }
     String volume = obj.getVolumeName();
-    metadataManager.getLock().acquireLock(VOLUME_LOCK, volume);
+    metadataManager.getLock().acquireReadLock(VOLUME_LOCK, volume);
     try {
       String dbVolumeKey = metadataManager.getVolumeKey(volume);
       OmVolumeArgs volumeArgs =
@@ -653,7 +653,7 @@ public class VolumeManagerImpl implements VolumeManager {
       }
       throw ex;
     } finally {
-      metadataManager.getLock().releaseLock(VOLUME_LOCK, volume);
+      metadataManager.getLock().releaseReadLock(VOLUME_LOCK, volume);
     }
   }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketCreateRequest.java

@@ -148,10 +148,10 @@ public class OMBucketCreateRequest extends OMClientRequest {
             volumeName, bucketName, null);
       }
 
-      acquiredVolumeLock = metadataManager.getLock().acquireLock(VOLUME_LOCK,
-          volumeName);
-      acquiredBucketLock = metadataManager.getLock().acquireLock(BUCKET_LOCK,
-          volumeName, bucketName);
+      acquiredVolumeLock =
+          metadataManager.getLock().acquireReadLock(VOLUME_LOCK, volumeName);
+      acquiredBucketLock = metadataManager.getLock().acquireWriteLock(
+          BUCKET_LOCK, volumeName, bucketName);
 
       OmVolumeArgs omVolumeArgs =
           metadataManager.getVolumeTable().get(volumeKey);
@@ -191,11 +191,11 @@ public class OMBucketCreateRequest extends OMClientRequest {
                 transactionLogIndex));
       }
       if (acquiredBucketLock) {
-        metadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+        metadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
             bucketName);
       }
       if (acquiredVolumeLock) {
-        metadataManager.getLock().releaseLock(VOLUME_LOCK, volumeName);
+        metadataManager.getLock().releaseReadLock(VOLUME_LOCK, volumeName);
       }
     }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketDeleteRequest.java

@@ -50,6 +50,7 @@ import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
 import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
 
 import static org.apache.hadoop.ozone.om.lock.OzoneManagerLock.Resource.BUCKET_LOCK;
+import static org.apache.hadoop.ozone.om.lock.OzoneManagerLock.Resource.VOLUME_LOCK;
 
 /**
  * Handles DeleteBucket Request.
@@ -87,7 +88,8 @@ public class OMBucketDeleteRequest extends OMClientRequest {
     OzoneManagerProtocolProtos.UserInfo userInfo = getOmRequest().getUserInfo();
     IOException exception = null;
 
-    boolean acquiredLock = false;
+    boolean acquiredBucketLock = false;
+    boolean acquiredVolumeLock = false;
     OMClientResponse omClientResponse = null;
     try {
       // check Acl
@@ -99,7 +101,10 @@ public class OMBucketDeleteRequest extends OMClientRequest {
 
 
       // acquire lock
-      acquiredLock = omMetadataManager.getLock().acquireLock(BUCKET_LOCK,
+      acquiredVolumeLock =
+          omMetadataManager.getLock().acquireReadLock(VOLUME_LOCK, volumeName);
+      acquiredBucketLock =
+          omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK,
           volumeName, bucketName);
 
       // No need to check volume exists here, as bucket cannot be created
@@ -142,10 +147,13 @@ public class OMBucketDeleteRequest extends OMClientRequest {
             ozoneManagerDoubleBufferHelper.add(omClientResponse,
                 transactionLogIndex));
       }
-      if (acquiredLock) {
-        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+      if (acquiredBucketLock) {
+        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
             bucketName);
       }
+      if (acquiredVolumeLock) {
+        omMetadataManager.getLock().releaseReadLock(VOLUME_LOCK, volumeName);
+      }
     }
 
     // Performing audit logging outside of the lock.

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketSetPropertyRequest.java

@@ -102,7 +102,7 @@ public class OMBucketSetPropertyRequest extends OMClientRequest {
     AuditLogger auditLogger = ozoneManager.getAuditLogger();
     OzoneManagerProtocolProtos.UserInfo userInfo = getOmRequest().getUserInfo();
     IOException exception = null;
-    boolean acquiredLock = false;
+    boolean acquiredBucketLock = false;
     OMClientResponse omClientResponse = null;
     try {
       // check Acl
@@ -112,10 +112,9 @@ public class OMBucketSetPropertyRequest extends OMClientRequest {
             volumeName, bucketName, null);
       }
 
-
-      // acquire lock
-      acquiredLock =  omMetadataManager.getLock().acquireLock(BUCKET_LOCK,
-          volumeName, bucketName);
+      // acquire lock.
+      acquiredBucketLock =  omMetadataManager.getLock().acquireWriteLock(
+          BUCKET_LOCK, volumeName, bucketName);
 
       String bucketKey = omMetadataManager.getBucketKey(volumeName, bucketName);
       OmBucketInfo oldBucketInfo =
@@ -181,8 +180,8 @@ public class OMBucketSetPropertyRequest extends OMClientRequest {
             ozoneManagerDoubleBufferHelper.add(omClientResponse,
                 transactionLogIndex));
       }
-      if (acquiredLock) {
-        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+      if (acquiredBucketLock) {
+        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
             bucketName);
       }
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/acl/OMBucketAclRequest.java

@@ -91,7 +91,8 @@ public abstract class OMBucketAclRequest extends OMClientRequest {
             volume, null, null);
       }
       lockAcquired =
-          omMetadataManager.getLock().acquireLock(BUCKET_LOCK, volume, bucket);
+          omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK, volume,
+              bucket);
 
       String dbBucketKey = omMetadataManager.getBucketKey(volume, bucket);
       omBucketInfo = omMetadataManager.getBucketTable().get(dbBucketKey);
@@ -120,7 +121,8 @@ public abstract class OMBucketAclRequest extends OMClientRequest {
                 transactionLogIndex));
       }
       if (lockAcquired) {
-        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volume, bucket);
+        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volume,
+            bucket);
       }
     }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMDirectoryCreateRequest.java

@@ -136,7 +136,7 @@ public class OMDirectoryCreateRequest extends OMKeyRequest {
                 CreateDirectoryResponse.newBuilder()).build());
       }
       // acquire lock
-      acquiredLock = omMetadataManager.getLock().acquireLock(BUCKET_LOCK,
+      acquiredLock = omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK,
           volumeName, bucketName);
 
       // TODO: Not checking volume exist here, once we have full cache we can
@@ -192,7 +192,7 @@ public class OMDirectoryCreateRequest extends OMKeyRequest {
                 transactionLogIndex));
       }
       if (acquiredLock) {
-        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
             bucketName);
       }
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java

@@ -180,7 +180,7 @@ public class OMFileCreateRequest extends OMKeyRequest {
       checkBucketAcls(ozoneManager, volumeName, bucketName, keyName);
 
       // acquire lock
-      acquiredLock = omMetadataManager.getLock().acquireLock(BUCKET_LOCK,
+      acquiredLock = omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK,
           volumeName, bucketName);
 
       OmBucketInfo bucketInfo =
@@ -280,7 +280,7 @@ public class OMFileCreateRequest extends OMKeyRequest {
                 transactionLogIndex));
       }
       if (acquiredLock) {
-        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
             bucketName);
       }
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java

@@ -127,7 +127,7 @@ public class OMKeyCommitRequest extends OMKeyRequest {
       String dbOpenKey = omMetadataManager.getOpenKey(volumeName, bucketName,
           keyName, commitKeyRequest.getClientID());
 
-      omMetadataManager.getLock().acquireLock(BUCKET_LOCK, volumeName,
+      omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK, volumeName,
           bucketName);
 
       validateBucketAndVolume(omMetadataManager, volumeName, bucketName);
@@ -166,7 +166,7 @@ public class OMKeyCommitRequest extends OMKeyRequest {
             ozoneManagerDoubleBufferHelper.add(omClientResponse,
                 transactionLogIndex));
       }
-      omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+      omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
           bucketName);
     }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java

@@ -164,7 +164,7 @@ public class OMKeyCreateRequest extends OMKeyRequest {
       // check Acl
       checkBucketAcls(ozoneManager, volumeName, bucketName, keyName);
 
-      acquireLock = omMetadataManager.getLock().acquireLock(BUCKET_LOCK,
+      acquireLock = omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK,
           volumeName, bucketName);
       validateBucketAndVolume(omMetadataManager, volumeName, bucketName);
       //TODO: We can optimize this get here, if getKmsProvider is null, then
@@ -198,7 +198,7 @@ public class OMKeyCreateRequest extends OMKeyRequest {
                 transactionLogIndex));
       }
       if (acquireLock) {
-        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
             bucketName);
       }
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyDeleteRequest.java

@@ -114,7 +114,7 @@ public class OMKeyDeleteRequest extends OMKeyRequest {
       String objectKey = omMetadataManager.getOzoneKey(
           volumeName, bucketName, keyName);
 
-      acquiredLock = omMetadataManager.getLock().acquireLock(BUCKET_LOCK,
+      acquiredLock = omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK,
           volumeName, bucketName);
 
       // Not doing bucket/volume checks here. In this way we can avoid db
@@ -153,7 +153,7 @@ public class OMKeyDeleteRequest extends OMKeyRequest {
                 transactionLogIndex));
       }
       if (acquiredLock) {
-        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
             bucketName);
       }
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRenameRequest.java

@@ -120,7 +120,7 @@ public class OMKeyRenameRequest extends OMKeyRequest {
       // check Acl
       checkKeyAcls(ozoneManager, volumeName, bucketName, fromKeyName);
 
-      acquiredLock = omMetadataManager.getLock().acquireLock(BUCKET_LOCK,
+      acquiredLock = omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK,
           volumeName, bucketName);
 
       // Not doing bucket/volume checks here. In this way we can avoid db
@@ -176,7 +176,7 @@ public class OMKeyRenameRequest extends OMKeyRequest {
                 transactionLogIndex));
       }
       if (acquiredLock) {
-        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
             bucketName);
       }
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/OMKeyAclRequest.java

@@ -81,7 +81,8 @@ public abstract class OMKeyAclRequest extends OMClientRequest {
             volume, bucket, key);
       }
       lockAcquired =
-          omMetadataManager.getLock().acquireLock(BUCKET_LOCK, volume, bucket);
+          omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK, volume,
+              bucket);
 
       String dbKey = omMetadataManager.getOzoneKey(volume, bucket, key);
       omKeyInfo = omMetadataManager.getKeyTable().get(dbKey);
@@ -111,7 +112,8 @@ public abstract class OMKeyAclRequest extends OMClientRequest {
                 transactionLogIndex));
       }
       if (lockAcquired) {
-        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volume, bucket);
+        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volume,
+            bucket);
       }
     }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/acl/prefix/OMPrefixAclRequest.java

@@ -81,7 +81,7 @@ public abstract class OMPrefixAclRequest extends OMClientRequest {
       }
 
       lockAcquired =
-          omMetadataManager.getLock().acquireLock(PREFIX_LOCK, prefixPath);
+          omMetadataManager.getLock().acquireWriteLock(PREFIX_LOCK, prefixPath);
 
       omPrefixInfo = omMetadataManager.getPrefixTable().get(prefixPath);
 
@@ -128,7 +128,7 @@ public abstract class OMPrefixAclRequest extends OMClientRequest {
                 transactionLogIndex));
       }
       if (lockAcquired) {
-        omMetadataManager.getLock().releaseLock(PREFIX_LOCK,
+        omMetadataManager.getLock().releaseWriteLock(PREFIX_LOCK,
             getOzoneObj().getPath());
       }
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/bucket/S3BucketCreateRequest.java

@@ -153,8 +153,8 @@ public class S3BucketCreateRequest extends OMVolumeRequest {
     try {
 
       // TODO to support S3 ACL later.
-      acquiredS3Lock = omMetadataManager.getLock().acquireLock(S3_BUCKET_LOCK,
-          s3BucketName);
+      acquiredS3Lock = omMetadataManager.getLock().acquireWriteLock(
+          S3_BUCKET_LOCK, s3BucketName);
 
       // First check if this s3Bucket exists
       if (omMetadataManager.getS3Table().isExist(s3BucketName)) {
@@ -165,9 +165,10 @@ public class S3BucketCreateRequest extends OMVolumeRequest {
       OMVolumeCreateResponse omVolumeCreateResponse = null;
       try {
         acquiredVolumeLock =
-            omMetadataManager.getLock().acquireLock(VOLUME_LOCK, volumeName);
-        acquiredUserLock = omMetadataManager.getLock().acquireLock(USER_LOCK,
-            userName);
+            omMetadataManager.getLock().acquireWriteLock(VOLUME_LOCK,
+                volumeName);
+        acquiredUserLock = omMetadataManager.getLock().acquireWriteLock(
+            USER_LOCK, userName);
         // Check if volume exists, if it does not exist create
         // ozone volume.
         String volumeKey = omMetadataManager.getVolumeKey(volumeName);
@@ -188,10 +189,10 @@ public class S3BucketCreateRequest extends OMVolumeRequest {
         }
       } finally {
         if (acquiredUserLock) {
-          omMetadataManager.getLock().releaseLock(USER_LOCK, userName);
+          omMetadataManager.getLock().releaseWriteLock(USER_LOCK, userName);
         }
         if (acquiredVolumeLock) {
-          omMetadataManager.getLock().releaseLock(VOLUME_LOCK, volumeName);
+          omMetadataManager.getLock().releaseWriteLock(VOLUME_LOCK, volumeName);
         }
       }
 
@@ -227,7 +228,8 @@ public class S3BucketCreateRequest extends OMVolumeRequest {
                 transactionLogIndex));
       }
       if (acquiredS3Lock) {
-        omMetadataManager.getLock().releaseLock(S3_BUCKET_LOCK, s3BucketName);
+        omMetadataManager.getLock().releaseWriteLock(
+            S3_BUCKET_LOCK, s3BucketName);
       }
     }
 
@@ -266,7 +268,7 @@ public class S3BucketCreateRequest extends OMVolumeRequest {
     OmBucketInfo omBucketInfo = null;
     try {
       acquireBucketLock =
-          omMetadataManager.getLock().acquireLock(BUCKET_LOCK, volumeName,
+          omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK, volumeName,
               s3BucketName);
       String bucketKey = omMetadataManager.getBucketKey(volumeName,
           s3BucketName);
@@ -285,7 +287,7 @@ public class S3BucketCreateRequest extends OMVolumeRequest {
       }
     } finally {
       if (acquireBucketLock) {
-        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
             s3BucketName);
       }
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/bucket/S3BucketDeleteRequest.java

@@ -106,8 +106,8 @@ public class S3BucketDeleteRequest extends OMVolumeRequest {
     OMClientResponse omClientResponse = null;
     try {
       // TODO to support S3 ACL later.
-      acquiredS3Lock = omMetadataManager.getLock().acquireLock(S3_BUCKET_LOCK,
-          s3BucketName);
+      acquiredS3Lock = omMetadataManager.getLock().acquireWriteLock(
+          S3_BUCKET_LOCK, s3BucketName);
 
       String s3Mapping = omMetadataManager.getS3Table().get(s3BucketName);
 
@@ -118,8 +118,8 @@ public class S3BucketDeleteRequest extends OMVolumeRequest {
         volumeName = getOzoneVolumeName(s3Mapping);
 
         acquiredBucketLock =
-            omMetadataManager.getLock().acquireLock(BUCKET_LOCK, volumeName,
-                s3BucketName);
+            omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK,
+                volumeName, s3BucketName);
 
         String bucketKey = omMetadataManager.getBucketKey(volumeName,
             s3BucketName);
@@ -149,11 +149,12 @@ public class S3BucketDeleteRequest extends OMVolumeRequest {
                 transactionLogIndex));
       }
       if (acquiredBucketLock) {
-        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
             s3BucketName);
       }
       if (acquiredS3Lock) {
-        omMetadataManager.getLock().releaseLock(S3_BUCKET_LOCK, s3BucketName);
+        omMetadataManager.getLock().releaseWriteLock(S3_BUCKET_LOCK,
+            s3BucketName);
       }
     }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3InitiateMultipartUploadRequest.java

@@ -114,7 +114,7 @@ public class S3InitiateMultipartUploadRequest extends OMKeyRequest {
     try {
       // TODO to support S3 ACL later.
       acquiredBucketLock =
-          omMetadataManager.getLock().acquireLock(BUCKET_LOCK, volumeName,
+          omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK, volumeName,
               bucketName);
 
       validateBucketAndVolume(omMetadataManager, volumeName, bucketName);
@@ -189,7 +189,7 @@ public class S3InitiateMultipartUploadRequest extends OMKeyRequest {
                 transactionLogIndex));
       }
       if (acquiredBucketLock) {
-        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
             bucketName);
       }
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadAbortRequest.java

@@ -99,7 +99,7 @@ public class S3MultipartUploadAbortRequest extends OMKeyRequest {
     try {
       // TODO to support S3 ACL later.
       acquiredLock =
-          omMetadataManager.getLock().acquireLock(BUCKET_LOCK, volumeName,
+          omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK, volumeName,
               bucketName);
 
       validateBucketAndVolume(omMetadataManager, volumeName, bucketName);
@@ -147,7 +147,7 @@ public class S3MultipartUploadAbortRequest extends OMKeyRequest {
                 transactionLogIndex));
       }
       if (acquiredLock) {
-        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
             bucketName);
       }
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java

@@ -111,7 +111,7 @@ public class S3MultipartUploadCommitPartRequest extends OMKeyRequest {
     try {
       // TODO to support S3 ACL later.
       acquiredLock =
-          omMetadataManager.getLock().acquireLock(BUCKET_LOCK, volumeName,
+          omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK, volumeName,
               bucketName);
 
       validateBucketAndVolume(omMetadataManager, volumeName, bucketName);
@@ -203,7 +203,7 @@ public class S3MultipartUploadCommitPartRequest extends OMKeyRequest {
                 transactionLogIndex));
       }
       if (acquiredLock) {
-        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
             bucketName);
       }
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCompleteRequest.java

@@ -124,7 +124,7 @@ public class S3MultipartUploadCompleteRequest extends OMKeyRequest {
 
       multipartUploadList = new OmMultipartUploadCompleteList(partsMap);
 
-      acquiredLock = omMetadataManager.getLock().acquireLock(BUCKET_LOCK,
+      acquiredLock = omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK,
           volumeName, bucketName);
 
       validateBucketAndVolume(omMetadataManager, volumeName, bucketName);
@@ -270,7 +270,7 @@ public class S3MultipartUploadCompleteRequest extends OMKeyRequest {
                 transactionLogIndex));
       }
       if (acquiredLock) {
-        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+        omMetadataManager.getLock().releaseWriteLock(BUCKET_LOCK, volumeName,
             bucketName);
       }
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/security/S3GetSecretRequest.java

@@ -127,7 +127,8 @@ public class S3GetSecretRequest extends OMClientRequest {
     try {
       String awsSecret = updateGetS3SecretRequest.getAwsSecret();
       acquiredLock =
-         omMetadataManager.getLock().acquireLock(S3_SECRET_LOCK, kerberosID);
+         omMetadataManager.getLock().acquireWriteLock(S3_SECRET_LOCK,
+             kerberosID);
 
       S3SecretValue s3SecretValue =
           omMetadataManager.getS3SecretTable().get(kerberosID);
@@ -168,7 +169,8 @@ public class S3GetSecretRequest extends OMClientRequest {
             omClientResponse, transactionLogIndex));
       }
       if (acquiredLock) {
-        omMetadataManager.getLock().releaseLock(S3_SECRET_LOCK, kerberosID);
+        omMetadataManager.getLock().releaseWriteLock(S3_SECRET_LOCK,
+            kerberosID);
       }
     }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/volume/OMVolumeCreateRequest.java

@@ -135,10 +135,10 @@ public class OMVolumeCreateRequest extends OMVolumeRequest {
       UserVolumeInfo volumeList = null;
 
       // acquire lock.
-      acquiredVolumeLock = omMetadataManager.getLock().acquireLock(VOLUME_LOCK,
-          volume);
+      acquiredVolumeLock = omMetadataManager.getLock().acquireWriteLock(
+          VOLUME_LOCK, volume);
 
-      acquiredUserLock = omMetadataManager.getLock().acquireLock(USER_LOCK,
+      acquiredUserLock = omMetadataManager.getLock().acquireWriteLock(USER_LOCK,
           owner);
 
       String dbVolumeKey = omMetadataManager.getVolumeKey(volume);
@@ -176,10 +176,10 @@ public class OMVolumeCreateRequest extends OMVolumeRequest {
                 transactionLogIndex));
       }
       if (acquiredUserLock) {
-        omMetadataManager.getLock().releaseLock(USER_LOCK, owner);
+        omMetadataManager.getLock().releaseWriteLock(USER_LOCK, owner);
       }
       if (acquiredVolumeLock) {
-        omMetadataManager.getLock().releaseLock(VOLUME_LOCK, volume);
+        omMetadataManager.getLock().releaseWriteLock(VOLUME_LOCK, volume);
       }
     }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/volume/OMVolumeDeleteRequest.java

@@ -97,10 +97,10 @@ public class OMVolumeDeleteRequest extends OMVolumeRequest {
       OmVolumeArgs omVolumeArgs = null;
       OzoneManagerProtocolProtos.UserVolumeInfo newVolumeList = null;
 
-      acquiredVolumeLock = omMetadataManager.getLock().acquireLock(VOLUME_LOCK,
-          volume);
+      acquiredVolumeLock = omMetadataManager.getLock().acquireWriteLock(
+          VOLUME_LOCK, volume);
       owner = getVolumeInfo(omMetadataManager, volume).getOwnerName();
-      acquiredUserLock = omMetadataManager.getLock().acquireLock(USER_LOCK,
+      acquiredUserLock = omMetadataManager.getLock().acquireWriteLock(USER_LOCK,
           owner);
 
       String dbUserKey = omMetadataManager.getUserKey(owner);
@@ -141,10 +141,10 @@ public class OMVolumeDeleteRequest extends OMVolumeRequest {
                 transactionLogIndex));
       }
       if (acquiredUserLock) {
-        omMetadataManager.getLock().releaseLock(USER_LOCK, owner);
+        omMetadataManager.getLock().releaseWriteLock(USER_LOCK, owner);
       }
       if (acquiredVolumeLock) {
-        omMetadataManager.getLock().releaseLock(VOLUME_LOCK, volume);
+        omMetadataManager.getLock().releaseWriteLock(VOLUME_LOCK, volume);
       }
     }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/volume/OMVolumeSetOwnerRequest.java

@@ -123,8 +123,8 @@ public class OMVolumeSetOwnerRequest extends OMVolumeRequest {
 
 
 
-      acquiredVolumeLock = omMetadataManager.getLock().acquireLock(VOLUME_LOCK,
-          volume);
+      acquiredVolumeLock = omMetadataManager.getLock().acquireWriteLock(
+          VOLUME_LOCK, volume);
 
       omVolumeArgs = omMetadataManager.getVolumeTable().get(dbVolumeKey);
 
@@ -188,7 +188,7 @@ public class OMVolumeSetOwnerRequest extends OMVolumeRequest {
         omMetadataManager.getLock().releaseMultiUserLock(newOwner, oldOwner);
       }
       if (acquiredVolumeLock) {
-        omMetadataManager.getLock().releaseLock(VOLUME_LOCK, volume);
+        omMetadataManager.getLock().releaseWriteLock(VOLUME_LOCK, volume);
       }
     }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/volume/OMVolumeSetQuotaRequest.java

@@ -114,8 +114,8 @@ public class OMVolumeSetQuotaRequest extends OMVolumeRequest {
 
       OmVolumeArgs omVolumeArgs = null;
 
-      acquireVolumeLock = omMetadataManager.getLock().acquireLock(VOLUME_LOCK,
-          volume);
+      acquireVolumeLock = omMetadataManager.getLock().acquireWriteLock(
+          VOLUME_LOCK, volume);
       String dbVolumeKey = omMetadataManager.getVolumeKey(volume);
       omVolumeArgs = omMetadataManager.getVolumeTable().get(dbVolumeKey);
 
@@ -146,7 +146,7 @@ public class OMVolumeSetQuotaRequest extends OMVolumeRequest {
                 transactionLogIndex));
       }
       if (acquireVolumeLock) {
-        omMetadataManager.getLock().releaseLock(VOLUME_LOCK, volume);
+        omMetadataManager.getLock().releaseWriteLock(VOLUME_LOCK, volume);
       }
     }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/volume/acl/OMVolumeAclRequest.java

@@ -81,7 +81,7 @@ public abstract class OMVolumeAclRequest extends OMClientRequest {
             volume, null, null);
       }
       lockAcquired =
-          omMetadataManager.getLock().acquireLock(VOLUME_LOCK, volume);
+          omMetadataManager.getLock().acquireWriteLock(VOLUME_LOCK, volume);
       String dbVolumeKey = omMetadataManager.getVolumeKey(volume);
       omVolumeArgs = omMetadataManager.getVolumeTable().get(dbVolumeKey);
       if (omVolumeArgs == null) {
@@ -115,7 +115,7 @@ public abstract class OMVolumeAclRequest extends OMClientRequest {
                 transactionLogIndex));
       }
       if (lockAcquired) {
-        omMetadataManager.getLock().releaseLock(VOLUME_LOCK, volume);
+        omMetadataManager.getLock().releaseWriteLock(VOLUME_LOCK, volume);
       }
     }