Merge branch '2.1'

Author: ddanielr

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

@@ -441,6 +441,14 @@ are performed (e.g. Bulk Import). This property specifies the maximum number of
       "Time to wait between scanning tables to identify ranges of tablets that can be "
           + " auto-merged. Valid ranges will be have merge fate ops submitted.",
       "4.0.0"),
+  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_TIMEOUT("manager.bulk.timeout", "5m", PropertyType.TIMEDURATION,
       "The time to wait for a tablet server to process a bulk import request.", "1.4.3"),
   MANAGER_RENAME_THREADS("manager.rename.threadpool.size", "20", PropertyType.COUNT,

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

@@ -110,7 +110,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,10 +23,10 @@
 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.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;
@@ -67,11 +67,15 @@ public CleanUp(TableId tableId, NamespaceId namespaceId) {
   public Repo<FateEnv> call(FateId fateId, FateEnv env) {
     int refCount = 0;
 
-    try {
+    if (!env.getContext().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 = env.getContext();
-      try (BatchScanner bs =
-          client.createBatchScanner(SystemTables.METADATA.tableName(), Authorizations.EMPTY, 8)) {
+      try (BatchScanner bs = env.getContext().createBatchScanner(SystemTables.METADATA.tableName(),
+          Authorizations.EMPTY, 8)) {
         Range allTables = TabletsSection.getRange();
         Range tableRange = TabletsSection.getRange(tableId);
         Range beforeTable =
@@ -85,15 +89,15 @@ public Repo<FateEnv> call(FateId fateId, FateEnv env) {
 
         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 {}",
+            SystemTables.METADATA.tableName(), tableId, e);
       }
-
-    } catch (Exception e) {
-      refCount = -1;
-      log.error("Failed to scan " + SystemTables.METADATA.tableName()
-          + " looking for references to deleted table " + tableId, e);
     }
 
     // remove metadata table entries
@@ -105,7 +109,7 @@ public Repo<FateEnv> call(FateId fateId, FateEnv env) {
       // are dropped and the operation completes, then the deletes will not be repeated.
       MetadataTableUtil.deleteTable(tableId, refCount != 0, env.getContext(), null);
     } catch (Exception e) {
-      log.error("error deleting " + tableId + " from metadata table", e);
+      log.error("error deleting {} from metadata table", tableId, e);
     }
 
     if (refCount == 0) {
@@ -132,7 +136,7 @@ public Repo<FateEnv> call(FateId fateId, FateEnv env) {
       env.getTableManager().removeTable(tableId, namespaceId);
       env.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
@@ -146,7 +150,7 @@ public Repo<FateEnv> call(FateId fateId, FateEnv env) {
     Utils.unreserveTable(env.getContext(), tableId, fateId, LockType.WRITE);
     Utils.unreserveNamespace(env.getContext(), namespaceId, fateId, LockType.READ);
 
-    LoggerFactory.getLogger(CleanUp.class).debug("Deleted table " + tableId);
+    log.debug("Deleted table {}", tableId);
 
     return null;
   }