HBASE-24418 Consolidate Normalizer implementations

Simplify our Normalizer story to have just a single, configurable
implementation.

* fold the features of `MergeNormalizer` into
  `SimpleRegionNormalizer`, removing the intermediate abstract class.
* configuration keys for merge-only features now share a common
  structure.
* add configuration to selectively disable normalizer split/merge
  operations.
* `RegionNormalizer` now extends `Configurable` instead of creating a
  new instance of `HBaseConfiguration` or snooping one off of other
  fields.
* avoid the extra RPCs by using `MasterServices` instead of
  `MasterRpcServices`.
* boost test coverage of all the various flags and feature
  combinations.

Signed-off-by: Michael Stack <stack@apache.org>
Signed-off-by: Viraj Jasani <vjasani@apache.org>
Signed-off-by: huaxiangsun <huaxiangsun@apache.org>

hbase-common/src/main/resources/hbase-default.xml

@@ -605,30 +605,54 @@ possible configurations would overwhelm and obscure the important.
     <name>hbase.balancer.period
     </name>
     <value>300000</value>
-    <description>Period at which the region balancer runs in the Master.</description>
+    <description>Period at which the region balancer runs in the Master, in
+      milliseconds.</description>
+  </property>
+  <property>
+    <name>hbase.regions.slop</name>
+    <value>0.001</value>
+    <description>Rebalance if any regionserver has average + (average * slop) regions.
+      The default value of this parameter is 0.001 in StochasticLoadBalancer (the default load
+      balancer), while the default is 0.2 in other load balancers (i.e.,
+      SimpleLoadBalancer).</description>
   </property>
   <property>
     <name>hbase.normalizer.period</name>
     <value>300000</value>
-    <description>Period at which the region normalizer runs in the Master.</description>
+    <description>Period at which the region normalizer runs in the Master, in
+      milliseconds.</description>
+  </property>
+  <property>
+    <name>hbase.normalizer.split.enabled</name>
+    <value>true</value>
+    <description>Whether to split a region as part of normalization.</description>
+  </property>
+  <property>
+    <name>hbase.normalizer.merge.enabled</name>
+    <value>true</value>
+    <description>Whether to merge a region as part of normalization.</description>
   </property>
   <property>
     <name>hbase.normalizer.min.region.count</name>
     <value>3</value>
-    <description>configure the minimum number of regions</description>
+    <description>The minimum number of regions in a table to consider it for merge
+      normalization.</description>
   </property>
   <property>
-    <name>hbase.normalizer.min.region.merge.age</name>
+    <name>hbase.normalizer.merge.min_region_age.days</name>
     <value>3</value>
-    <description>configure the minimum age in days for region before it is considered for merge while
-      normalizing</description>
+    <description>The minimum age for a region to be considered for a merge, in days.</description>
   </property>
   <property>
-    <name>hbase.regions.slop</name>
-    <value>0.001</value>
-    <description>Rebalance if any regionserver has average + (average * slop) regions.
-      The default value of this parameter is 0.001 in StochasticLoadBalancer (the default load balancer),
-      while the default is 0.2 in other load balancers (i.e., SimpleLoadBalancer).</description>
+    <name>hbase.normalizer.merge.min_region_age.days</name>
+    <value>3</value>
+    <description>The minimum age for a region to be considered for a merge, in days.</description>
+  </property>
+  <property>
+    <name>hbase.normalizer.merge.min_region_size.mb</name>
+    <value>1</value>
+    <description>The minimum size for a region to be considered for a merge, in whole
+      MBs.</description>
   </property>
   <property>
     <name>hbase.server.thread.wakefrequency</name>

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

@@ -49,6 +49,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import javax.servlet.ServletException;
@@ -387,6 +388,8 @@ public void run() {
   private final LockManager lockManager = new LockManager(this);
 
   private LoadBalancer balancer;
+  // a lock to prevent concurrent normalization actions.
+  private final ReentrantLock normalizationInProgressLock = new ReentrantLock();
   private RegionNormalizer normalizer;
   private BalancerChore balancerChore;
   private RegionNormalizerChore normalizerChore;
@@ -787,10 +790,11 @@ protected void initializeZKBasedSystemTrackers()
     this.balancer = LoadBalancerFactory.getLoadBalancer(conf);
     this.normalizer = RegionNormalizerFactory.getRegionNormalizer(conf);
     this.normalizer.setMasterServices(this);
-    this.normalizer.setMasterRpcServices((MasterRpcServices)rpcServices);
     this.loadBalancerTracker = new LoadBalancerTracker(zooKeeper, this);
     this.loadBalancerTracker.start();
 
+    this.normalizer = RegionNormalizerFactory.getRegionNormalizer(conf);
+    this.normalizer.setMasterServices(this);
     this.regionNormalizerTracker = new RegionNormalizerTracker(zooKeeper, this);
     this.regionNormalizerTracker.start();
 
@@ -1857,17 +1861,16 @@ public List<RegionPlan> executeRegionPlansWithThrottling(List<RegionPlan> plans)
   }
 
   @Override
