Disable delete optimization and exit ref loop faster (#6219)

* Exits scan after finding one reference

When delete table is called, the delete marker code checks to see if
any file references exist in other tables.
However, only a single reference has to exist for delete markers to be
created.

Added break out of for loop once a single entry was found.
Removed a nested try block in favor of a single try-with-resources

* Allows table delete optimization to be disabled

Adds a property to allow the scan of the metadata table to be skipped
for table deletes. This forces delete markers to always be created when
deleting tables instead of the manager deleting the volumes immediately.

* Adds tests that verify files are deleted from HDFS

Adds tests to verify that rfiles are removed from HDFS and not just from
the table metadata.

---------

Co-authored-by: Dave Marion <dlmarion@apache.org>

Author: ddanielr

core/src/main/java/org/apache/accumulo/core/conf/Property.java

@@ -391,6 +391,14 @@ public enum Property {
       "Maximum number of threads the TabletGroupWatcher will use in its BatchScanner to"
           + " look for tablets that need maintenance.",
       "2.1.4"),
+  MANAGER_TABLE_DELETE_OPTIMIZATION("manager.table.delete.optimization", "true",
+      PropertyType.BOOLEAN,
+      "When deleting a table the Manager will remove related table directories from "
+          + " the storage volumes if there are no other references to the files in the "
+          + " metadata table. When deleting a lot of tables this optimization can be costly. "
+          + " Setting this value to false will skip this optimization and the table directory "
+          + " cleanup will occur in the Garbage Collector instead.",
+      "2.1.5"),
   MANAGER_BULK_RETRIES("manager.bulk.retries", "3", PropertyType.COUNT,
       "The number of attempts to bulk import a RFile before giving up.", "1.4.0"),
   MANAGER_BULK_THREADPOOL_SIZE("manager.bulk.threadpool.size", "5", PropertyType.COUNT,

server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java

@@ -321,7 +321,7 @@ public static void deleteTable(TableId tableId, boolean insertDeletes, ServerCon
       Ample ample = context.getAmple();
       ms.setRange(new KeyExtent(tableId, null, null).toMetaRange());
 
-      // insert deletes before deleting data from metadata... this makes the code fault tolerant
+      // insert deletes before deleting data from metadata... this makes the code fault-tolerant
       if (insertDeletes) {
 
         ms.fetchColumnFamily(DataFileColumnFamily.NAME);

server/manager/src/main/java/org/apache/accumulo/manager/tableOps/delete/CleanUp.java

@@ -23,11 +23,11 @@
 import java.util.Arrays;
 import java.util.Map.Entry;
 
-import org.apache.accumulo.core.client.AccumuloClient;
 import org.apache.accumulo.core.client.BatchScanner;
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.client.Scanner;
 import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException;
+import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.NamespaceId;
 import org.apache.accumulo.core.data.Range;
@@ -124,11 +124,14 @@ public Repo<Manager> call(long tid, Manager manager) {
 
     int refCount = 0;
 
-    try {
+    if (!manager.getConfiguration().getBoolean(Property.MANAGER_TABLE_DELETE_OPTIMIZATION)) {
+      // Skip scanning the metadata table for each table delete and always allow the GC to handle
+      // file deletion.
+      refCount = -1;
+    } else {
       // look for other tables that references this table's files
-      AccumuloClient client = manager.getContext();
       try (BatchScanner bs =
-          client.createBatchScanner(MetadataTable.NAME, Authorizations.EMPTY, 8)) {
+          manager.getContext().createBatchScanner(MetadataTable.NAME, Authorizations.EMPTY, 8)) {
         Range allTables = TabletsSection.getRange();
         Range tableRange = TabletsSection.getRange(tableId);
         Range beforeTable =
@@ -142,15 +145,15 @@ public Repo<Manager> call(long tid, Manager manager) {
 
         for (Entry<Key,Value> entry : bs) {
           if (entry.getKey().getColumnQualifier().toString().contains("/" + tableId + "/")) {
-            refCount++;
+            refCount = 1;
+            break;
           }
         }
+      } catch (Exception e) {
+        refCount = -1;
+        log.error("Failed to scan {} looking for references to deleted table {}",
+            MetadataTable.NAME, tableId, e);
       }
-
-    } catch (Exception e) {
-      refCount = -1;
-      log.error("Failed to scan " + MetadataTable.NAME + " looking for references to deleted table "
-          + tableId, e);
     }
 
     // remove metadata table entries
@@ -162,14 +165,14 @@ public Repo<Manager> call(long tid, Manager manager) {
       // are dropped and the operation completes, then the deletes will not be repeated.
       MetadataTableUtil.deleteTable(tableId, refCount != 0, manager.getContext(), null);
     } catch (Exception e) {
-      log.error("error deleting " + tableId + " from metadata table", e);
+      log.error("error deleting {} from metadata table", tableId, e);
     }
 
     // remove any problem reports the table may have
     try {
       ProblemReports.getInstance(manager.getContext()).deleteProblemReports(tableId);
     } catch (Exception e) {
-      log.error("Failed to delete problem reports for table " + tableId, e);
+      log.error("Failed to delete problem reports for table {}", tableId, e);
     }
 
     if (refCount == 0) {
@@ -196,7 +199,7 @@ public Repo<Manager> call(long tid, Manager manager) {
       manager.getTableManager().removeTable(tableId);
       manager.getContext().clearTableListCache();
     } catch (Exception e) {
-      log.error("Failed to find table id in zookeeper", e);
+      log.error("Failed to find table id {} in zookeeper", tableId, e);
     }
 
     // remove any permissions associated with this table
@@ -210,7 +213,7 @@ public Repo<Manager> call(long tid, Manager manager) {
     Utils.unreserveTable(manager, tableId, tid, true);
     Utils.unreserveNamespace(manager, namespaceId, tid, false);
 
-    LoggerFactory.getLogger(CleanUp.class).debug("Deleted table " + tableId);
+    log.debug("Deleted table {}", tableId);
 
     return null;
   }