HBASE-24588 : Submit task for NormalizationPlan

hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncHBaseAdmin.java

@@ -1294,7 +1294,7 @@ public CompletableFuture<Void> mergeRegions(List<byte[]> nameOfRegionsToMerge, b
         return;
       }
 
-      MergeTableRegionsRequest request = null;
+      final MergeTableRegionsRequest request;
       try {
         request = RequestConverter.buildMergeTableRegionsRequest(encodedNameOfRegionsToMerge,
           forcible, ng.getNonceGroup(), ng.newNonce());
@@ -1304,8 +1304,8 @@ public CompletableFuture<Void> mergeRegions(List<byte[]> nameOfRegionsToMerge, b
       }
 
       addListener(
-        this.<MergeTableRegionsRequest, MergeTableRegionsResponse> procedureCall(tableName, request,
-          (s, c, req, done) -> s.mergeTableRegions(c, req, done), (resp) -> resp.getProcId(),
+        this.procedureCall(tableName, request,
+          MasterService.Interface::mergeTableRegions, MergeTableRegionsResponse::getProcId,
           new MergeTableRegionProcedureBiConsumer(tableName)),
         (ret, err2) -> {
           if (err2 != null) {
@@ -1480,7 +1480,7 @@ public CompletableFuture<Void> splitRegion(byte[] regionName, byte[] splitPoint)
   private CompletableFuture<Void> split(final RegionInfo hri, byte[] splitPoint) {
     CompletableFuture<Void> future = new CompletableFuture<>();
     TableName tableName = hri.getTable();
-    SplitTableRegionRequest request = null;
+    final SplitTableRegionRequest request;
     try {
       request = RequestConverter.buildSplitTableRegionRequest(hri, splitPoint, ng.getNonceGroup(),
         ng.newNonce());
@@ -1490,8 +1490,8 @@ private CompletableFuture<Void> split(final RegionInfo hri, byte[] splitPoint) {
     }
 
     addListener(
-      this.<SplitTableRegionRequest, SplitTableRegionResponse> procedureCall(tableName,
-        request, (s, c, req, done) -> s.splitRegion(c, req, done), (resp) -> resp.getProcId(),
+      this.procedureCall(tableName,
+        request, MasterService.Interface::splitRegion, SplitTableRegionResponse::getProcId,
         new SplitTableRegionProcedureBiConsumer(tableName)),
       (ret, err2) -> {
         if (err2 != null) {

hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java

@@ -473,6 +473,10 @@ public void run() {
   // Cached clusterId on stand by masters to serve clusterID requests from clients.
   private final CachedClusterId cachedClusterId;
 
+  // Split/Merge Normalization plan executes asynchronously and the caller blocks on
+  // waiting max 5 sec for single plan to complete with success/failure.
+  private static final int NORMALIZATION_PLAN_WAIT_TIMEOUT = 5;
+
   public static class RedirectServlet extends HttpServlet {
     private static final long serialVersionUID = 2894774810058302473L;
     private final int regionServerInfoPort;
@@ -1935,6 +1939,8 @@ public boolean normalizeRegions() throws IOException {
       Collections.shuffle(allEnabledTables);
 
       try (final Admin admin = asyncClusterConnection.toConnection().getAdmin()) {
+        int failedNormalizationPlans = 0;
+        final List<Future<?>> submittedPlanList = new ArrayList<>();
         for (TableName table : allEnabledTables) {
           if (table.isSystemTable()) {
             continue;
@@ -1957,17 +1963,30 @@ public boolean normalizeRegions() throws IOException {
             continue;
           }
 
-          // as of this writing, `plan.execute()` is non-blocking, so there's no artificial rate-
-          // limiting of merge requests due to this serial loop.
+          // as of this writing, `plan.submit()` is non-blocking and uses Async Admin APIs to
+          // submit task , so there's no artificial rate-
+          // limiting of merge/split requests due to this serial loop.
           for (NormalizationPlan plan : plans) {
-            plan.execute(admin);
+            Future<Void> future = plan.submit(admin);
+            submittedPlanList.add(future);
             if (plan.getType() == PlanType.SPLIT) {
               splitPlanCount++;
             } else if (plan.getType() == PlanType.MERGE) {
               mergePlanCount++;
             }
           }
         }
+        for (Future<?> submittedPlan : submittedPlanList) {
+          try {
+            submittedPlan.get(NORMALIZATION_PLAN_WAIT_TIMEOUT, TimeUnit.SECONDS);
+          } catch (Exception e) {
+            failedNormalizationPlans++;
+            LOG.error("Submitted normalization plan failed with error: ", e);
+          }
+        }
+        int totalNormalizationPlans = submittedPlanList.size();
+        LOG.info("Normalizer run was able to successfully execute {}/{} merge or split plans",
+          totalNormalizationPlans - failedNormalizationPlans, totalNormalizationPlans);
       }
     } finally {
       normalizationInProgressLock.unlock();

hbase-server/src/main/java/org/apache/hadoop/hbase/master/normalizer/EmptyNormalizationPlan.java

@@ -20,7 +20,8 @@
 
 import org.apache.yetus.audience.InterfaceAudience;
 import org.apache.hadoop.hbase.client.Admin;
-import org.apache.hadoop.hbase.master.normalizer.NormalizationPlan.PlanType;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
 
 /**
  * Plan which signifies that no normalization is required,
@@ -44,7 +45,8 @@ public static EmptyNormalizationPlan getInstance(){
    * No-op for empty plan.
    */
   @Override
-  public void execute(Admin admin) {
+  public Future<Void> submit(Admin admin) {
+    return CompletableFuture.completedFuture(null);
   }
 
   @Override

hbase-server/src/main/java/org/apache/hadoop/hbase/master/normalizer/MergeNormalizationPlan.java

@@ -19,6 +19,7 @@
 package org.apache.hadoop.hbase.master.normalizer;
 
 import java.io.IOException;
+import java.util.concurrent.Future;
 
 import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.RegionInfo;
@@ -41,6 +42,21 @@ public MergeNormalizationPlan(RegionInfo firstRegion, RegionInfo secondRegion) {
     this.secondRegion = secondRegion;
   }
 
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public Future<Void> submit(Admin admin) throws IOException {
+    LOG.info("Executing merging normalization plan: " + this);
+    byte[][] regionsToMerge = new byte[2][];
+    regionsToMerge[0] = firstRegion.getEncodedNameAsBytes();
+    regionsToMerge[1] = secondRegion.getEncodedNameAsBytes();
+    // Do not use force=true as corner cases can happen, non adjacent regions,
+    // merge with a merged child region with no GC done yet, it is going to
+    // cause all different issues.
+    return admin.mergeRegionsAsync(regionsToMerge, false);
+  }
+
   @Override
   public PlanType getType() {
     return PlanType.MERGE;
@@ -62,20 +78,4 @@ public String toString() {
       '}';
   }
 
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public void execute(Admin admin) {
-    LOG.info("Executing merging normalization plan: " + this);
-    try {
-      // Do not use force=true as corner cases can happen, non adjacent regions,
-      // merge with a merged child region with no GC done yet, it is going to
-      // cause all different issues.
-      admin.mergeRegionsAsync(firstRegion.getEncodedNameAsBytes(),
-        secondRegion.getEncodedNameAsBytes(), false);
-    } catch (IOException ex) {
-      LOG.error("Error during region merge: ", ex);
-    }
-  }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/master/normalizer/NormalizationPlan.java

@@ -20,6 +20,8 @@
 
 import org.apache.yetus.audience.InterfaceAudience;
 import org.apache.hadoop.hbase.client.Admin;
+import java.io.IOException;
+import java.util.concurrent.Future;
 
 /**
  * Interface for normalization plan.
@@ -33,10 +35,13 @@ enum PlanType {
   }
 
   /**
-   * Executes normalization plan on cluster (does actual splitting/merging work).
+   * Submits normalization plan on cluster (does actual splitting/merging work) and
+   * returns Future reference for caller.
    * @param admin instance of Admin
+   * @return Future reference to submitted task
+   * @throws IOException If plan submission to Admin fails
    */
-  void execute(Admin admin);
+  Future<Void> submit(Admin admin) throws IOException;
 
   /**
    * @return the type of this plan

hbase-server/src/main/java/org/apache/hadoop/hbase/master/normalizer/SplitNormalizationPlan.java

@@ -18,7 +18,9 @@
  */
 package org.apache.hadoop.hbase.master.normalizer;
 
+import java.io.IOException;
 import java.util.Arrays;
+import java.util.concurrent.Future;
 import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.yetus.audience.InterfaceAudience;
@@ -40,6 +42,15 @@ public SplitNormalizationPlan(RegionInfo regionInfo, byte[] splitPoint) {
     this.splitPoint = splitPoint;
   }
 
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public Future<Void> submit(final Admin admin) throws IOException {
+    LOG.info("Executing splitting normalization plan: " + this);
+    return admin.splitRegionAsync(regionInfo.getRegionName());
+  }
+
   @Override
   public PlanType getType() {
     return PlanType.SPLIT;
@@ -69,16 +80,4 @@ public String toString() {
       '}';
   }
 
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public void execute(Admin admin) {
-    LOG.info("Executing splitting normalization plan: " + this);
-    try {
-      admin.splitRegionAsync(regionInfo.getRegionName()).get();
-    } catch (Exception ex) {
-      LOG.error("Error during region split: ", ex);
-    }
-  }
 }