Use a single setting to limit normalizer plans size

Author: charlesconnell

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

@@ -58,6 +58,15 @@ public List<NormalizationTarget> getNormalizationTargets() {
     return normalizationTargets;
   }
 
+  @Override
+  public long getPlanSizeMb() {
+    long total = 0;
+    for (NormalizationTarget target : normalizationTargets) {
+      total += target.getRegionSizeMb();
+    }
+    return total;
+  }
+
   @Override
   public String toString() {
     return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)

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

@@ -34,4 +34,6 @@ enum PlanType {
 
   /** Returns the type of this plan */
   PlanType getType();
+
+  long getPlanSizeMb();
 }

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

@@ -22,7 +22,6 @@
 import java.time.Instant;
 import java.time.Period;
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
@@ -80,10 +79,8 @@ class SimpleRegionNormalizer implements RegionNormalizer, ConfigurationObserver
   static final int DEFAULT_MERGE_MIN_REGION_AGE_DAYS = 3;
   static final String MERGE_MIN_REGION_SIZE_MB_KEY = "hbase.normalizer.merge.min_region_size.mb";
   static final int DEFAULT_MERGE_MIN_REGION_SIZE_MB = 0;
-  static final String CUMULATIVE_MERGE_SIZE_LIMIT_MB_KEY = "hbase.normalizer.merge.plans_size_limit.mb";
-  static final long DEFAULT_CUMULATIVE_MERGE_SIZE_LIMIT_MB = Long.MAX_VALUE;
-  static final String CUMULATIVE_SPLIT_SIZE_LIMIT_MB_KEY = "hbase.normalizer.split.plans_size_limit.mb";
-  static final long DEFAULT_CUMULATIVE_SPLIT_SIZE_LIMIT_MB = Long.MAX_VALUE;
+  static final String CUMULATIVE_SIZE_LIMIT_MB_KEY = "hbase.normalizer.plans_size_limit.mb";
+  static final long DEFAULT_CUMULATIVE_SIZE_LIMIT_MB = Long.MAX_VALUE;
 
   private MasterServices masterServices;
   private NormalizerConfiguration normalizerConfiguration;
