HDDS-11508. Decouple delete batch limits from Ratis request size for DirectoryDeletingService.

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/Table.java

@@ -336,6 +336,14 @@ interface KeyValue<KEY, VALUE> {
     KEY getKey() throws IOException;
 
     VALUE getValue() throws IOException;
+
+    default byte[] getRawKey() throws IOException {
+      return null;
+    }
+
+    default byte[] getRawValue() throws IOException {
+      return null;
+    }
   }
 
   static <K, V> KeyValue<K, V> newKeyValue(K key, V value) {
@@ -375,6 +383,54 @@ public int hashCode() {
     };
   }
 
+  static <K, V> KeyValue<K, V> newKeyValue(K key, V value, byte[] rawKey, byte[] rawValue) {
+    return new KeyValue<K, V>() {
+      @Override
+      public K getKey() {
+        return key;
+      }
+
+      @Override
+      public V getValue() {
+        return value;
+      }
+
+      @Override
+      public byte[] getRawKey() throws IOException {
+        return rawKey;
+      }
+
+      @Override
+      public byte[] getRawValue() throws IOException {
+        return rawValue;
+      }
+
+      @Override
+      public String toString() {
+        return "(key=" + key + ", value=" + value + ")";
+      }
+
+      @Override
+      public boolean equals(Object obj) {
+        if (!(obj instanceof KeyValue)) {
+          return false;
+        }
+        KeyValue<?, ?> kv = (KeyValue<?, ?>) obj;
+        try {
+          return getKey().equals(kv.getKey()) && getValue().equals(kv.getValue());
+        } catch (IOException e) {
+          throw new RuntimeException(e);
+        }
+      }
+
+      @Override
+      public int hashCode() {
+        return Objects.hash(getKey(), getValue());
+      }
+    };
+  }
+
+
   /** A {@link TableIterator} to iterate {@link KeyValue}s. */
   interface KeyValueIterator<KEY, VALUE>
       extends TableIterator<KEY, KeyValue<KEY, VALUE>> {

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/TypedTable.java

@@ -573,6 +573,14 @@ public KEY getKey() throws IOException {
     public VALUE getValue() throws IOException {
       return decodeValue(rawKeyValue.getValue());
     }
+
+    public byte[] getRawKey() throws IOException {
+      return rawKeyValue.getKey();
+    }
+
+    public byte[] getRawValue() throws IOException {
+      return rawKeyValue.getValue();
+    }
   }
 
   RawIterator<CodecBuffer> newCodecBufferTableIterator(
@@ -597,9 +605,11 @@ public CodecBuffer get() {
       @Override
       KeyValue<KEY, VALUE> convert(KeyValue<CodecBuffer, CodecBuffer> raw)
           throws IOException {
-        final KEY key = keyCodec.fromCodecBuffer(raw.getKey());
-        final VALUE value = valueCodec.fromCodecBuffer(raw.getValue());
-        return Table.newKeyValue(key, value);
+        CodecBuffer keyCodecBuffer = raw.getKey();
+        final KEY key = keyCodec.fromCodecBuffer(keyCodecBuffer);
+        CodecBuffer valueCodecBuffer = raw.getValue();
+        final VALUE value = valueCodec.fromCodecBuffer(valueCodecBuffer);
+        return Table.newKeyValue(key, value, keyCodecBuffer.getArray(), valueCodecBuffer.getArray());
       }
     };
   }

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestDirectoryDeletingServiceWithFSO.java

@@ -90,7 +90,6 @@
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_BLOCK_DELETING_SERVICE_INTERVAL;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_ITERATE_BATCH_SIZE;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyList;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.eq;
@@ -155,7 +154,7 @@ public static void teardown() {
   }
 
   @AfterEach
-  public void cleanup() {
+  public void cleanup() throws InterruptedException, TimeoutException {
     assertDoesNotThrow(() -> {
       Path root = new Path("/");
       FileStatus[] fileStatuses = fs.listStatus(root);
@@ -273,8 +272,6 @@ public void testDeleteWithLargeSubPathsThanBatchSize() throws Exception {
     assertTableRowCount(dirTable, 1);
 
     assertSubPathsCount(dirDeletingService::getMovedFilesCount, 15);
-    // 15 subDir + 3 parentDir
-    assertSubPathsCount(dirDeletingService::getMovedDirsCount, 18);
     assertSubPathsCount(dirDeletingService::getDeletedDirsCount, 19);
 
     assertEquals(15, metrics.getNumSubFilesSentForPurge());
@@ -335,7 +332,7 @@ public void testDeleteWithMultiLevels() throws Exception {
     assertTableRowCount(dirTable, 0);
 
     assertSubPathsCount(dirDeletingService::getMovedFilesCount, 3);
-    assertSubPathsCount(dirDeletingService::getMovedDirsCount, 2);
+    assertSubPathsCount(dirDeletingService::getMovedDirsCount, 0);
     assertSubPathsCount(dirDeletingService::getDeletedDirsCount, 5);
     assertEquals(5, metrics.getNumDirsSentForPurge());
     assertEquals(5, metrics.getNumDirsPurged());
@@ -431,7 +428,8 @@ public void testDeleteWithMultiLevelsBlockDoubleBuffer() throws Exception {
     omDoubleBuffer.stopDaemon();
 
     OzoneVolume volume = client.getObjectStore().getVolume(volumeName);
-    OzoneBucket bucket = volume.getBucket(bucketName);    long volumeId = metadataManager.getVolumeId(volumeName);
+    OzoneBucket bucket = volume.getBucket(bucketName);
+    long volumeId = metadataManager.getVolumeId(volumeName);
 
     // manually delete dir and add to deleted table. namespace count occupied "1" as manual deletion do not reduce
     long bucketId = metadataManager.getBucketId(volumeName, bucketName);
@@ -629,13 +627,14 @@ public void testAOSKeyDeletingWithSnapshotCreateParallelExecution()
     assertTableRowCount(deletedDirTable, initialDeletedCount + 1);
     assertTableRowCount(renameTable, initialRenameCount + 1);
     Mockito.doAnswer(i -> {
-      List<OzoneManagerProtocolProtos.PurgePathRequest> purgePathRequestList = i.getArgument(5);
+      List<OzoneManagerProtocolProtos.PurgePathRequest> purgePathRequestList = i.getArgument(4);
       for (OzoneManagerProtocolProtos.PurgePathRequest purgeRequest : purgePathRequestList) {
         Assertions.assertNotEquals(deletePathKey, purgeRequest.getDeletedDir());
       }
-      return i.callRealMethod();
-    }).when(service).optimizeDirDeletesAndSubmitRequest(anyLong(), anyLong(), anyLong(),
-        anyLong(), anyList(), anyList(), eq(null), anyLong(), anyInt(), Mockito.any(), any(), anyLong());
+      return null;
+    }).when(service).optimizeDirDeletesAndSubmitRequest(anyLong(), anyLong(),
+        anyLong(), anyList(), anyList(), eq(null), anyLong(), anyLong(), Mockito.any(), any(),
+        anyLong());
 
     Mockito.doAnswer(i -> {
       store.createSnapshot(testVolumeName, testBucketName, snap2);
@@ -783,9 +782,54 @@ public void testDirDeletedTableCleanUpForSnapshot() throws Exception {
     assertSubPathsCount(dirDeletingService::getDeletedDirsCount, 0);
 
     // Manual cleanup deletedDirTable for next tests
+    client.getObjectStore().deleteSnapshot(volumeName, bucketName, "snap1");
     cleanupTables();
   }
 
+  @Test
+  public void testDeleteWithRequestSizeExceedingRatisRequestSizeLimit() throws Exception {
+    Table<String, OmKeyInfo> deletedDirTable =
+        cluster.getOzoneManager().getMetadataManager().getDeletedDirTable();
+    Table<String, OmKeyInfo> keyTable =
+        cluster.getOzoneManager().getMetadataManager()
+            .getKeyTable(getFSOBucketLayout());
+    Table<String, OmDirectoryInfo> dirTable =
+        cluster.getOzoneManager().getMetadataManager().getDirectoryTable();
+    assertTableRowCount(deletedDirTable, 0);
+    assertTableRowCount(keyTable, 0);
+    assertTableRowCount(dirTable, 0);
+    Path root = new Path("/rootDir1");
+    Path appRoot = new Path(root, "appRoot1");
+    // Creates 2 parent dirs from root.
+    fs.mkdirs(appRoot);
+    for (int i = 1; i <= 5; i++) {
+      Path childDir = new Path(appRoot, "dir" + i);
+      fs.mkdirs(childDir);
+    }
+    DirectoryDeletingService dirDeletingService =
+        (DirectoryDeletingService) cluster.getOzoneManager().getKeyManager()
+            .getDirDeletingService();
+    // Before delete
+    assertTableRowCount(deletedDirTable, 0);
+    assertTableRowCount(keyTable, 0);
+    assertTableRowCount(dirTable, 7);
+    assertSubPathsCount(dirDeletingService::getMovedFilesCount, 0);
+    assertSubPathsCount(dirDeletingService::getMovedDirsCount, 0);
+    assertSubPathsCount(dirDeletingService::getDeletedDirsCount, 0);
+    long preRunCount = dirDeletingService.getRunCount().get();
+    dirDeletingService.setRatisByteLimit(1000);
+    // Delete the appRoot
+    fs.delete(appRoot, true);
+    // After Delete
+    checkPath(appRoot);
+    assertTableRowCount(deletedDirTable, 0);
+    assertTableRowCount(dirTable, 1);
+    assertSubPathsCount(dirDeletingService::getMovedDirsCount, 4);
+    assertSubPathsCount(dirDeletingService::getDeletedDirsCount, 5);
+    assertThat(dirDeletingService.getRunCount().get()).isGreaterThan(preRunCount);
+  }
+
+
   private void cleanupTables() throws IOException {
     OMMetadataManager metadataManager =
         cluster.getOzoneManager().getMetadataManager();

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/recon/TestReconInsightsForDeletedDirectories.java

@@ -52,6 +52,8 @@
 
 import javax.ws.rs.core.Response;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
@@ -124,7 +126,7 @@ public static void teardown() {
   }
 
   @AfterEach
-  public void cleanup() {
+  public void cleanup() throws IOException {
     assertDoesNotThrow(() -> {
       Path root = new Path("/");
       FileStatus[] fileStatuses = fs.listStatus(root);
@@ -417,24 +419,31 @@ private void cleanupTables() throws IOException {
     OMMetadataManager metadataManager =
         cluster.getOzoneManager().getMetadataManager();
 
-    try (TableIterator<?, ?> it = metadataManager.getDeletedDirTable()
-        .iterator()) {
-      removeAllFromDB(it);
+    Table<String, OmKeyInfo> deletedDirTable =
+        metadataManager.getDeletedDirTable();
+    try (TableIterator<String, ? extends Table.KeyValue<String, ?>> it = deletedDirTable.iterator()) {
+      removeAllFromDB(it, deletedDirTable);
     }
-    try (TableIterator<?, ?> it = metadataManager.getFileTable().iterator()) {
-      removeAllFromDB(it);
+    Table<String, OmKeyInfo> fileTable = metadataManager.getFileTable();
+    try (TableIterator<String, ? extends Table.KeyValue<String, ?>> it = fileTable.iterator()) {
+      removeAllFromDB(it, fileTable);
     }
-    try (TableIterator<?, ?> it = metadataManager.getDirectoryTable()
-        .iterator()) {
-      removeAllFromDB(it);
+    Table<String, OmDirectoryInfo> directoryTable =
+        metadataManager.getDirectoryTable();
+    try (TableIterator<String, ? extends Table.KeyValue<String, ?>> it = directoryTable.iterator()) {
+      removeAllFromDB(it, directoryTable);
     }
   }
 
-  private static void removeAllFromDB(TableIterator<?, ?> iterator)
-      throws IOException {
+  private static void removeAllFromDB(
+      TableIterator<String, ? extends Table.KeyValue<String, ?>> iterator,
+      Table<String, ?> table) throws IOException {
+    List<String> keysToDelete = new ArrayList<>();
     while (iterator.hasNext()) {
-      iterator.next();
-      iterator.removeFromDB();
+      keysToDelete.add(iterator.next().getKey());
+    }
+    for (String keyToDelete : keysToDelete) {
+      table.delete(keyToDelete);
     }
   }
 

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

@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.ozone.om;
+
+import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
+
+import java.util.List;
+
+/**
+ * Used in {@link org.apache.hadoop.ozone.om.service.DirectoryDeletingService}
+ * to capture the result of each delete task.
+ */
+public class DeleteKeysResult {
+
+  private List<OmKeyInfo> keysToDelete;
+  private long consumedSize;
+
+  private boolean processedKeys;
+
+  public DeleteKeysResult(List<OmKeyInfo> keysToDelete,
+      long consumedSize, boolean processedKeys) {
+    this.keysToDelete = keysToDelete;
+    this.consumedSize = consumedSize;
+    this.processedKeys = processedKeys;
+  }
+
+  public List<OmKeyInfo> getKeysToDelete() {
+    return keysToDelete;
+  }
+
+  public long getConsumedSize() {
+    return consumedSize;
+  }
+
+  public boolean isProcessedKeys() {
+    return processedKeys;
+  }
+}

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

@@ -277,23 +277,21 @@ default List<Table.KeyValue<String, OmKeyInfo>> getDeletedDirEntries(String volu
    * Returns all sub directories under the given parent directory.
    *
    * @param parentInfo
-   * @param numEntries
    * @return list of dirs
    * @throws IOException
    */
-  List<OmKeyInfo> getPendingDeletionSubDirs(long volumeId, long bucketId,
-      OmKeyInfo parentInfo, long numEntries) throws IOException;
+  DeleteKeysResult getPendingDeletionSubDirs(long volumeId, long bucketId,
+      OmKeyInfo parentInfo, long remainingBufLimit) throws IOException;
 
   /**
    * Returns all sub files under the given parent directory.
    *
    * @param parentInfo
-   * @param numEntries
    * @return list of files
    * @throws IOException
    */
-  List<OmKeyInfo> getPendingDeletionSubFiles(long volumeId,
-      long bucketId, OmKeyInfo parentInfo, long numEntries)
+  DeleteKeysResult getPendingDeletionSubFiles(long volumeId,
+      long bucketId, OmKeyInfo parentInfo, long remainingBufLimit)
           throws IOException;
 
   /**