From f1b6c6d317ec0b18ae330da40e6aa97277c6cb1a Mon Sep 17 00:00:00 2001
From: Michael Stack <stack@apache.org>
Date: Thu, 14 Mar 2013 17:10:26 +0000
Subject: [PATCH] HBASE-7876 Got exception when manually triggers a split on an
 empty region

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1456543 13f79535-47bb-0310-9956-ffa450edef68
---
 .../hbase/regionserver/SplitTransaction.java  | 11 ++---
 .../TestSplitTransactionOnCluster.java        | 45 -------------------
 2 files changed, 4 insertions(+), 52 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SplitTransaction.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SplitTransaction.java
index eb3ed79be39..e09024d63ae 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SplitTransaction.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SplitTransaction.java
@@ -287,13 +287,6 @@ public class SplitTransaction {
       if (exceptionToThrow instanceof IOException) throw (IOException)exceptionToThrow;
       throw new IOException(exceptionToThrow);
     }
-
-
-    if (hstoreFilesToSplit.size() == 0) {
-      String errorMsg = "No store files to split for the region "+this.parent.getRegionInfo();
-      LOG.error(errorMsg);
-      throw new IOException(errorMsg);
-    }
     if (!testing) {
       services.removeFromOnlineRegions(this.parent, null);
     }
@@ -607,6 +600,10 @@ public class SplitTransaction {
     // there's files to split. It then fires up everything, waits for
     // completion and finally checks for any exception
     int nbFiles = hstoreFilesToSplit.size();
+    if (nbFiles == 0) {
+      // no file needs to be splitted.
+      return;
+    }
     ThreadFactoryBuilder builder = new ThreadFactoryBuilder();
     builder.setNameFormat("StoreFileSplitter-%1$d");
     ThreadFactory factory = builder.build();
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestSplitTransactionOnCluster.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestSplitTransactionOnCluster.java
index c9f3898af41..c33aa6b8d99 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestSplitTransactionOnCluster.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestSplitTransactionOnCluster.java
@@ -641,51 +641,6 @@ public class TestSplitTransactionOnCluster {
     admin.flush(tableName);
   }
 
-  @Test
-  public void testShouldThrowIOExceptionIfStoreFileSizeIsEmptyAndShouldSuccessfullyExecuteRollback()
-      throws Exception {
-    final byte[] tableName = Bytes.toBytes("testRollBackShudBeSuccessfulIfStoreFileIsEmpty");
-    // Create table then get the single region for our new table.
-    createTableAndWait(tableName, HConstants.CATALOG_FAMILY);
-    List<HRegion> regions = cluster.getRegions(tableName);
-    HRegionInfo hri = getAndCheckSingleTableRegion(regions);
-    int tableRegionIndex = ensureTableRegionNotOnSameServerAsMeta(admin, hri);
-    int regionServerIndex = cluster.getServerWith(regions.get(0).getRegionName());
-    HRegionServer regionServer = cluster.getRegionServer(regionServerIndex);
-    // Turn off balancer so it doesn't cut in and mess up our placements.
-    this.admin.setBalancerRunning(false, true);
-    // Turn off the meta scanner so it don't remove parent on us.
-    cluster.getMaster().setCatalogJanitorEnabled(false);
-    try {
-      printOutRegions(regionServer, "Initial regions: ");
-
-      // find a splittable region.  Refresh the regions list
-      regions = cluster.getRegions(tableName);
-      final HRegion region = findSplittableRegion(regions);
-      assertTrue("not able to find a splittable region", region != null);
-
-      // Now split.
-      SplitTransaction st = new MockedSplitTransaction(region, Bytes.toBytes("row2"));
-      try {
-        st.prepare();
-        st.execute(regionServer, regionServer);
-      } catch (IOException e) {
-        List<HRegion> daughters = cluster.getRegions(tableName);
-        assertTrue(daughters.size() == 1);
-
-        String node = ZKAssign.getNodeName(regionServer.getZooKeeper(),
-            region.getRegionInfo().getEncodedName());
-        assertFalse(ZKUtil.checkExists(regionServer.getZooKeeper(), node) == -1);
-        assertTrue(st.rollback(regionServer, regionServer));
-        assertTrue(ZKUtil.checkExists(regionServer.getZooKeeper(), node) == -1);
-      }
-    } finally {
-      admin.setBalancerRunning(true, false);
-      cluster.getMaster().setCatalogJanitorEnabled(true);
-    }
-
-  }
-
   private void testSplitBeforeSettingSplittingInZKInternals() throws Exception {
     final byte[] tableName = Bytes.toBytes("testSplitBeforeSettingSplittingInZK");
     try {