HDDS-7565. FSO purge directory for old bucket can update quota for new bucket

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

@@ -75,12 +75,13 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
                 volumeName, bucketName);
             lockSet.add(volBucketPair);
           }
-          OmBucketInfo omBucketInfo = getBucketInfo(omMetadataManager,
-              volumeName, bucketName);
+          updateBucketInfo(volBucketInfoMap, omMetadataManager, path,
+              volumeName, bucketName, volBucketPair);
+          OmBucketInfo omBucketInfo = volBucketInfoMap.get(volBucketPair);
           // bucketInfo can be null in case of delete volume or bucket
+          // or key does not belong to bucket as bucket is recreated
           if (null != omBucketInfo) {
             omBucketInfo.incrUsedNamespace(-1L);
-            volBucketInfoMap.putIfAbsent(volBucketPair, omBucketInfo);
           }
         }
 
@@ -95,13 +96,14 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
                 volumeName, bucketName);
             lockSet.add(volBucketPair);
           }
-          OmBucketInfo omBucketInfo = getBucketInfo(omMetadataManager,
-              volumeName, bucketName);
+          updateBucketInfo(volBucketInfoMap, omMetadataManager,
+              path, volumeName, bucketName, volBucketPair);
+          OmBucketInfo omBucketInfo = volBucketInfoMap.get(volBucketPair);
           // bucketInfo can be null in case of delete volume or bucket
+          // or key does not belong to bucket as bucket is recreated
           if (null != omBucketInfo) {
             omBucketInfo.incrUsedBytes(-sumBlockLengths(keyInfo));
             omBucketInfo.incrUsedNamespace(-1L);
-            volBucketInfoMap.putIfAbsent(volBucketPair, omBucketInfo);
           }
         }
       }
@@ -116,7 +118,9 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
               e.getValue()));
       for (Map.Entry<Pair<String, String>, OmBucketInfo> entry :
           volBucketInfoMap.entrySet()) {
-        entry.setValue(entry.getValue().copyObject());
+        if (null != entry.getValue()) {
+          entry.setValue(entry.getValue().copyObject());
+        }
       }
     }
 
@@ -130,4 +134,23 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
 
     return omClientResponse;
   }
+
+  private void updateBucketInfo(
+      Map<Pair<String, String>, OmBucketInfo> volBucketInfoMap,
+      OMMetadataManager omMetadataManager,
+      OzoneManagerProtocolProtos.PurgePathRequest path, String volumeName,
+      String bucketName, Pair<String, String> volBucketPair) {
+    if (!volBucketInfoMap.containsKey(volBucketPair)) {
+      OmBucketInfo omBucketInfo = getBucketInfo(omMetadataManager,
+          volumeName, bucketName);
+      // null is added to avoid again lookup, in case of bucket not found
+      // or bucket is not matching as recreated
+      if (null == omBucketInfo 
+          || omBucketInfo.getObjectID() != path.getBucketId()) {
+        volBucketInfoMap.put(volBucketPair, null);
+      } else {
+        volBucketInfoMap.put(volBucketPair, omBucketInfo);
+      }
+    }
+  }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMDirectoriesPurgeResponseWithFSO.java

@@ -133,9 +133,11 @@ public void addToDBBatch(OMMetadataManager omMetadataManager,
 
       // update bucket usedBytes.
       for (OmBucketInfo omBucketInfo : volBucketInfoMap.values()) {
-        omMetadataManager.getBucketTable().putWithBatch(batchOperation,
-            omMetadataManager.getBucketKey(omBucketInfo.getVolumeName(),
-                omBucketInfo.getBucketName()), omBucketInfo);
+        if (null != omBucketInfo) {
+          omMetadataManager.getBucketTable().putWithBatch(batchOperation,
+              omMetadataManager.getBucketKey(omBucketInfo.getVolumeName(),
+                  omBucketInfo.getBucketName()), omBucketInfo);
+        }
       }
     }
   }

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMDirectoriesPurgeRequestAndResponse.java

@@ -39,6 +39,7 @@
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
 import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
 
 /**
@@ -52,6 +53,13 @@ public class TestOMDirectoriesPurgeRequestAndResponse extends TestOMKeyRequest {
    * Creates volume, bucket and key entries and adds to OM DB and then
    * deletes these keys to move them to deletedKeys table.
    */
