HDDS-10524. [Snapshot] Invalidate the cache entry from snapshotInfoTable cache in OMSnapshotPurgeRequest (#6443)

Author: hemantk-12

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/SnapshotInfo.java

@@ -718,7 +718,7 @@ public SnapshotInfo copyObject() {
   @Override
   public String toString() {
     return "SnapshotInfo{" +
-        ", snapshotId: '" + snapshotId + '\'' +
+        "snapshotId: '" + snapshotId + '\'' +
         ", name: '" + name + "'," +
         ", volumeName: '" + volumeName + '\'' +
         ", bucketName: '" + bucketName + '\'' +

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

@@ -102,7 +102,11 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
             trxnLogIndex, updatedSnapInfos);
         updateSnapshotChainAndCache(omMetadataManager, fromSnapshot,
             trxnLogIndex, updatedPathPreviousAndGlobalSnapshots);
+        // Remove and close snapshot's RocksDB instance from SnapshotCache.
         ozoneManager.getOmSnapshotManager().invalidateCacheEntry(fromSnapshot.getSnapshotId());
+        // Update SnapshotInfoTable cache.
+        ozoneManager.getMetadataManager().getSnapshotInfoTable()
+            .addCacheEntry(new CacheKey<>(fromSnapshot.getTableKey()), CacheValue.get(trxnLogIndex));
       }
 
       omClientResponse = new OMSnapshotPurgeResponse(omResponse.build(),

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

@@ -84,8 +84,9 @@ protected void addToDBBatch(OMMetadataManager omMetadataManager,
     updateSnapInfo(metadataManager, batchOperation,
         updatedPreviousAndGlobalSnapInfos);
     for (String dbKey: snapshotDbKeys) {
+      // Skip the cache here because snapshot is purged from cache in OMSnapshotPurgeRequest.
       SnapshotInfo snapshotInfo = omMetadataManager
-          .getSnapshotInfoTable().get(dbKey);
+          .getSnapshotInfoTable().getSkipCache(dbKey);
       // Even though snapshot existed when SnapshotDeletingService
       // was running. It might be deleted in the previous run and
       // the DB might not have been updated yet. So snapshotInfo
@@ -96,8 +97,7 @@ protected void addToDBBatch(OMMetadataManager omMetadataManager,
 
       // Delete Snapshot checkpoint directory.
       deleteCheckpointDirectory(omMetadataManager, snapshotInfo);
-      omMetadataManager.getSnapshotInfoTable().deleteWithBatch(batchOperation,
-          dbKey);
+      omMetadataManager.getSnapshotInfoTable().deleteWithBatch(batchOperation, dbKey);
     }
   }
 

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/snapshot/TestOMSnapshotPurgeRequestAndResponse.java

@@ -67,6 +67,7 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.anyString;
@@ -77,8 +78,6 @@
  * Tests OMSnapshotPurgeRequest class.
  */
 public class TestOMSnapshotPurgeRequestAndResponse {
-
-  private BatchOperation batchOperation;
   private List<Path> checkpointPaths = new ArrayList<>();
 
   private OzoneManager ozoneManager;
@@ -177,7 +176,6 @@ private void createSnapshotCheckpoint(String snapshotName) throws Exception {
   private void createSnapshotCheckpoint(String volume,
                                         String bucket,
                                         String snapshotName) throws Exception {
-    batchOperation = omMetadataManager.getStore().initBatchOperation();
     OMRequest omRequest = OMRequestTestUtils
         .createSnapshotRequest(volume, bucket, snapshotName);
     // Pre-Execute OMSnapshotCreateRequest.
@@ -188,9 +186,10 @@ private void createSnapshotCheckpoint(String volume,
     OMSnapshotCreateResponse omClientResponse = (OMSnapshotCreateResponse)
         omSnapshotCreateRequest.validateAndUpdateCache(ozoneManager, 1);
     // Add to batch and commit to DB.
-    omClientResponse.addToDBBatch(omMetadataManager, batchOperation);
-    omMetadataManager.getStore().commitBatchOperation(batchOperation);
-    batchOperation.close();
+    try (BatchOperation batchOperation = omMetadataManager.getStore().initBatchOperation()) {
+      omClientResponse.addToDBBatch(omMetadataManager, batchOperation);
+      omMetadataManager.getStore().commitBatchOperation(batchOperation);
+    }
 
     String key = SnapshotInfo.getTableKey(volume, bucket, snapshotName);
     SnapshotInfo snapshotInfo =
@@ -226,9 +225,10 @@ private void purgeSnapshots(OMRequest snapshotPurgeRequest)
         omSnapshotPurgeRequest.validateAndUpdateCache(ozoneManager, 200L);
 
     // Commit to DB.
-    batchOperation = omMetadataManager.getStore().initBatchOperation();
-    omSnapshotPurgeResponse.checkAndUpdateDB(omMetadataManager, batchOperation);
-    omMetadataManager.getStore().commitBatchOperation(batchOperation);
+    try (BatchOperation batchOperation = omMetadataManager.getStore().initBatchOperation()) {
+      omSnapshotPurgeResponse.checkAndUpdateDB(omMetadataManager, batchOperation);
+      omMetadataManager.getStore().commitBatchOperation(batchOperation);
+    }
   }
 
   @Test
@@ -238,7 +238,20 @@ public void testValidateAndUpdateCache() throws Exception {
     assertFalse(omMetadataManager.getSnapshotInfoTable().isEmpty());
     OMRequest snapshotPurgeRequest = createPurgeKeysRequest(
         snapshotDbKeysToPurge);
-    purgeSnapshots(snapshotPurgeRequest);
+
+    OMSnapshotPurgeRequest omSnapshotPurgeRequest = preExecute(snapshotPurgeRequest);
+
+    OMSnapshotPurgeResponse omSnapshotPurgeResponse = (OMSnapshotPurgeResponse)
+        omSnapshotPurgeRequest.validateAndUpdateCache(ozoneManager, 200L);
+
+    for (String snapshotTableKey: snapshotDbKeysToPurge) {
+      assertNull(omMetadataManager.getSnapshotInfoTable().get(snapshotTableKey));
+    }
+
+    try (BatchOperation batchOperation = omMetadataManager.getStore().initBatchOperation()) {
+      omSnapshotPurgeResponse.checkAndUpdateDB(omMetadataManager, batchOperation);
+      omMetadataManager.getStore().commitBatchOperation(batchOperation);
+    }
 
     // Check if the entries are deleted.
     assertTrue(omMetadataManager.getSnapshotInfoTable().isEmpty());