HBASE-20226: Parallelize region manifest deletes (#2159)

We observed this delete call to be a bottleneck for table with lots of
regions. Patch attempts to parallelize them.

Signed-off-by: Andrew Purtell <apurtell@apache.org>
(cherry picked from commit f07f30ae24)

hbase-server/src/main/java/org/apache/hadoop/hbase/snapshot/SnapshotManifest.java

@@ -20,11 +20,15 @@ package org.apache.hadoop.hbase.snapshot;
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.io.InterruptedIOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorCompletionService;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import org.apache.hadoop.conf.Configuration;
@@ -513,9 +517,6 @@ public final class SnapshotManifest {
           workingDir, desc);
       v2Regions = SnapshotManifestV2.loadRegionManifests(conf, tpool, workingDirFs,
           workingDir, desc, manifestSizeLimit);
-    } finally {
-      tpool.shutdown();
-    }
 
       SnapshotDataManifest.Builder dataManifestBuilder = SnapshotDataManifest.newBuilder();
       dataManifestBuilder.setTableSchema(ProtobufUtil.toTableSchema(htd));
@@ -541,16 +542,39 @@ public final class SnapshotManifest {
       // The extra files in the snapshot directory will not give any problem,
       // since they have the same content as the data manifest, and even by re-reading
       // them we will get the same information.
-    if (v1Regions != null && v1Regions.size() > 0) {
+      int totalDeletes = 0;
+      ExecutorCompletionService<Void> completionService = new ExecutorCompletionService<>(tpool);
+      if (v1Regions != null) {
         for (SnapshotRegionManifest regionManifest: v1Regions) {
+          ++totalDeletes;
+          completionService.submit(() -> {
             SnapshotManifestV1.deleteRegionManifest(workingDirFs, workingDir, regionManifest);
+            return null;
+          });
         }
       }
-    if (v2Regions != null && v2Regions.size() > 0) {
+      if (v2Regions != null) {
         for (SnapshotRegionManifest regionManifest: v2Regions) {
+          ++totalDeletes;
+          completionService.submit(() -> {
             SnapshotManifestV2.deleteRegionManifest(workingDirFs, workingDir, regionManifest);
+            return null;
+          });
         }
       }
+      // Wait for the deletes to finish.
+      for (int i = 0; i < totalDeletes; i++) {
+        try {
+          completionService.take().get();
+        } catch (InterruptedException ie) {
+          throw new InterruptedIOException(ie.getMessage());
+        } catch (ExecutionException e) {
+          throw new IOException("Error deleting region manifests", e.getCause());
+        }
+      }
+    } finally {
+      tpool.shutdown();
+    }
   }
 
   /*