HBASE-5009 Failure of creating split dir if it already exists prevents splits from happening further

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1225153 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -843,8 +843,9 @@ Release 0.90.6 - Unreleased
   BUG FIXES
    HBASE-4970  Add a parameter so that keepAliveTime of Htable thread pool can be changed (gaojinchao)
    HBASE-5060  HBase client is blocked forever (Jinchao)
+   HBASE-5009  Failure of creating split dir if it already exists prevents splits from happening further
 
-Release 0.90.5 - Unreleased
+Release 0.90.5 - Released
 
   BUG FIXES
    HBASE-4160  HBase shell move and online may be unusable if region name

src/main/java/org/apache/hadoop/hbase/io/Reference.java

@@ -125,8 +125,7 @@ public class Reference implements Writable {
 
   public Path write(final FileSystem fs, final Path p)
   throws IOException {
-    FSUtils.create(fs, p);
-    FSDataOutputStream out = fs.create(p);
+    FSDataOutputStream out = fs.create(p, false);
     try {
       write(out);
     } finally {
@@ -153,4 +152,4 @@ public class Reference implements Writable {
       in.close();
     }
   }
-}
+}

src/main/java/org/apache/hadoop/hbase/regionserver/SplitTransaction.java

@@ -529,7 +529,14 @@ public class SplitTransaction {
    */
   private static void createSplitDir(final FileSystem fs, final Path splitdir)
   throws IOException {
-    if (fs.exists(splitdir)) throw new IOException("Splitdir already exits? " + splitdir);
+    if (fs.exists(splitdir)) {
+      LOG.info("The " + splitdir
+          + " directory exists.  Hence deleting it to recreate it");
+      if (!fs.delete(splitdir, true)) {
+        throw new IOException("Failed deletion of " + splitdir
+            + " before creating them again.");
+      }
+    }
     if (!fs.mkdirs(splitdir)) throw new IOException("Failed create of " + splitdir);
   }
 
@@ -589,6 +596,10 @@ public class SplitTransaction {
           this.fileSplitTimeout, TimeUnit.MILLISECONDS);
       if (stillRunning) {
         threadPool.shutdownNow();
+        // wait for the thread to shutdown completely.
+        while (!threadPool.isTerminated()) {
+          Thread.sleep(50);
+        }
         throw new IOException("Took too long to split the" +
             " files and create the references, aborting split");
       }