Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HDFS-11897. Ozone: KSM: Changing log level for client calls in KSM. Contributed by Shashikant Banerjee.

This commit is contained in:
Nandakumar 2017-09-22 20:33:17 +05:30 committed by Owen O'Malley
parent 27ec22e20d
commit 0313bdf886
3 changed files with 90 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/BucketManagerImpl.java
View File

@ -85,14 +85,14 @@ public class BucketManagerImpl implements BucketManager {
byte[] bucketKey = metadataManager.getBucketKey(volumeName, bucketName); byte[] bucketKey = metadataManager.getBucketKey(volumeName, bucketName);
//Check if the volume exists //Check if the volume exists
if(metadataManager.get(volumeKey) == null) { if (metadataManager.get(volumeKey) == null) {
LOG.error("volume: {} not found ", volumeName); LOG.debug("volume: {} not found ", volumeName);
throw new KSMException("Volume doesn't exist", throw new KSMException("Volume doesn't exist",
KSMException.ResultCodes.FAILED_VOLUME_NOT_FOUND); KSMException.ResultCodes.FAILED_VOLUME_NOT_FOUND);
} }
//Check if bucket already exists //Check if bucket already exists
if(metadataManager.get(bucketKey) != null) { if (metadataManager.get(bucketKey) != null) {
LOG.error("bucket: {} already exists ", bucketName); LOG.debug("bucket: {} already exists ", bucketName);
throw new KSMException("Bucket already exist", throw new KSMException("Bucket already exist",
KSMException.ResultCodes.FAILED_BUCKET_ALREADY_EXISTS); KSMException.ResultCodes.FAILED_BUCKET_ALREADY_EXISTS);
} }
@ -100,8 +100,10 @@ public class BucketManagerImpl implements BucketManager {
LOG.debug("created bucket: {} in volume: {}", bucketName, volumeName); LOG.debug("created bucket: {} in volume: {}", bucketName, volumeName);
} catch (IOException | DBException ex) { } catch (IOException | DBException ex) {
LOG.error("Bucket creation failed for bucket:{} in volume:{}", if (!(ex instanceof KSMException)) {
bucketName, volumeName, ex); LOG.error("Bucket creation failed for bucket:{} in volume:{}",
bucketName, volumeName, ex);
}
throw ex; throw ex;
} finally { } finally {
metadataManager.writeLock().unlock(); metadataManager.writeLock().unlock();
@ -123,16 +125,18 @@ public class BucketManagerImpl implements BucketManager {
try { try {
byte[] bucketKey = metadataManager.getBucketKey(volumeName, bucketName); byte[] bucketKey = metadataManager.getBucketKey(volumeName, bucketName);
byte[] value = metadataManager.get(bucketKey); byte[] value = metadataManager.get(bucketKey);
if(value == null) { if (value == null) {
LOG.error("bucket: {} not found in volume: {}.", LOG.debug("bucket: {} not found in volume: {}.", bucketName,
bucketName, volumeName); volumeName);
throw new KSMException("Bucket not found", throw new KSMException("Bucket not found",
KSMException.ResultCodes.FAILED_BUCKET_NOT_FOUND); KSMException.ResultCodes.FAILED_BUCKET_NOT_FOUND);
} }
return KsmBucketInfo.getFromProtobuf(BucketInfo.parseFrom(value)); return KsmBucketInfo.getFromProtobuf(BucketInfo.parseFrom(value));
} catch (IOException | DBException ex) { } catch (IOException | DBException ex) {
LOG.error("Exception while getting bucket info for bucket: {}", if (!(ex instanceof KSMException)) {
bucketName, ex); LOG.error("Exception while getting bucket info for bucket: {}",
bucketName, ex);
}
throw ex; throw ex;
} finally { } finally {
metadataManager.readLock().unlock(); metadataManager.readLock().unlock();
@ -155,14 +159,14 @@ public class BucketManagerImpl implements BucketManager {
//Check if volume exists //Check if volume exists
if(metadataManager.get(metadataManager.getVolumeKey(volumeName)) == if(metadataManager.get(metadataManager.getVolumeKey(volumeName)) ==
null) { null) {
LOG.error("volume: {} not found ", volumeName); LOG.debug("volume: {} not found ", volumeName);
throw new KSMException("Volume doesn't exist", throw new KSMException("Volume doesn't exist",
KSMException.ResultCodes.FAILED_VOLUME_NOT_FOUND); KSMException.ResultCodes.FAILED_VOLUME_NOT_FOUND);
} }
byte[] value = metadataManager.get(bucketKey); byte[] value = metadataManager.get(bucketKey);
//Check if bucket exist //Check if bucket exist
if(value == null) { if(value == null) {
LOG.error("bucket: {} not found ", bucketName); LOG.debug("bucket: {} not found ", bucketName);
throw new KSMException("Bucket doesn't exist", throw new KSMException("Bucket doesn't exist",
KSMException.ResultCodes.FAILED_BUCKET_NOT_FOUND); KSMException.ResultCodes.FAILED_BUCKET_NOT_FOUND);
} }
@ -184,7 +188,7 @@ public class BucketManagerImpl implements BucketManager {
//Check StorageType to update //Check StorageType to update
StorageType storageType = args.getStorageType(); StorageType storageType = args.getStorageType();
if(storageType != null) { if (storageType != null) {
bucketInfoBuilder.setStorageType(storageType); bucketInfoBuilder.setStorageType(storageType);
LOG.debug("Updating bucket storage type for bucket: {} in volume: {}", LOG.debug("Updating bucket storage type for bucket: {} in volume: {}",
bucketName, volumeName); bucketName, volumeName);
@ -194,21 +198,23 @@ public class BucketManagerImpl implements BucketManager {
//Check Versioning to update //Check Versioning to update
Boolean versioning = args.getIsVersionEnabled(); Boolean versioning = args.getIsVersionEnabled();
if(versioning != null) { if (versioning != null) {
bucketInfoBuilder.setIsVersionEnabled(versioning); bucketInfoBuilder.setIsVersionEnabled(versioning);
LOG.debug("Updating bucket versioning for bucket: {} in volume: {}", LOG.debug("Updating bucket versioning for bucket: {} in volume: {}",
bucketName, volumeName); bucketName, volumeName);
} else { } else {
bucketInfoBuilder.setIsVersionEnabled( bucketInfoBuilder
oldBucketInfo.getIsVersionEnabled()); .setIsVersionEnabled(oldBucketInfo.getIsVersionEnabled());
} }
bucketInfoBuilder.setCreationTime(oldBucketInfo.getCreationTime()); bucketInfoBuilder.setCreationTime(oldBucketInfo.getCreationTime());
metadataManager.put(bucketKey, bucketInfoBuilder.build() metadataManager.put(bucketKey,
.getProtobuf().toByteArray()); bucketInfoBuilder.build().getProtobuf().toByteArray());
} catch (IOException | DBException ex) { } catch (IOException | DBException ex) {
LOG.error("Setting bucket property failed for bucket:{} in volume:{}", if (!(ex instanceof KSMException)) {
bucketName, volumeName, ex); LOG.error("Setting bucket property failed for bucket:{} in volume:{}",
bucketName, volumeName, ex);
}
throw ex; throw ex;
} finally { } finally {
metadataManager.writeLock().unlock(); metadataManager.writeLock().unlock();
@ -250,28 +256,30 @@ public class BucketManagerImpl implements BucketManager {
try { try {
byte[] bucketKey = metadataManager.getBucketKey(volumeName, bucketName); byte[] bucketKey = metadataManager.getBucketKey(volumeName, bucketName);
//Check if volume exists //Check if volume exists
if(metadataManager.get(metadataManager.getVolumeKey(volumeName)) == if (metadataManager.get(metadataManager.getVolumeKey(volumeName))
null) { == null) {
LOG.error("volume: {} not found ", volumeName); LOG.debug("volume: {} not found ", volumeName);
throw new KSMException("Volume doesn't exist", throw new KSMException("Volume doesn't exist",
KSMException.ResultCodes.FAILED_VOLUME_NOT_FOUND); KSMException.ResultCodes.FAILED_VOLUME_NOT_FOUND);
} }
//Check if bucket exist //Check if bucket exist
if(metadataManager.get(bucketKey) == null) { if (metadataManager.get(bucketKey) == null) {
LOG.error("bucket: {} not found ", bucketName); LOG.debug("bucket: {} not found ", bucketName);
throw new KSMException("Bucket doesn't exist", throw new KSMException("Bucket doesn't exist",
KSMException.ResultCodes.FAILED_BUCKET_NOT_FOUND); KSMException.ResultCodes.FAILED_BUCKET_NOT_FOUND);
} }
//Check if bucket is empty //Check if bucket is empty
if(!metadataManager.isBucketEmpty(volumeName, bucketName)) { if (!metadataManager.isBucketEmpty(volumeName, bucketName)) {
LOG.error("bucket: {} is not empty ", bucketName); LOG.debug("bucket: {} is not empty ", bucketName);
throw new KSMException("Bucket is not empty", throw new KSMException("Bucket is not empty",
KSMException.ResultCodes.FAILED_BUCKET_NOT_EMPTY); KSMException.ResultCodes.FAILED_BUCKET_NOT_EMPTY);
} }
metadataManager.delete(bucketKey); metadataManager.delete(bucketKey);
} catch (IOException ex) { } catch (IOException ex) {
LOG.error("Delete bucket failed for bucket:{} in volume:{}", if (!(ex instanceof KSMException)) {
bucketName, volumeName, ex); LOG.error("Delete bucket failed for bucket:{} in volume:{}", bucketName,
volumeName, ex);
}
throw ex; throw ex;
} finally { } finally {
metadataManager.writeLock().unlock(); metadataManager.writeLock().unlock();

25
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeyManagerImpl.java
View File

@ -99,18 +99,18 @@ public class KeyManagerImpl implements KeyManager {
try { try {
byte[] volumeKey = metadataManager.getVolumeKey(volumeName); byte[] volumeKey = metadataManager.getVolumeKey(volumeName);
byte[] bucketKey = metadataManager.getBucketKey(volumeName, bucketName); byte[] bucketKey = metadataManager.getBucketKey(volumeName, bucketName);
byte[] keyKey = metadataManager.getDBKeyForKey( byte[] keyKey =
volumeName, bucketName, keyName); metadataManager.getDBKeyForKey(volumeName, bucketName, keyName);
//Check if the volume exists //Check if the volume exists
if(metadataManager.get(volumeKey) == null) { if (metadataManager.get(volumeKey) == null) {
LOG.error("volume not found: {}", volumeName); LOG.debug("volume not found: {}", volumeName);
throw new KSMException("Volume not found", throw new KSMException("Volume not found",
KSMException.ResultCodes.FAILED_VOLUME_NOT_FOUND); KSMException.ResultCodes.FAILED_VOLUME_NOT_FOUND);
} }
//Check if bucket already exists //Check if bucket already exists
if(metadataManager.get(bucketKey) == null) { if (metadataManager.get(bucketKey) == null) {
LOG.error("bucket not found: {}/{} ", volumeName, bucketName); LOG.debug("bucket not found: {}/{} ", volumeName, bucketName);
throw new KSMException("Bucket not found", throw new KSMException("Bucket not found",
KSMException.ResultCodes.FAILED_BUCKET_NOT_FOUND); KSMException.ResultCodes.FAILED_BUCKET_NOT_FOUND);
} }
@ -123,7 +123,6 @@ public class KeyManagerImpl implements KeyManager {
// clean up those blocks when it can. Right now making this change // clean up those blocks when it can. Right now making this change
// allows us to pass tests that expect ozone can overwrite a key. // allows us to pass tests that expect ozone can overwrite a key.
// When we talk to SCM make sure that we ask for at least a byte in the // When we talk to SCM make sure that we ask for at least a byte in the
// block. This way even if the call is for a zero length key, we back it // block. This way even if the call is for a zero length key, we back it
// with a actual SCM block. // with a actual SCM block.
@ -164,14 +163,16 @@ public class KeyManagerImpl implements KeyManager {
.setModificationTime(currentTime) .setModificationTime(currentTime)
.build(); .build();
metadataManager.put(keyKey, keyBlock.getProtobuf().toByteArray()); metadataManager.put(keyKey, keyBlock.getProtobuf().toByteArray());
LOG.debug("Key {} allocated in volume {} bucket {}", LOG.debug("Key {} allocated in volume {} bucket {}", keyName, volumeName,
keyName, volumeName, bucketName); bucketName);
return keyBlock; return keyBlock;
} catch (KSMException e) { } catch (KSMException e) {
throw e; throw e;
} catch (IOException ex) { } catch (IOException ex) {
LOG.error("Key allocation failed for volume:{} bucket:{} key:{}", if (!(ex instanceof KSMException)) {
volumeName, bucketName, keyName, ex); LOG.error("Key allocation failed for volume:{} bucket:{} key:{}",
volumeName, bucketName, keyName, ex);
}
throw new KSMException(ex.getMessage(), throw new KSMException(ex.getMessage(),
KSMException.ResultCodes.FAILED_KEY_ALLOCATION); KSMException.ResultCodes.FAILED_KEY_ALLOCATION);
} finally { } finally {
@ -191,7 +192,7 @@ public class KeyManagerImpl implements KeyManager {
volumeName, bucketName, keyName); volumeName, bucketName, keyName);
byte[] value = metadataManager.get(keyKey); byte[] value = metadataManager.get(keyKey);
if (value == null) { if (value == null) {
LOG.error("Key: {} not found", keyKey); LOG.debug("Key: {} not found", keyKey);
throw new KSMException("Key not found", throw new KSMException("Key not found",
KSMException.ResultCodes.FAILED_KEY_NOT_FOUND); KSMException.ResultCodes.FAILED_KEY_NOT_FOUND);
} }

58
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/VolumeManagerImpl.java
View File

@ -75,7 +75,7 @@ public class VolumeManagerImpl implements VolumeManager {
// Check the volume count // Check the volume count
if (prevVolList.size() >= maxUserVolumeCount) { if (prevVolList.size() >= maxUserVolumeCount) {
LOG.error("Too many volumes for user:{}", owner); LOG.debug("Too many volumes for user:{}", owner);
throw new KSMException(ResultCodes.FAILED_TOO_MANY_USER_VOLUMES); throw new KSMException(ResultCodes.FAILED_TOO_MANY_USER_VOLUMES);
} }
@ -97,6 +97,7 @@ public class VolumeManagerImpl implements VolumeManager {
VolumeList vlist = VolumeList.parseFrom(volumeList); VolumeList vlist = VolumeList.parseFrom(volumeList);
prevVolList.addAll(vlist.getVolumeNamesList()); prevVolList.addAll(vlist.getVolumeNamesList());
} else { } else {
LOG.debug("volume:{} not found for user:{}");
throw new KSMException(ResultCodes.FAILED_USER_NOT_FOUND); throw new KSMException(ResultCodes.FAILED_USER_NOT_FOUND);
} }
@ -125,7 +126,7 @@ public class VolumeManagerImpl implements VolumeManager {
// Check of the volume already exists // Check of the volume already exists
if (volumeInfo != null) { if (volumeInfo != null) {
LOG.error("volume:{} already exists", args.getVolume()); LOG.debug("volume:{} already exists", args.getVolume());
throw new KSMException(ResultCodes.FAILED_VOLUME_ALREADY_EXISTS); throw new KSMException(ResultCodes.FAILED_VOLUME_ALREADY_EXISTS);
} }
@ -137,11 +138,13 @@ public class VolumeManagerImpl implements VolumeManager {
// Add volume to user list // Add volume to user list
addVolumeToOwnerList(args.getVolume(), args.getOwnerName(), batch); addVolumeToOwnerList(args.getVolume(), args.getOwnerName(), batch);
metadataManager.writeBatch(batch); metadataManager.writeBatch(batch);
LOG.info("created volume:{} user:{}", LOG.debug("created volume:{} user:{}", args.getVolume(),
args.getVolume(), args.getOwnerName()); args.getOwnerName());
} catch (IOException ex) { } catch (IOException ex) {
LOG.error("Volume creation failed for user:{} volname:{}", if (!(ex instanceof KSMException)) {
args.getOwnerName(), args.getVolume(), ex); LOG.error("Volume creation failed for user:{} volume:{}",
args.getOwnerName(), args.getVolume(), ex);
}
throw ex; throw ex;
} finally { } finally {
metadataManager.writeLock().unlock(); metadataManager.writeLock().unlock();
@ -164,6 +167,8 @@ public class VolumeManagerImpl implements VolumeManager {
byte[] dbVolumeKey = metadataManager.getVolumeKey(volume); byte[] dbVolumeKey = metadataManager.getVolumeKey(volume);
byte[] volInfo = metadataManager.get(dbVolumeKey); byte[] volInfo = metadataManager.get(dbVolumeKey);
if (volInfo == null) { if (volInfo == null) {
LOG.debug("Changing volume ownership failed for user:{} volume:{}",
owner, volume);
throw new KSMException(ResultCodes.FAILED_VOLUME_NOT_FOUND); throw new KSMException(ResultCodes.FAILED_VOLUME_NOT_FOUND);
} }
@ -188,8 +193,10 @@ public class VolumeManagerImpl implements VolumeManager {
metadataManager.writeBatch(batch); metadataManager.writeBatch(batch);
} catch (IOException ex) { } catch (IOException ex) {
LOG.error("Changing volume ownership failed for user:{} volume:{}", if (!(ex instanceof KSMException)) {
owner, volume, ex); LOG.error("Changing volume ownership failed for user:{} volume:{}",
owner, volume, ex);
}
throw ex; throw ex;
} finally { } finally {
metadataManager.writeLock().unlock(); metadataManager.writeLock().unlock();
@ -210,7 +217,8 @@ public class VolumeManagerImpl implements VolumeManager {
byte[] dbVolumeKey = metadataManager.getVolumeKey(volume); byte[] dbVolumeKey = metadataManager.getVolumeKey(volume);
byte[] volInfo = metadataManager.get(dbVolumeKey); byte[] volInfo = metadataManager.get(dbVolumeKey);
if (volInfo == null) { if (volInfo == null) {
throw new KSMException(ResultCodes.FAILED_VOLUME_NOT_FOUND); LOG.debug("volume:{} does not exist", volume);
throw new KSMException(ResultCodes.FAILED_VOLUME_NOT_FOUND);
} }
VolumeInfo volumeInfo = VolumeInfo.parseFrom(volInfo); VolumeInfo volumeInfo = VolumeInfo.parseFrom(volInfo);
@ -218,18 +226,20 @@ public class VolumeManagerImpl implements VolumeManager {
Preconditions.checkState(volume.equals(volumeInfo.getVolume())); Preconditions.checkState(volume.equals(volumeInfo.getVolume()));
KsmVolumeArgs newVolumeArgs = KsmVolumeArgs newVolumeArgs =
KsmVolumeArgs.newBuilder().setVolume(volumeArgs.getVolume()) KsmVolumeArgs.newBuilder()
.setVolume(volumeArgs.getVolume())
.setAdminName(volumeArgs.getAdminName()) .setAdminName(volumeArgs.getAdminName())
.setOwnerName(volumeArgs.getOwnerName()) .setOwnerName(volumeArgs.getOwnerName())
.setQuotaInBytes(quota) .setQuotaInBytes(quota)
.setCreationTime(volumeArgs.getCreationTime()) .setCreationTime(volumeArgs.getCreationTime()).build();
.build();
VolumeInfo newVolumeInfo = newVolumeArgs.getProtobuf(); VolumeInfo newVolumeInfo = newVolumeArgs.getProtobuf();
metadataManager.put(dbVolumeKey, newVolumeInfo.toByteArray()); metadataManager.put(dbVolumeKey, newVolumeInfo.toByteArray());
} catch (IOException ex) { } catch (IOException ex) {
LOG.error("Changing volume quota failed for volume:{} quota:{}", if (!(ex instanceof KSMException)) {
volume, quota, ex); LOG.error("Changing volume quota failed for volume:{} quota:{}", volume,
quota, ex);
}
throw ex; throw ex;
} finally { } finally {
metadataManager.writeLock().unlock(); metadataManager.writeLock().unlock();
@ -249,7 +259,8 @@ public class VolumeManagerImpl implements VolumeManager {
byte[] dbVolumeKey = metadataManager.getVolumeKey(volume); byte[] dbVolumeKey = metadataManager.getVolumeKey(volume);
byte[] volInfo = metadataManager.get(dbVolumeKey); byte[] volInfo = metadataManager.get(dbVolumeKey);
if (volInfo == null) { if (volInfo == null) {
throw new KSMException(ResultCodes.FAILED_VOLUME_NOT_FOUND); LOG.debug("volume:{} does not exist", volume);
throw new KSMException(ResultCodes.FAILED_VOLUME_NOT_FOUND);
} }
VolumeInfo volumeInfo = VolumeInfo.parseFrom(volInfo); VolumeInfo volumeInfo = VolumeInfo.parseFrom(volInfo);
@ -257,7 +268,9 @@ public class VolumeManagerImpl implements VolumeManager {
Preconditions.checkState(volume.equals(volumeInfo.getVolume())); Preconditions.checkState(volume.equals(volumeInfo.getVolume()));
return volumeArgs; return volumeArgs;
} catch (IOException ex) { } catch (IOException ex) {
LOG.error("Info volume failed for volume:{}", volume, ex); if (!(ex instanceof KSMException)) {
LOG.warn("Info volume failed for volume:{}", volume, ex);
}
throw ex; throw ex;
} finally { } finally {
metadataManager.readLock().unlock(); metadataManager.readLock().unlock();
@ -279,10 +292,12 @@ public class VolumeManagerImpl implements VolumeManager {
byte[] dbVolumeKey = metadataManager.getVolumeKey(volume); byte[] dbVolumeKey = metadataManager.getVolumeKey(volume);
byte[] volInfo = metadataManager.get(dbVolumeKey); byte[] volInfo = metadataManager.get(dbVolumeKey);
if (volInfo == null) { if (volInfo == null) {
LOG.debug("volume:{} does not exist", volume);
throw new KSMException(ResultCodes.FAILED_VOLUME_NOT_FOUND); throw new KSMException(ResultCodes.FAILED_VOLUME_NOT_FOUND);
} }
if (!metadataManager.isVolumeEmpty(volume)) { if (!metadataManager.isVolumeEmpty(volume)) {
LOG.debug("volume:{} is not empty", volume);
throw new KSMException(ResultCodes.FAILED_VOLUME_NOT_EMPTY); throw new KSMException(ResultCodes.FAILED_VOLUME_NOT_EMPTY);
} }
@ -294,7 +309,9 @@ public class VolumeManagerImpl implements VolumeManager {
batch.delete(dbVolumeKey); batch.delete(dbVolumeKey);
metadataManager.writeBatch(batch); metadataManager.writeBatch(batch);
} catch (IOException ex) { } catch (IOException ex) {
LOG.error("Delete volume failed for volume:{}", volume, ex); if (!(ex instanceof KSMException)) {
LOG.error("Delete volume failed for volume:{}", volume, ex);
}
throw ex; throw ex;
} finally { } finally {
metadataManager.writeLock().unlock(); metadataManager.writeLock().unlock();
@ -318,6 +335,7 @@ public class VolumeManagerImpl implements VolumeManager {
byte[] dbVolumeKey = metadataManager.getVolumeKey(volume); byte[] dbVolumeKey = metadataManager.getVolumeKey(volume);
byte[] volInfo = metadataManager.get(dbVolumeKey); byte[] volInfo = metadataManager.get(dbVolumeKey);
if (volInfo == null) { if (volInfo == null) {
LOG.debug("volume:{} does not exist", volume);
throw new KSMException(ResultCodes.FAILED_VOLUME_NOT_FOUND); throw new KSMException(ResultCodes.FAILED_VOLUME_NOT_FOUND);
} }
@ -326,8 +344,10 @@ public class VolumeManagerImpl implements VolumeManager {
Preconditions.checkState(volume.equals(volumeInfo.getVolume())); Preconditions.checkState(volume.equals(volumeInfo.getVolume()));
return volumeArgs.getAclMap().hasAccess(userAcl); return volumeArgs.getAclMap().hasAccess(userAcl);
} catch (IOException ex) { } catch (IOException ex) {
LOG.error("Check volume access failed for volume:{} user:{} rights:{}", if (!(ex instanceof KSMException)) {
volume, userAcl.getName(), userAcl.getRights(), ex); LOG.error("Check volume access failed for volume:{} user:{} rights:{}",
volume, userAcl.getName(), userAcl.getRights(), ex);
}
throw ex; throw ex;
} finally { } finally {
metadataManager.readLock().unlock(); metadataManager.readLock().unlock();