Addressed review comments

Author: hemantk-12

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotPurgeRequest.java

@@ -21,6 +21,7 @@
 
 import org.apache.commons.lang3.tuple.Triple;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
 import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
@@ -110,18 +111,18 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
         // snapshotTableKey is nothing but /volumeName/bucketName/snapshotName.
         // Once all the locks are acquired, it performs the three steps mentioned above and
         // release all the locks after that.
-        Set<String> lockSet = new HashSet<>();
+        Set<Triple<String, String, String>> lockSet = new HashSet<>(4);
         try {
-          acquireLock(lockSet, snapTableKey, omMetadataManager);
-          SnapshotInfo fromSnapshot = omMetadataManager.getSnapshotInfoTable().get(snapTableKey);
-
-          if (fromSnapshot == null) {
+          if (omMetadataManager.getSnapshotInfoTable().get(snapTableKey) == null) {
             // Snapshot may have been purged in the previous iteration of SnapshotDeletingService.
             LOG.warn("The snapshot {} is not longer in snapshot table, It maybe removed in the previous " +
                 "Snapshot purge request.", snapTableKey);
             continue;
           }
 
+          acquireLock(lockSet, snapTableKey, omMetadataManager);
+          SnapshotInfo fromSnapshot = omMetadataManager.getSnapshotInfoTable().get(snapTableKey);
+
           SnapshotInfo nextSnapshot =
               SnapshotUtils.getNextActiveSnapshot(fromSnapshot, snapshotChainManager, omSnapshotManager);
 
@@ -139,8 +140,9 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
           omMetadataManager.getSnapshotInfoTable()
               .addCacheEntry(new CacheKey<>(fromSnapshot.getTableKey()), CacheValue.get(trxnLogIndex));
         } finally {
-          for (String lockString: lockSet) {
-            releaseLock(lockString, omMetadataManager);
+          for (Triple<String, String, String> lockKey: lockSet) {
+            omMetadataManager.getLock()
+                .releaseWriteLock(SNAPSHOT_LOCK, lockKey.getLeft(), lockKey.getMiddle(), lockKey.getRight());
           }
         }
       }
@@ -156,23 +158,25 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
     return omClientResponse;
   }
 