+  @Before
+  public void cleanup() throws Exception {
+    String bucketKey = omMetadataManager.getBucketKey(volumeName,
+        bucketName);
+    omMetadataManager.getBucketTable().delete(bucketKey);
+  }
+
   private List<OmKeyInfo> createAndDeleteKeys(Integer trxnIndex, String bucket)
       throws Exception {
     if (bucket == null) {
@@ -108,7 +116,7 @@ private void updateBlockInfo(OmKeyInfo omKeyInfo) throws IOException {
    * @return OMRequest
    */
   private OMRequest createPurgeKeysRequest(String purgeDeletedDir,
-      List<OmKeyInfo> keyList) throws IOException {
+      List<OmKeyInfo> keyList, OmBucketInfo bucketInfo) throws IOException {
     List<OzoneManagerProtocolProtos.PurgePathRequest> purgePathRequestList
         = new ArrayList<>();
     List<OmKeyInfo> subFiles = new ArrayList<>();
@@ -117,7 +125,7 @@ private OMRequest createPurgeKeysRequest(String purgeDeletedDir,
     }
     List<OmKeyInfo> subDirs = new ArrayList<>();
     Long volumeId = 1L;
-    Long bucketId = 1L;
+    Long bucketId = bucketInfo.getObjectID();
     OzoneManagerProtocolProtos.PurgePathRequest request = wrapPurgeRequest(
         volumeId, bucketId, purgeDeletedDir, subFiles, subDirs);
     purgePathRequestList.add(request);
@@ -190,15 +198,15 @@ public void testValidateAndUpdateCacheCheckQuota() throws Exception {
     }
 
     // Create PurgeKeysRequest to purge the deleted keys
-    OMRequest omRequest = createPurgeKeysRequest(null, deletedKeyInfos);
-
+    String bucketKey = omMetadataManager.getBucketKey(volumeName, bucketName);
+    OmBucketInfo omBucketInfo = omMetadataManager.getBucketTable().get(
+        bucketKey);
+    OMRequest omRequest = createPurgeKeysRequest(
+        null, deletedKeyInfos, omBucketInfo);
     OMRequest preExecutedRequest = preExecute(omRequest);
     OMDirectoriesPurgeRequestWithFSO omKeyPurgeRequest =
         new OMDirectoriesPurgeRequestWithFSO(preExecutedRequest);
 
-    String bucketKey = omMetadataManager.getBucketKey(volumeName, bucketName);
-    OmBucketInfo omBucketInfo = omMetadataManager.getBucketTable().get(
-        bucketKey);
     Assert.assertEquals(1000L * deletedKeyNames.size(),
         omBucketInfo.getUsedBytes());
     OMDirectoriesPurgeResponseWithFSO omClientResponse
@@ -225,4 +233,73 @@ public void testValidateAndUpdateCacheCheckQuota() throws Exception {
           deletedKey));
     }
   }
+
+  @Test
+  public void testValidateAndUpdateCacheQuotaBucketRecreated()
+      throws Exception {
+    // Create and Delete keys. The keys should be moved to DeletedKeys table
+    List<OmKeyInfo> deletedKeyInfos = createAndDeleteKeys(1, null);
+    // The keys should be present in the DeletedKeys table before purging
+    List<String> deletedKeyNames = new ArrayList<>();
+    for (OmKeyInfo deletedKey : deletedKeyInfos) {
+      String keyName = omMetadataManager.getOzoneKey(deletedKey.getVolumeName(),
+          deletedKey.getBucketName(), deletedKey.getKeyName());
+      Assert.assertTrue(omMetadataManager.getDeletedTable().isExist(
+          keyName));
+      deletedKeyNames.add(keyName);
+    }
+
+
+    // Create PurgeKeysRequest to purge the deleted keys
+    String bucketKey = omMetadataManager.getBucketKey(volumeName, bucketName);
+    OmBucketInfo omBucketInfo = omMetadataManager.getBucketTable().get(
+        bucketKey);
+    OMRequest omRequest = createPurgeKeysRequest(
+        null, deletedKeyInfos, omBucketInfo);
+    OMRequest preExecutedRequest = preExecute(omRequest);
+    OMDirectoriesPurgeRequestWithFSO omKeyPurgeRequest =
+        new OMDirectoriesPurgeRequestWithFSO(preExecutedRequest);
+
+    // recreate bucket
+    omMetadataManager.getBucketTable().delete(bucketKey);
+    OMRequestTestUtils.addBucketToDB(volumeName, bucketName,
+        omMetadataManager);
+    omBucketInfo = omMetadataManager.getBucketTable().get(
+        bucketKey);
+    omBucketInfo.incrUsedBytes(1000);
+    omBucketInfo.incrUsedNamespace(100L);
+    omMetadataManager.getBucketTable().addCacheEntry(new CacheKey<>(bucketKey),
+        new CacheValue<>(Optional.of(omBucketInfo), 1L));
+    omMetadataManager.getBucketTable().put(bucketKey, omBucketInfo);
+
+    // prevalidate bucket
+    omBucketInfo = omMetadataManager.getBucketTable().get(bucketKey);
+    Assert.assertEquals(1000L, omBucketInfo.getUsedBytes());
+
+    // perform delete
+    OMDirectoriesPurgeResponseWithFSO omClientResponse
+        = (OMDirectoriesPurgeResponseWithFSO) omKeyPurgeRequest
+        .validateAndUpdateCache(ozoneManager, 100L,
+            ozoneManagerDoubleBufferHelper);
+
+    // validate bucket info, no change expected
+    omBucketInfo = omMetadataManager.getBucketTable().get(
+        bucketKey);
+    Assert.assertEquals(1000L, omBucketInfo.getUsedBytes());
+
+    try (BatchOperation batchOperation =
+             omMetadataManager.getStore().initBatchOperation()) {
+
+      omClientResponse.addToDBBatch(omMetadataManager, batchOperation);
+
+      // Do manual commit and see whether addToBatch is successful or not.
+      omMetadataManager.getStore().commitBatchOperation(batchOperation);
+    }
+
+    // The keys should exist in the DeletedKeys table after dir delete
+    for (String deletedKey : deletedKeyNames) {
+      Assert.assertTrue(omMetadataManager.getDeletedTable().isExist(
+          deletedKey));
+    }
+  }
 }