HBASE-24583 Normalizer can't actually merge empty regions...

when neighbor is larger than average size

* add `testMergeEmptyRegions` to explicitly cover different
  interleaving of 0-sized regions.
* fix bug where merging a 0-size region is skipped due to large
  neighbor.
* remove unused `splitPoint` from `SplitNormalizationPlan`.
* generate `toString`, `hashCode`, and `equals` methods from Apache
  Commons Lang3 template on `SplitNormalizationPlan` and
  `MergeNormalizationPlan`.
* simplify test to use equality matching over `*NormalizationPlan`
  instances as plain pojos.
* test make use of this handy `TableNameTestRule`.
* fix line-length issues in `TestSimpleRegionNormalizer`

Signed-off-by: Wellington Chevreuil <wchevreuil@apache.org>
Signed-off-by: Viraj Jasani <vjasani@apache.org>
Signed-off-by: huaxiangsun <huaxiangsun@apache.org>
Signed-off-by: Aman Poonia <aman.poonia.29@gmail.com>

hbase-common/src/test/java/org/apache/hadoop/hbase/TableNameTestRule.java

@@ -21,7 +21,8 @@ import org.junit.rules.TestWatcher;
 import org.junit.runner.Description;
 
 /**
- * Returns a {@code TableName} based on currently running test method name.
+ * Returns a {@code TableName} based on currently running test method name. Supports
+ *  tests built on the {@link org.junit.runners.Parameterized} runner.
  */
 public class TableNameTestRule extends TestWatcher {
 

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

@@ -1,4 +1,4 @@
-/**
+/*
  *
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
@@ -19,20 +19,20 @@
 package org.apache.hadoop.hbase.master.normalizer;
 
 import java.io.IOException;
-
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.master.MasterServices;
 import org.apache.yetus.audience.InterfaceAudience;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * Normalization plan to merge regions (smallest region in the table with its smallest neighbor).
  */
 @InterfaceAudience.Private
 public class MergeNormalizationPlan implements NormalizationPlan {
-  private static final Logger LOG = LoggerFactory.getLogger(MergeNormalizationPlan.class.getName());
 
   private final RegionInfo firstRegion;
   private final RegionInfo secondRegion;
@@ -47,7 +47,6 @@ public class MergeNormalizationPlan implements NormalizationPlan {
    */
   @Override
   public long submit(MasterServices masterServices) throws IOException {
-    LOG.info("Executing merging normalization plan: " + this);
     // 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.
@@ -71,10 +70,35 @@ public class MergeNormalizationPlan implements NormalizationPlan {
 
   @Override
   public String toString() {
-    return "MergeNormalizationPlan{" +
+    return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)
-      "firstRegion=" + firstRegion +
+      .append("firstRegion", firstRegion)
-      ", secondRegion=" + secondRegion +
+      .append("secondRegion", secondRegion)
-      '}';
+      .toString();
   }
 
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    MergeNormalizationPlan that = (MergeNormalizationPlan) o;
+
+    return new EqualsBuilder()
+      .append(firstRegion, that.firstRegion)
+      .append(secondRegion, that.secondRegion)
+      .isEquals();
+  }
+
+  @Override
+  public int hashCode() {
+    return new HashCodeBuilder(17, 37)
+      .append(firstRegion)
+      .append(secondRegion)
+      .toHashCode();
+  }
 }

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

@@ -369,7 +369,8 @@ public class SimpleRegionNormalizer implements RegionNormalizer {
       }
       final long currentSizeMb = getRegionSizeMB(current);
       final long nextSizeMb = getRegionSizeMB(next);
-      if (currentSizeMb + nextSizeMb < avgRegionSizeMb) {
+      // always merge away empty regions when they present themselves.
+      if (currentSizeMb == 0 || nextSizeMb == 0 || currentSizeMb + nextSizeMb < avgRegionSizeMb) {
         plans.add(new MergeNormalizationPlan(current, next));
         candidateIdx++;
       }
@@ -411,7 +412,7 @@ public class SimpleRegionNormalizer implements RegionNormalizer {
       if (regionSize > 2 * avgRegionSize) {
         LOG.info("Table {}, large region {} has size {}, more than twice avg size {}, splitting",
           ctx.getTableName(), hri.getRegionNameAsString(), regionSize, avgRegionSize);
-        plans.add(new SplitNormalizationPlan(hri, null));
+        plans.add(new SplitNormalizationPlan(hri));
       }
     }
     return plans;

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

@@ -1,4 +1,4 @@
-/**
+/*
  *
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
@@ -19,35 +19,31 @@
 package org.apache.hadoop.hbase.master.normalizer;
 
 import java.io.IOException;
-import java.util.Arrays;
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.master.MasterServices;
 import org.apache.yetus.audience.InterfaceAudience;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * Normalization plan to split region.
  */
 @InterfaceAudience.Private
 public class SplitNormalizationPlan implements NormalizationPlan {
-  private static final Logger LOG = LoggerFactory.getLogger(SplitNormalizationPlan.class.getName());
 
-  private RegionInfo regionInfo;
+  private final RegionInfo regionInfo;
-  private byte[] splitPoint;
 
-  public SplitNormalizationPlan(RegionInfo regionInfo, byte[] splitPoint) {
+  public SplitNormalizationPlan(RegionInfo regionInfo) {
     this.regionInfo = regionInfo;
-    this.splitPoint = splitPoint;
   }
 
-  /**
-   * {@inheritDoc}
-   */
   @Override
   public long submit(MasterServices masterServices) throws IOException {
-    return masterServices.splitRegion(regionInfo, null, HConstants.NO_NONCE, HConstants.NO_NONCE);
+    return masterServices.splitRegion(regionInfo, null, HConstants.NO_NONCE,
+      HConstants.NO_NONCE);
   }
 
   @Override
@@ -59,24 +55,33 @@ public class SplitNormalizationPlan implements NormalizationPlan {
     return regionInfo;
   }
 
-  public void setRegionInfo(RegionInfo regionInfo) {
+  @Override
-    this.regionInfo = regionInfo;
+  public String toString() {
-  }
+    return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)
-
+      .append("regionInfo", regionInfo)
-  public byte[] getSplitPoint() {
+      .toString();
-    return splitPoint;
-  }
-
-  public void setSplitPoint(byte[] splitPoint) {
-    this.splitPoint = splitPoint;
   }
 
   @Override
-  public String toString() {
+  public boolean equals(Object o) {
-    return "SplitNormalizationPlan{" +
+    if (this == o) {
-      "regionInfo=" + regionInfo +
+      return true;
-      ", splitPoint=" + Arrays.toString(splitPoint) +
+    }
-      '}';
+
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    SplitNormalizationPlan that = (SplitNormalizationPlan) o;
+
+    return new EqualsBuilder()
+      .append(regionInfo, that.regionInfo)
+      .isEquals();
   }
 
+  @Override public int hashCode() {
+    return new HashCodeBuilder(17, 37)
+      .append(regionInfo)
+      .toHashCode();
+  }
 }

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

@@ -30,11 +30,9 @@ import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.everyItem;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.instanceOf;
-import static org.hamcrest.Matchers.iterableWithSize;
 import static org.hamcrest.Matchers.not;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
 import static org.mockito.Mockito.when;
@@ -52,6 +50,7 @@ import org.apache.hadoop.hbase.RegionMetrics;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.Size;
 import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.TableNameTestRule;
 import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.client.RegionInfoBuilder;
 import org.apache.hadoop.hbase.master.MasterServices;
@@ -65,7 +64,6 @@ import org.junit.ClassRule;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
-import org.junit.rules.TestName;
 import org.mockito.Mockito;
 
 /**
@@ -83,7 +81,7 @@ public class TestSimpleRegionNormalizer {
   private MasterServices masterServices;
 
   @Rule
-  public TestName name = new TestName();
+  public TableNameTestRule name = new TableNameTestRule();
 
   @Before
   public void before() {
@@ -103,7 +101,7 @@ public class TestSimpleRegionNormalizer {
 
   @Test
   public void testNoNormalizationIfTooFewRegions() {
-    final TableName tableName = TableName.valueOf(name.getMethodName());
+    final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 2);
     final Map<byte[], Integer> regionSizes = createRegionSizesMap(regionInfos, 10, 15);
     setupMocksForNormalizer(regionSizes, regionInfos);
@@ -114,9 +112,10 @@ public class TestSimpleRegionNormalizer {
 
   @Test
   public void testNoNormalizationOnNormalizedCluster() {
-    final TableName tableName = TableName.valueOf(name.getMethodName());
+    final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 4);
-    final Map<byte[], Integer> regionSizes = createRegionSizesMap(regionInfos, 10, 15, 8, 10);
+    final Map<byte[], Integer> regionSizes =
+      createRegionSizesMap(regionInfos, 10, 15, 8, 10);
     setupMocksForNormalizer(regionSizes, regionInfos);
 
     List<NormalizationPlan> plans = normalizer.computePlansForTable(tableName);
@@ -124,7 +123,7 @@ public class TestSimpleRegionNormalizer {
   }
 
   private void noNormalizationOnTransitioningRegions(final RegionState.State state) {
-    final TableName tableName = TableName.valueOf(name.getMethodName());
+    final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 3);
     final Map<byte[], Integer> regionSizes = createRegionSizesMap(regionInfos, 10, 1, 100);
 
@@ -170,38 +169,33 @@ public class TestSimpleRegionNormalizer {
 
   @Test
   public void testMergeOfSmallRegions() {
-    final TableName tableName = TableName.valueOf(name.getMethodName());
+    final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 5);
     final Map<byte[], Integer> regionSizes =
       createRegionSizesMap(regionInfos, 15, 5, 5, 15, 16);
     setupMocksForNormalizer(regionSizes, regionInfos);
 
-    List<NormalizationPlan> plans = normalizer.computePlansForTable(tableName);
+    assertThat(normalizer.computePlansForTable(tableName), contains(
-    assertThat(plans.get(0), instanceOf(MergeNormalizationPlan.class));
+      new MergeNormalizationPlan(regionInfos.get(1), regionInfos.get(2))));
-    MergeNormalizationPlan plan = (MergeNormalizationPlan) plans.get(0);
-    assertEquals(regionInfos.get(1), plan.getFirstRegion());
-    assertEquals(regionInfos.get(2), plan.getSecondRegion());
   }
 
   // Test for situation illustrated in HBASE-14867
   @Test
   public void testMergeOfSecondSmallestRegions() {
-    final TableName tableName = TableName.valueOf(name.getMethodName());
+    final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 6);
     final Map<byte[], Integer> regionSizes =
       createRegionSizesMap(regionInfos, 1, 10000, 10000, 10000, 2700, 2700);
     setupMocksForNormalizer(regionSizes, regionInfos);
 
-    List<NormalizationPlan> plans = normalizer.computePlansForTable(tableName);
+    assertThat(normalizer.computePlansForTable(tableName), contains(
-    assertThat(plans.get(0), instanceOf(MergeNormalizationPlan.class));
+      new MergeNormalizationPlan(regionInfos.get(4), regionInfos.get(5))
-    MergeNormalizationPlan plan = (MergeNormalizationPlan) plans.get(0);
+    ));
-    assertEquals(regionInfos.get(4), plan.getFirstRegion());
-    assertEquals(regionInfos.get(5), plan.getSecondRegion());
   }
 
   @Test
   public void testMergeOfSmallNonAdjacentRegions() {
-    final TableName tableName = TableName.valueOf(name.getMethodName());
+    final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 5);
     final Map<byte[], Integer> regionSizes =
       createRegionSizesMap(regionInfos, 15, 5, 16, 15, 5);
@@ -213,21 +207,19 @@ public class TestSimpleRegionNormalizer {
 
   @Test
   public void testSplitOfLargeRegion() {
-    final TableName tableName = TableName.valueOf(name.getMethodName());
+    final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 4);
     final Map<byte[], Integer> regionSizes =
       createRegionSizesMap(regionInfos, 8, 6, 10, 30);
     setupMocksForNormalizer(regionSizes, regionInfos);
 
-    List<NormalizationPlan> plans = normalizer.computePlansForTable(tableName);
+    assertThat(normalizer.computePlansForTable(tableName), contains(
-    assertThat(plans.get(0), instanceOf(SplitNormalizationPlan.class));
+      new SplitNormalizationPlan(regionInfos.get(3))));
-    SplitNormalizationPlan plan = (SplitNormalizationPlan) plans.get(0);
-    assertEquals(regionInfos.get(3), plan.getRegionInfo());
   }
 
   @Test
-  public void testSplitWithTargetRegionCount() throws Exception {
+  public void testSplitWithTargetRegionSize() throws Exception {
-    final TableName tableName = TableName.valueOf(name.getMethodName());
+    final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 6);
     final Map<byte[], Integer> regionSizes =
       createRegionSizesMap(regionInfos, 20, 40, 60, 80, 100, 120);
@@ -236,49 +228,47 @@ public class TestSimpleRegionNormalizer {
     // test when target region size is 20
     when(masterServices.getTableDescriptors().get(any()).getNormalizerTargetRegionSize())
         .thenReturn(20L);
-    List<NormalizationPlan> plans = normalizer.computePlansForTable(tableName);
+    assertThat(normalizer.computePlansForTable(tableName), contains(
-    assertThat(plans, iterableWithSize(4));
+      new SplitNormalizationPlan(regionInfos.get(2)),
-    assertThat(plans, everyItem(instanceOf(SplitNormalizationPlan.class)));
+      new SplitNormalizationPlan(regionInfos.get(3)),
+      new SplitNormalizationPlan(regionInfos.get(4)),
+      new SplitNormalizationPlan(regionInfos.get(5))
+    ));
 
     // test when target region size is 200
     when(masterServices.getTableDescriptors().get(any()).getNormalizerTargetRegionSize())
         .thenReturn(200L);
-    plans = normalizer.computePlansForTable(tableName);
+    assertThat(normalizer.computePlansForTable(tableName), contains(
-    assertThat(plans, iterableWithSize(2));
+      new MergeNormalizationPlan(regionInfos.get(0), regionInfos.get(1)),
-    assertTrue(plans.get(0) instanceof MergeNormalizationPlan);
+      new MergeNormalizationPlan(regionInfos.get(2), regionInfos.get(3))));
-    MergeNormalizationPlan plan = (MergeNormalizationPlan) plans.get(0);
-    assertEquals(regionInfos.get(0), plan.getFirstRegion());
-    assertEquals(regionInfos.get(1), plan.getSecondRegion());
   }
 
   @Test
-  public void testSplitWithTargetRegionSize() throws Exception {
+  public void testSplitWithTargetRegionCount() throws Exception {
-    final TableName tableName = TableName.valueOf(name.getMethodName());
+    final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 4);
-    final Map<byte[], Integer> regionSizes = createRegionSizesMap(regionInfos, 20, 40, 60, 80);
+    final Map<byte[], Integer> regionSizes =
+      createRegionSizesMap(regionInfos, 20, 40, 60, 80);
     setupMocksForNormalizer(regionSizes, regionInfos);
 
     // test when target region count is 8
     when(masterServices.getTableDescriptors().get(any()).getNormalizerTargetRegionCount())
         .thenReturn(8);
-    List<NormalizationPlan> plans = normalizer.computePlansForTable(tableName);
+    assertThat(normalizer.computePlansForTable(tableName), contains(
-    assertThat(plans, iterableWithSize(2));
+      new SplitNormalizationPlan(regionInfos.get(2)),
-    assertThat(plans, everyItem(instanceOf(SplitNormalizationPlan.class)));
+      new SplitNormalizationPlan(regionInfos.get(3))));
 
     // test when target region count is 3
     when(masterServices.getTableDescriptors().get(any()).getNormalizerTargetRegionCount())
         .thenReturn(3);
-    plans = normalizer.computePlansForTable(tableName);
+    assertThat(normalizer.computePlansForTable(tableName), contains(
-    assertThat(plans, contains(instanceOf(MergeNormalizationPlan.class)));
+      new MergeNormalizationPlan(regionInfos.get(0), regionInfos.get(1))));
-    MergeNormalizationPlan plan = (MergeNormalizationPlan) plans.get(0);
-    assertEquals(regionInfos.get(0), plan.getFirstRegion());
-    assertEquals(regionInfos.get(1), plan.getSecondRegion());
   }
 
   @Test
   public void testHonorsSplitEnabled() {
     conf.setBoolean(SPLIT_ENABLED_KEY, true);
-    final TableName tableName = TableName.valueOf(name.getMethodName());
+    final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 5);
     final Map<byte[], Integer> regionSizes =
       createRegionSizesMap(regionInfos, 5, 5, 20, 5, 5);
@@ -295,7 +285,7 @@ public class TestSimpleRegionNormalizer {
   @Test
   public void testHonorsMergeEnabled() {
     conf.setBoolean(MERGE_ENABLED_KEY, true);
-    final TableName tableName = TableName.valueOf(name.getMethodName());
+    final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 5);
     final Map<byte[], Integer> regionSizes =
       createRegionSizesMap(regionInfos, 20, 5, 5, 20, 20);
@@ -312,7 +302,7 @@ public class TestSimpleRegionNormalizer {
   @Test
   public void testHonorsMinimumRegionCount() {
     conf.setInt(MIN_REGION_COUNT_KEY, 1);
-    final TableName tableName = TableName.valueOf(name.getMethodName());
+    final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 3);
     // create a table topology that results in both a merge plan and a split plan. Assert that the
     // merge is only created when the when the number of table regions is above the region count
@@ -322,27 +312,20 @@ public class TestSimpleRegionNormalizer {
 
     List<NormalizationPlan> plans = normalizer.computePlansForTable(tableName);
     assertThat(plans, contains(
-      instanceOf(SplitNormalizationPlan.class),
+      new SplitNormalizationPlan(regionInfos.get(2)),
-      instanceOf(MergeNormalizationPlan.class)));
+      new MergeNormalizationPlan(regionInfos.get(0), regionInfos.get(1))));
-    SplitNormalizationPlan splitPlan = (SplitNormalizationPlan) plans.get(0);
-    assertEquals(regionInfos.get(2), splitPlan.getRegionInfo());
-    MergeNormalizationPlan mergePlan = (MergeNormalizationPlan) plans.get(1);
-    assertEquals(regionInfos.get(0), mergePlan.getFirstRegion());
-    assertEquals(regionInfos.get(1), mergePlan.getSecondRegion());
 
     // have to call setupMocks again because we don't have dynamic config update on normalizer.
     conf.setInt(MIN_REGION_COUNT_KEY, 4);
     setupMocksForNormalizer(regionSizes, regionInfos);
-    plans = normalizer.computePlansForTable(tableName);
+    assertThat(normalizer.computePlansForTable(tableName), contains(
-    assertThat(plans, contains(instanceOf(SplitNormalizationPlan.class)));
+      new SplitNormalizationPlan(regionInfos.get(2))));
-    splitPlan = (SplitNormalizationPlan) plans.get(0);
-    assertEquals(regionInfos.get(2), splitPlan.getRegionInfo());
   }
 
   @Test
   public void testHonorsMergeMinRegionAge() {
     conf.setInt(MERGE_MIN_REGION_AGE_DAYS_KEY, 7);
-    final TableName tableName = TableName.valueOf(name.getMethodName());
+    final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 4);
     final Map<byte[], Integer> regionSizes =
       createRegionSizesMap(regionInfos, 1, 1, 10, 10);
@@ -365,19 +348,16 @@ public class TestSimpleRegionNormalizer {
   @Test
   public void testHonorsMergeMinRegionSize() {
     conf.setBoolean(SPLIT_ENABLED_KEY, false);
-    final TableName tableName = TableName.valueOf(name.getMethodName());
+    final TableName tableName = name.getTableName();
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, 5);
-    final Map<byte[], Integer> regionSizes = createRegionSizesMap(regionInfos, 1, 2, 0, 10, 10);
+    final Map<byte[], Integer> regionSizes =
+      createRegionSizesMap(regionInfos, 1, 2, 0, 10, 10);
     setupMocksForNormalizer(regionSizes, regionInfos);
 
     assertFalse(normalizer.isSplitEnabled());
     assertEquals(1, normalizer.getMergeMinRegionSizeMb());
-    final List<NormalizationPlan> plans = normalizer.computePlansForTable(tableName);
+    assertThat(normalizer.computePlansForTable(tableName), contains(
-    assertThat(plans, everyItem(instanceOf(MergeNormalizationPlan.class)));
+      new MergeNormalizationPlan(regionInfos.get(0), regionInfos.get(1))));
-    assertThat(plans, iterableWithSize(1));
-    final MergeNormalizationPlan plan = (MergeNormalizationPlan) plans.get(0);
-    assertEquals(regionInfos.get(0), plan.getFirstRegion());
-    assertEquals(regionInfos.get(1), plan.getSecondRegion());
 
     conf.setInt(MERGE_MIN_REGION_SIZE_MB_KEY, 3);
     setupMocksForNormalizer(regionSizes, regionInfos);
@@ -385,10 +365,28 @@ public class TestSimpleRegionNormalizer {
     assertThat(normalizer.computePlansForTable(tableName), empty());
   }
 
+  @Test
+  public void testMergeEmptyRegions() {
+    conf.setBoolean(SPLIT_ENABLED_KEY, false);
+    conf.setInt(MERGE_MIN_REGION_SIZE_MB_KEY, 0);
+    final TableName tableName = name.getTableName();
+    final List<RegionInfo> regionInfos = createRegionInfos(tableName, 7);
+    final Map<byte[], Integer> regionSizes =
+      createRegionSizesMap(regionInfos, 0, 1, 10, 0, 9, 10, 0);
+    setupMocksForNormalizer(regionSizes, regionInfos);
+
+    assertFalse(normalizer.isSplitEnabled());
+    assertEquals(0, normalizer.getMergeMinRegionSizeMb());
+    assertThat(normalizer.computePlansForTable(tableName), contains(
+      new MergeNormalizationPlan(regionInfos.get(0), regionInfos.get(1)),
+      new MergeNormalizationPlan(regionInfos.get(2), regionInfos.get(3)),
+      new MergeNormalizationPlan(regionInfos.get(5), regionInfos.get(6))));
+  }
+
   // This test is to make sure that normalizer is only going to merge adjacent regions.
   @Test
   public void testNormalizerCannotMergeNonAdjacentRegions() {
-    final TableName tableName = TableName.valueOf(name.getMethodName());
+    final TableName tableName = name.getTableName();
     // create 5 regions with sizes to trigger merge of small regions. region ranges are:
     // [, "aa"), ["aa", "aa1"), ["aa1", "aa1!"), ["aa1!", "aa2"), ["aa2", )
     // Region ["aa", "aa1") and ["aa1!", "aa2") are not adjacent, they are not supposed to
@@ -402,7 +400,8 @@ public class TestSimpleRegionNormalizer {
       null,
     };
     final List<RegionInfo> regionInfos = createRegionInfos(tableName, keys);
-    final Map<byte[], Integer> regionSizes = createRegionSizesMap(regionInfos, 3, 1, 1, 3, 5);
+    final Map<byte[], Integer> regionSizes =
+      createRegionSizesMap(regionInfos, 3, 1, 1, 3, 5);
     setupMocksForNormalizer(regionSizes, regionInfos);
 
     // Compute the plan, no merge plan returned as they are not adjacent.