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,10 +75,13 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
                 volumeName, bucketName);
             lockSet.add(volBucketPair);
           }
+
           OmBucketInfo omBucketInfo = getBucketInfo(omMetadataManager,
               volumeName, bucketName);
           // bucketInfo can be null in case of delete volume or bucket
-          if (null != omBucketInfo) {
+          // or key does not belong to bucket as bucket is recreated
+          if (null != omBucketInfo
+              && omBucketInfo.getObjectID() == path.getBucketId()) {
             omBucketInfo.incrUsedNamespace(-1L);
             volBucketInfoMap.putIfAbsent(volBucketPair, omBucketInfo);
           }
@@ -98,7 +101,9 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
           OmBucketInfo omBucketInfo = getBucketInfo(omMetadataManager,
               volumeName, bucketName);
           // bucketInfo can be null in case of delete volume or bucket
-          if (null != omBucketInfo) {
+          // or key does not belong to bucket as bucket is recreated
+          if (null != omBucketInfo
+              && omBucketInfo.getObjectID() == path.getBucketId()) {
             omBucketInfo.incrUsedBytes(-sumBlockLengths(keyInfo));
             omBucketInfo.incrUsedNamespace(-1L);
             volBucketInfoMap.putIfAbsent(volBucketPair, omBucketInfo);

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

@@ -38,6 +38,7 @@
 import org.apache.hadoop.ozone.om.response.key.OMKeyPurgeResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
+import org.jetbrains.annotations.NotNull;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -108,7 +109,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 +118,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);
@@ -180,25 +181,18 @@ public void testValidateAndUpdateCacheCheckQuota() 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);
-    }
+    List<String> deletedKeyNames = validateDeletedKeysTable(deletedKeyInfos);
 
     // 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
@@ -210,16 +204,91 @@ public void testValidateAndUpdateCacheCheckQuota() throws Exception {
     Assert.assertEquals(0L * deletedKeyNames.size(),
         omBucketInfo.getUsedBytes());
 
+    performBatchOperationCommit(omClientResponse);
+
+    // The keys should exist in the DeletedKeys table after dir delete
+    validateDeletedKeys(deletedKeyNames);
+  }
+
+  @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 = validateDeletedKeysTable(deletedKeyInfos);
+
+    // 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());
+
+    performBatchOperationCommit(omClientResponse);
+
+    // The keys should exist in the DeletedKeys table after dir delete
+    validateDeletedKeys(deletedKeyNames);
+  }
+
+  private void performBatchOperationCommit(
+      OMDirectoriesPurgeResponseWithFSO omClientResponse) throws IOException {
     try (BatchOperation batchOperation =
-        omMetadataManager.getStore().initBatchOperation()) {
+             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
+  @NotNull
+  private List<String> validateDeletedKeysTable(
+      List<OmKeyInfo> deletedKeyInfos) throws IOException {
+    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);
+    }
+    return deletedKeyNames;
+  }
+
+  private void validateDeletedKeys(
+      List<String> deletedKeyNames) throws IOException {
     for (String deletedKey : deletedKeyNames) {
       Assert.assertTrue(omMetadataManager.getDeletedTable().isExist(
           deletedKey));