@@ -220,7 +217,7 @@ public List<NormalizationPlan> computePlansForTable(final TableDescriptor tableD
     LOG.debug("Computing normalization plan for table:  {}, number of regions: {}", table,
       ctx.getTableRegions().size());
 
-    final List<NormalizationPlan> plans = new ArrayList<>();
+    List<NormalizationPlan> plans = new ArrayList<>();
     int splitPlansCount = 0;
     if (proceedWithSplitPlanning) {
       List<NormalizationPlan> splitPlans = computeSplitNormalizationPlans(ctx);
@@ -234,19 +231,13 @@ public List<NormalizationPlan> computePlansForTable(final TableDescriptor tableD
       plans.addAll(mergePlans);
     }
 
+    plans = truncateForSize(plans);
+
     LOG.debug("Computed normalization plans for table {}. Total plans: {}, split plans: {}, "
       + "merge plans: {}", table, plans.size(), splitPlansCount, mergePlansCount);
     return plans;
   }
 
-  private long getTotalRegionSizeMB(Collection<NormalizationTarget> targets) {
-    long total = 0;
-    for (NormalizationTarget target : targets) {
-      total += target.getRegionSizeMb();
-    }
-    return total;
-  }
-
   /** Returns size of region in MB and if region is not found than -1 */
   private long getRegionSizeMB(RegionInfo hri) {
     ServerName sn =
@@ -370,7 +361,6 @@ private List<NormalizationPlan> computeMergeNormalizationPlans(final NormalizeCo
     final List<NormalizationPlan> plans = new LinkedList<>();
     final List<NormalizationTarget> rangeMembers = new LinkedList<>();
     long sumRangeMembersSizeMb;
-    long cumulativePlansSizeMb = 0;
     int current = 0;
     for (int rangeStart = 0; rangeStart < ctx.getTableRegions().size() - 1
       && current < ctx.getTableRegions().size();) {
@@ -408,12 +398,7 @@ private List<NormalizationPlan> computeMergeNormalizationPlans(final NormalizeCo
         break;
       }
       if (rangeMembers.size() > 1) {
-        cumulativePlansSizeMb += getTotalRegionSizeMB(rangeMembers);
-        if (cumulativePlansSizeMb > normalizerConfiguration.getCumulativeMergePlansSizeLimitMb()) {
-          return plans;
-        } else {
-          plans.add(new MergeNormalizationPlan.Builder().setTargets(rangeMembers).build());
-        }
+        plans.add(new MergeNormalizationPlan.Builder().setTargets(rangeMembers).build());
       }
     }
     return plans;
@@ -441,7 +426,6 @@ private List<NormalizationPlan> computeSplitNormalizationPlans(final NormalizeCo
     LOG.debug("Table {}, average region size: {} MB", ctx.getTableName(),
       String.format("%.3f", avgRegionSize));
 
-    long cumulativePlansSizeMb = 0;
     final List<NormalizationPlan> plans = new ArrayList<>();
     for (final RegionInfo hri : ctx.getTableRegions()) {
       if (skipForSplit(ctx.getRegionStates().getRegionState(hri), hri)) {
@@ -454,12 +438,7 @@ private List<NormalizationPlan> computeSplitNormalizationPlans(final NormalizeCo
             + "splitting",
           ctx.getTableName(), hri.getRegionNameAsString(), regionSizeMb,
           String.format("%.3f", avgRegionSize));
-        cumulativePlansSizeMb += getRegionSizeMB(hri);
-        if (cumulativePlansSizeMb > normalizerConfiguration.getCumulativeSplitPlansSizeLimitMb()) {
-          return plans;
-        } else {
-          plans.add(new SplitNormalizationPlan(hri, regionSizeMb));
-        }
+        plans.add(new SplitNormalizationPlan(hri, regionSizeMb));
       }
     }
     return plans;
@@ -489,6 +468,27 @@ private boolean isLargeEnoughForMerge(final NormalizerConfiguration normalizerCo
     return getRegionSizeMB(regionInfo) >= normalizerConfiguration.getMergeMinRegionSizeMb(ctx);
   }
 
+  private List<NormalizationPlan> truncateForSize(List<NormalizationPlan> plans) {
+    if (
+      normalizerConfiguration.getCumulativePlansSizeLimitMb() != DEFAULT_CUMULATIVE_SIZE_LIMIT_MB
+    ) {
+      // If we are going to truncate our list of plans, shuffle the split and merge plans together
+      // so that the merge plans, which are listed last, are not starved out.
+      List<NormalizationPlan> maybeTruncatedPlans = new ArrayList<>();
+      Collections.shuffle(plans);
+      long cumulativeSizeMb = 0;
+      for (NormalizationPlan plan : plans) {
+        cumulativeSizeMb += plan.getPlanSizeMb();
+        if (cumulativeSizeMb < normalizerConfiguration.getCumulativePlansSizeLimitMb()) {
+          maybeTruncatedPlans.add(plan);
+        }
+      }
+      return maybeTruncatedPlans;
+    } else {
+      return plans;
+    }
+  }
+
   private static boolean logTraceReason(final BooleanSupplier predicate, final String fmtWhenTrue,
     final Object... args) {
     final boolean value = predicate.getAsBoolean();
@@ -509,8 +509,7 @@ private static final class NormalizerConfiguration {
     private final int mergeMinRegionCount;
     private final Period mergeMinRegionAge;
     private final long mergeMinRegionSizeMb;
-    private final long cumulativeMergePlansSizeLimitMb;
-    private final long cumulativeSplitPlansSizeLimitMb;
+    private final long cumulativePlansSizeLimitMb;
 
     private NormalizerConfiguration() {
       conf = null;
@@ -519,8 +518,7 @@ private NormalizerConfiguration() {
       mergeMinRegionCount = DEFAULT_MERGE_MIN_REGION_COUNT;
       mergeMinRegionAge = Period.ofDays(DEFAULT_MERGE_MIN_REGION_AGE_DAYS);
       mergeMinRegionSizeMb = DEFAULT_MERGE_MIN_REGION_SIZE_MB;
-      cumulativeMergePlansSizeLimitMb = DEFAULT_CUMULATIVE_MERGE_SIZE_LIMIT_MB;
-      cumulativeSplitPlansSizeLimitMb = DEFAULT_CUMULATIVE_SPLIT_SIZE_LIMIT_MB;
+      cumulativePlansSizeLimitMb = DEFAULT_CUMULATIVE_SIZE_LIMIT_MB;
     }
 
     private NormalizerConfiguration(final Configuration conf,
@@ -531,8 +529,8 @@ private NormalizerConfiguration(final Configuration conf,
       mergeMinRegionCount = parseMergeMinRegionCount(conf);
       mergeMinRegionAge = parseMergeMinRegionAge(conf);
       mergeMinRegionSizeMb = parseMergeMinRegionSizeMb(conf);
-      cumulativeSplitPlansSizeLimitMb = conf.getLong(CUMULATIVE_SPLIT_SIZE_LIMIT_MB_KEY, DEFAULT_CUMULATIVE_MERGE_SIZE_LIMIT_MB);
-      cumulativeMergePlansSizeLimitMb = conf.getLong(CUMULATIVE_MERGE_SIZE_LIMIT_MB_KEY, DEFAULT_CUMULATIVE_MERGE_SIZE_LIMIT_MB);
+      cumulativePlansSizeLimitMb =
+        conf.getLong(CUMULATIVE_SIZE_LIMIT_MB_KEY, DEFAULT_CUMULATIVE_SIZE_LIMIT_MB);
       logConfigurationUpdated(SPLIT_ENABLED_KEY, currentConfiguration.isSplitEnabled(),
         splitEnabled);
       logConfigurationUpdated(MERGE_ENABLED_KEY, currentConfiguration.isMergeEnabled(),
@@ -543,8 +541,8 @@ private NormalizerConfiguration(final Configuration conf,
         currentConfiguration.getMergeMinRegionAge(), mergeMinRegionAge);
       logConfigurationUpdated(MERGE_MIN_REGION_SIZE_MB_KEY,
         currentConfiguration.getMergeMinRegionSizeMb(), mergeMinRegionSizeMb);
-      logConfigurationUpdated(CUMULATIVE_SPLIT_SIZE_LIMIT_MB_KEY, currentConfiguration.getCumulativeSplitPlansSizeLimitMb(), cumulativeSplitPlansSizeLimitMb);
-      logConfigurationUpdated(CUMULATIVE_MERGE_SIZE_LIMIT_MB_KEY, currentConfiguration.getCumulativeMergePlansSizeLimitMb(), cumulativeMergePlansSizeLimitMb);
+      logConfigurationUpdated(CUMULATIVE_SIZE_LIMIT_MB_KEY,
+        currentConfiguration.getCumulativePlansSizeLimitMb(), cumulativePlansSizeLimitMb);
     }
 
     public Configuration getConf() {
@@ -608,12 +606,8 @@ public long getMergeMinRegionSizeMb(NormalizeContext context) {
       return mergeMinRegionSizeMb;
     }
 
-    public long getCumulativeSplitPlansSizeLimitMb() {
-      return cumulativeSplitPlansSizeLimitMb;
-    }
-
-    public long getCumulativeMergePlansSizeLimitMb() {
-      return cumulativeMergePlansSizeLimitMb;
+    public long getCumulativePlansSizeLimitMb() {
+      return cumulativePlansSizeLimitMb;
     }
   }
 

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

@@ -45,6 +45,11 @@ public NormalizationTarget getSplitTarget() {
     return splitTarget;
   }
 
+  @Override
+  public long getPlanSizeMb() {
+    return splitTarget.getRegionSizeMb();
+  }
+
   @Override
   public String toString() {
     return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)

hbase-server/src/test/java/org/apache/hadoop/hbase/master/normalizer/TestSimpleRegionNormalizer.java

@@ -18,8 +18,7 @@
 package org.apache.hadoop.hbase.master.normalizer;
 
 import static java.lang.String.format;
-import static org.apache.hadoop.hbase.master.normalizer.SimpleRegionNormalizer.CUMULATIVE_MERGE_SIZE_LIMIT_MB_KEY;
-import static org.apache.hadoop.hbase.master.normalizer.SimpleRegionNormalizer.CUMULATIVE_SPLIT_SIZE_LIMIT_MB_KEY;
+import static org.apache.hadoop.hbase.master.normalizer.SimpleRegionNormalizer.CUMULATIVE_SIZE_LIMIT_MB_KEY;
 import static org.apache.hadoop.hbase.master.normalizer.SimpleRegionNormalizer.DEFAULT_MERGE_MIN_REGION_AGE_DAYS;
 import static org.apache.hadoop.hbase.master.normalizer.SimpleRegionNormalizer.MERGE_ENABLED_KEY;
 import static org.apache.hadoop.hbase.master.normalizer.SimpleRegionNormalizer.MERGE_MIN_REGION_AGE_DAYS_KEY;
@@ -613,35 +612,34 @@ public void testNormalizerCannotMergeNonAdjacentRegions() {
   @Test
   public void testMergeSizeLimit() {
     conf.setBoolean(SPLIT_ENABLED_KEY, false);
-    conf.setLong(CUMULATIVE_MERGE_SIZE_LIMIT_MB_KEY, 5);
+    conf.setLong(CUMULATIVE_SIZE_LIMIT_MB_KEY, 5);
     final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 6);
-    final Map<byte[], Integer> regionSizes =
-      createRegionSizesMap(regionInfos,1, 1, 1, 1, 1, 1);
+    final Map<byte[], Integer> regionSizes = createRegionSizesMap(regionInfos, 1, 1, 1, 1, 1, 1);
     setupMocksForNormalizer(regionSizes, regionInfos);
     when(tableDescriptor.getNormalizerTargetRegionSize()).thenReturn(2L);
 
     assertTrue(normalizer.isMergeEnabled());
     assertFalse(normalizer.isSplitEnabled());
-    assertThat(normalizer.computePlansForTable(tableDescriptor),
-      hasSize(2)); // creates 2 merge plans, even though there are 3 otherwise eligible pairs of regions
+    // creates 2 merge plans, even though there are 3 otherwise eligible pairs of regions
+    assertThat(normalizer.computePlansForTable(tableDescriptor), hasSize(2));
   }
 
   @Test
   public void testSplitSizeLimit() {
     conf.setBoolean(MERGE_ENABLED_KEY, false);
-    conf.setLong(CUMULATIVE_SPLIT_SIZE_LIMIT_MB_KEY, 10);
+    conf.setLong(CUMULATIVE_SIZE_LIMIT_MB_KEY, 10);
     final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 4);
-    final Map<byte[], Integer> regionSizes =
-      createRegionSizesMap(regionInfos, 3, 3, 3, 3);
+    final Map<byte[], Integer> regionSizes = createRegionSizesMap(regionInfos, 3, 3, 3, 3);
     setupMocksForNormalizer(regionSizes, regionInfos);
     when(tableDescriptor.getNormalizerTargetRegionSize()).thenReturn(1L);
 
     assertTrue(normalizer.isSplitEnabled());
     assertFalse(normalizer.isMergeEnabled());
-    assertThat(normalizer.computePlansForTable(tableDescriptor),
-      hasSize(3));  // the plan includes only 3 regions, even though the table has 4 otherwise split-eligible regions
+    // the plan includes only 3 regions, even though the table has 4 otherwise split-eligible
+    // regions
+    assertThat(normalizer.computePlansForTable(tableDescriptor), hasSize(3));
   }
 
   @SuppressWarnings("MockitoCast")