-  @VisibleForTesting
   public RegionNormalizer getRegionNormalizer() {
     return this.normalizer;
   }
 
   /**
    * Perform normalization of cluster (invoked by {@link RegionNormalizerChore}).
    *
-   * @return true if normalization step was performed successfully, false otherwise
-   *    (specifically, if HMaster hasn't been initialized properly or normalization
-   *    is globally disabled)
+   * @return true if an existing normalization was already in progress, or if a new normalization
+   *   was performed successfully; false otherwise (specifically, if HMaster finished initializing
+   *   or normalization is globally disabled).
    */
   public boolean normalizeRegions() throws IOException {
     if (regionNormalizerTracker == null || !regionNormalizerTracker.isNormalizerOn()) {
@@ -1881,34 +1884,42 @@ public boolean normalizeRegions() throws IOException {
       return false;
     }
 
-    synchronized (this.normalizer) {
+    if (!normalizationInProgressLock.tryLock()) {
       // Don't run the normalizer concurrently
+      LOG.info("Normalization already in progress. Skipping request.");
+      return true;
+    }
 
-      List<TableName> allEnabledTables = new ArrayList<>(
-        this.tableStateManager.getTablesInStates(TableState.State.ENABLED));
-
+    try {
+      final List<TableName> allEnabledTables = new ArrayList<>(
+        tableStateManager.getTablesInStates(TableState.State.ENABLED));
       Collections.shuffle(allEnabledTables);
 
-      for (TableName table : allEnabledTables) {
-        TableDescriptor tblDesc = getTableDescriptors().get(table);
-        if (table.isSystemTable() || (tblDesc != null &&
-            !tblDesc.isNormalizationEnabled())) {
-          LOG.trace("Skipping normalization for {}, as it's either system"
-              + " table or doesn't have auto normalization turned on", table);
-          continue;
-        }
+      try (final Admin admin = clusterConnection.getAdmin()) {
+        for (TableName table : allEnabledTables) {
+          if (table.isSystemTable()) {
+            continue;
+          }
+          final TableDescriptor tblDesc = getTableDescriptors().get(table);
+          if (tblDesc != null && !tblDesc.isNormalizationEnabled()) {
+            LOG.debug("Skipping table {} because normalization is disabled in its"
+              + " table properties.", table);
+            continue;
+          }
 
-        // make one last check that the cluster isn't shutting down before proceeding.
-        if (skipRegionManagementAction("region normalizer")) {
-          return false;
-        }
+          // make one last check that the cluster isn't shutting down before proceeding.
+          if (skipRegionManagementAction("region normalizer")) {
+            return false;
+          }
 
-        final List<NormalizationPlan> plans = this.normalizer.computePlanForTable(table);
-        if (CollectionUtils.isEmpty(plans)) {
-          return true;
-        }
+          final List<NormalizationPlan> plans = normalizer.computePlansForTable(table);
+          if (CollectionUtils.isEmpty(plans)) {
+            LOG.debug("No normalization required for table {}.", table);
+            continue;
+          }
 
-        try (final Admin admin = clusterConnection.getAdmin()) {
+          // as of this writing, `plan.execute()` is non-blocking, so there's no artificial rate-
+          // limiting of merge requests due to this serial loop.
           for (NormalizationPlan plan : plans) {
             plan.execute(admin);
             if (plan.getType() == PlanType.SPLIT) {
@@ -1919,9 +1930,9 @@ public boolean normalizeRegions() throws IOException {
           }
         }
       }
+    } finally {
+      normalizationInProgressLock.unlock();
     }
-    // If Region did not generate any plans, it means the cluster is already balanced.
-    // Return true indicating a success.
     return true;
   }