HDDS-16773. Release bucket quota once for duplicate DeleteKeys entries

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

@@ -37,10 +37,12 @@
 import java.nio.file.InvalidPathException;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.hdds.utils.db.Table;
 import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
@@ -184,6 +186,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       // Validate bucket and volume exists or not.
       validateBucketAndVolume(omMetadataManager, volumeName, bucketName);
       String volumeOwner = getVolumeOwner(omMetadataManager, volumeName);
+      Set<String> resolvedKeys = new HashSet<>();
 
       for (indexFailed = 0; indexFailed < length; indexFailed++) {
         String keyName = deleteKeyArgs.getKeys(indexFailed);
@@ -224,6 +227,11 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
           perfMetrics.setDeleteKeysAclCheckLatencyNs(Time.monotonicNowNanos() - startNanosDeleteKeysAclCheckLatency);
           OzoneFileStatus fileStatus = getOzoneKeyStatus(
               ozoneManager, omMetadataManager, volumeName, bucketName, keyName);
+          // Different FSO path spellings can resolve to the same entry. Release its quota only once.
+          if (!resolvedKeys.add(getBucketLayout().isFileSystemOptimized() ? omKeyInfo.getPath() : objectKey)) {
+            deleteKeys.remove(keyName);
+            continue;
+          }
           addKeyToAppropriateList(omKeyInfoList, omKeyInfo, dirList,
               fileStatus);
           deleteKeysInfo.add(omKeyInfo);

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

@@ -187,6 +187,42 @@ public void testPartialDeletePublishesBucketUsage() throws Exception {
         omMetadataManager.getBucketTable().getSkipCache(bucketKey).getUsedNamespace());
   }
 
+  @Test
+  public void testDuplicateKeysReleaseQuotaOnce() throws Exception {
+    createPreRequisites();
+    DeleteKeyArgs keys = omRequest.getDeleteKeysRequest().getDeleteKeys();
+    OMRequest request = omRequest.toBuilder().setDeleteKeysRequest(omRequest.getDeleteKeysRequest().toBuilder()
+        .setDeleteKeys(keys.toBuilder().addAllKeys(deleteKeyList))).build();
+    checkDuplicateKeysReleaseQuotaOnce(request, deleteKeyList.size());
+  }
+
+  protected void checkDuplicateKeysReleaseQuotaOnce(OMRequest request, int uniqueKeys) throws Exception {
+    String bucketKey = omMetadataManager.getBucketKey(volumeName, bucketName);
+    // The fixtures insert entries directly, so account for the entries being deleted here.
+    omMetadataManager.getBucketTable().getCacheValue(new CacheKey<>(bucketKey))
+        .getCacheValue().incrUsedNamespace(uniqueKeys);
+    long usedNamespace = omMetadataManager.getBucketTable().get(bucketKey).getUsedNamespace();
+    long totalNamespace = omMetadataManager.getBucketTable().get(bucketKey).getTotalBucketNamespace();
+    OMClientResponse response = getOmKeysDeleteRequest(request).validateAndUpdateCache(ozoneManager, 100L);
+
+    assertEquals(Status.OK, response.getOMResponse().getStatus());
+    assertEquals(0, response.getOMResponse().getDeleteKeysResponse().getUnDeletedKeys().getKeysCount());
+    assertEquals(0, response.getOMResponse().getDeleteKeysResponse().getErrorsCount());
+    assertEquals(usedNamespace - uniqueKeys, omMetadataManager.getBucketTable().get(bucketKey).getUsedNamespace());
+    if (!getBucketLayout().isFileSystemOptimized()) {
+      // The base fixture contains empty keys, whose namespace is released immediately.
+      assertEquals(totalNamespace - uniqueKeys,
+          omMetadataManager.getBucketTable().get(bucketKey).getTotalBucketNamespace());
+    }
+    try (BatchOperation batchOperation = omMetadataManager.getStore().initBatchOperation()) {
+      response.checkAndUpdateDB(omMetadataManager, batchOperation);
+      omMetadataManager.getStore().commitBatchOperation(batchOperation);
+    }
+    assertEquals(usedNamespace - uniqueKeys,
+        omMetadataManager.getBucketTable().getSkipCache(bucketKey).getUsedNamespace());
+    assertEquals(uniqueKeys, ozoneManager.getMetrics().getNumKeyDeletes());
+  }
+
   protected long expectedNamespaceReleasedOnPartialDelete() {
     return KEY_COUNT;
   }

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

@@ -101,6 +101,18 @@ public void testUpdateIDCountMatchKeyCount() throws Exception {
     checkDeleteKeysResponse(omKeysDeleteRequest);
   }
 
+  @Test
+  public void testDuplicatePathsReleaseQuotaOnce() throws Exception {
+    createPreRequisites();
+    DeleteKeyArgs keys = getOmRequest().getDeleteKeysRequest().getDeleteKeys();
+    OMRequest request = getOmRequest().toBuilder()
+        .setDeleteKeysRequest(getOmRequest().getDeleteKeysRequest().toBuilder()
+            .setDeleteKeys(keys.toBuilder().clearKeys()
+                .addKeys("dir0/file0").addKeys("dir0//file0").addKeys("/dir0/file0")
+                .addKeys("dir1/").addKeys("/dir1"))).build();
+    checkDuplicateKeysReleaseQuotaOnce(request, 2);
+  }
+
   @Override
   protected void createPreRequisites() throws Exception {
     createPreRequisites(RequestSource.USER);