Disable delete optimization and exit ref loop faster

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;
   }