-  private void acquireLock(Set<String> lockSet, String snapshotTableKey,
+  private void acquireLock(Set<Triple<String, String, String>> lockSet, String snapshotTableKey,
                            OMMetadataManager omMetadataManager) throws IOException {
-    if (!lockSet.contains(snapshotTableKey)) {
-      Triple<String, String, String> triple = SnapshotUtils.getSnapshotTableKeyFromString(snapshotTableKey);
+    SnapshotInfo snapshotInfo = omMetadataManager.getSnapshotInfoTable().get(snapshotTableKey);
+
+    // It should not be the case that lock is required for non-existing snapshot.
+    if (snapshotInfo == null) {
+      LOG.error("Snapshot: '{}' doesn't not exist in snapshot table.", snapshotTableKey);
+      throw new OMException("Snapshot: '{" + snapshotTableKey + "}' doesn't not exist in snapshot table.",
+          OMException.ResultCodes.FILE_NOT_FOUND);
+    }
+    Triple<String, String, String> lockKey = Triple.of(snapshotInfo.getVolumeName(), snapshotInfo.getBucketName(),
+        snapshotInfo.getName());
+    if (!lockSet.contains(lockKey)) {
       mergeOmLockDetails(omMetadataManager.getLock()
-          .acquireWriteLock(SNAPSHOT_LOCK, triple.getLeft(), triple.getMiddle(), triple.getRight()));
-      lockSet.add(snapshotTableKey);
+          .acquireWriteLock(SNAPSHOT_LOCK, lockKey.getLeft(), lockKey.getMiddle(), lockKey.getRight()));
+      lockSet.add(lockKey);
     }
   }
 
-  private void releaseLock(String snapshotTableKey,
-                           OMMetadataManager omMetadataManager) throws IOException {
-    Triple<String, String, String> triple = SnapshotUtils.getSnapshotTableKeyFromString(snapshotTableKey);
-    omMetadataManager.getLock()
-        .releaseWriteLock(SNAPSHOT_LOCK, triple.getLeft(), triple.getMiddle(), triple.getRight());
-  }
-
   private void updateSnapshotInfoAndCache(SnapshotInfo snapInfo,
       OmMetadataManagerImpl omMetadataManager, long trxnLogIndex,
       Map<String, SnapshotInfo> updatedSnapInfos) throws IOException {
@@ -199,7 +203,7 @@ private void updateSnapshotInfoAndCache(SnapshotInfo snapInfo,
    * update in DB.
    */
   private void updateSnapshotChainAndCache(
-      Set<String> lockSet,
+      Set<Triple<String, String, String>> lockSet,
       OmMetadataManagerImpl metadataManager,
       SnapshotInfo snapInfo,
       long trxnLogIndex,
@@ -238,20 +242,20 @@ private void updateSnapshotChainAndCache(
       acquireLock(lockSet, nextPathSnapshotKey, metadataManager);
     }
 
-    String nestGlobalSnapshotKey = null;
+    String nextGlobalSnapshotKey = null;
     if (hasNextGlobalSnapshot) {
       UUID nextGlobalSnapshotId = snapshotChainManager.nextGlobalSnapshot(snapInfo.getSnapshotId());
-      nestGlobalSnapshotKey = snapshotChainManager.getTableKey(nextGlobalSnapshotId);
+      nextGlobalSnapshotKey = snapshotChainManager.getTableKey(nextGlobalSnapshotId);
 
       // Acquire lock from the snapshot
-      acquireLock(lockSet, nestGlobalSnapshotKey, metadataManager);
+      acquireLock(lockSet, nextGlobalSnapshotKey, metadataManager);
     }
 
     SnapshotInfo nextPathSnapInfo =
         nextPathSnapshotKey != null ? metadataManager.getSnapshotInfoTable().get(nextPathSnapshotKey) : null;
 
     SnapshotInfo nextGlobalSnapInfo =
-        nestGlobalSnapshotKey != null ? metadataManager.getSnapshotInfoTable().get(nestGlobalSnapshotKey) : null;
+        nextGlobalSnapshotKey != null ? metadataManager.getSnapshotInfoTable().get(nextGlobalSnapshotKey) : null;
 
     // Updates next path snapshot's previous snapshot ID
     if (nextPathSnapInfo != null) {

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotSetPropertyRequest.java

@@ -17,8 +17,6 @@
  */
 package org.apache.hadoop.ozone.om.request.snapshot;
 
-import org.apache.commons.lang3.tuple.Triple;
-import org.apache.hadoop.ozone.om.snapshot.SnapshotUtils;
 import org.apache.ratis.server.protocol.TermIndex;
 import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
 import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
@@ -38,7 +36,7 @@
 
 import java.io.IOException;
 
-import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.INVALID_SNAPSHOT_ERROR;
+import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.FILE_NOT_FOUND;
 import static org.apache.hadoop.ozone.om.lock.OzoneManagerLock.Resource.SNAPSHOT_LOCK;
 
 /**
@@ -72,10 +70,15 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
     String snapshotName = null;
 
     try {
-      Triple<String, String, String> triple = SnapshotUtils.getSnapshotTableKeyFromString(snapshotKey);
-      volumeName = triple.getLeft();
-      bucketName = triple.getMiddle();
-      snapshotName = triple.getRight();
+      SnapshotInfo snapshotInfo = metadataManager.getSnapshotInfoTable().get(snapshotKey);
+      if (snapshotInfo == null) {
+        LOG.error("Snapshot: '{}' doesn't not exist in snapshot table.", snapshotKey);
+        throw new OMException("Snapshot: '{" + snapshotKey + "}' doesn't not exist in snapshot table.", FILE_NOT_FOUND);
+      }
+
+      volumeName = snapshotInfo.getVolumeName();
+      bucketName = snapshotInfo.getBucketName();
+      snapshotName = snapshotInfo.getName();
 
       mergeOmLockDetails(metadataManager.getLock()
           .acquireWriteLock(SNAPSHOT_LOCK, volumeName, bucketName, snapshotName));
@@ -85,11 +88,6 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
       updatedSnapInfo = metadataManager.getSnapshotInfoTable()
           .get(snapshotKey);
 
-      if (updatedSnapInfo == null) {
-        LOG.error("SnapshotInfo for Snapshot: {} is not found", snapshotKey);
-        throw new OMException("SnapshotInfo for Snapshot: " + snapshotKey +
-            " is not found", INVALID_SNAPSHOT_ERROR);
-      }
 
       if (setSnapshotPropertyRequest.hasDeepCleanedDeletedDir()) {
         updatedSnapInfo.setDeepCleanedDeletedDir(setSnapshotPropertyRequest

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

@@ -18,7 +18,6 @@
 
 package org.apache.hadoop.ozone.om.snapshot;
 
-import org.apache.commons.lang3.tuple.Triple;
 import org.apache.hadoop.hdds.utils.db.managed.ManagedRocksDB;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.OmMetadataManagerImpl;
@@ -245,19 +244,4 @@ public static String getOzonePathKeyForFso(OMMetadataManager metadataManager,
     final long bucketId = metadataManager.getBucketId(volumeName, bucketName);
     return OM_KEY_PREFIX + volumeId + OM_KEY_PREFIX + bucketId + OM_KEY_PREFIX;
   }
-
-  /**
-   * Helper function which return a Triple of volumeName, bucketName and snapshotName from snapshotTableKey String.
-   */
-  public static Triple<String, String, String> getSnapshotTableKeyFromString(String snapshotKey) throws IOException {
-    if (snapshotKey == null) {
-      throw new IOException("Snapshot table key is null.");
-    }
-
-    String[] split = snapshotKey.split(OM_KEY_PREFIX);
-    if (split.length != 4) {
-      throw new IOException("Invalid snapshot table key: " + snapshotKey);
-    }
-    return Triple.of(split[1], split[2], split[3]);
-  }
 }