HBASE-3820 Splitlog() executed while the namenode was in safemode may cause data-loss

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

CHANGES.txt

@@ -266,6 +266,8 @@ Release 0.90.4 - Unreleased
                region assignment
    HBASE-3902  Add Bytes.toBigDecimal and Bytes.toBytes(BigDecimal)
                (Vaibhav Puranik)
+   HBASE-3820  Splitlog() executed while the namenode was in safemode may
+               cause data-loss (Jieshan Bean)
 
   IMPROVEMENT
    HBASE-3882  hbase-config.sh needs to be updated so it can auto-detects the

src/main/java/org/apache/hadoop/hbase/master/MasterFileSystem.java

@@ -147,6 +147,7 @@ public class MasterFileSystem {
     if (this.fsOk) {
       try {
         FSUtils.checkFileSystemAvailable(this.fs);
+        FSUtils.checkDfsSafeMode(this.conf);
       } catch (IOException e) {
         master.abort("Shutting down HBase cluster: file system not available", e);
         this.fsOk = false;
@@ -235,6 +236,9 @@ public class MasterFileSystem {
         HLogSplitter splitter = HLogSplitter.createLogSplitter(
             conf, rootdir, logDir, oldLogDir, this.fs);
         try {
+          // If FS is in safe mode, just wait till out of it.
+          FSUtils.waitOnSafeMode(conf,
+            conf.getInt(HConstants.THREAD_WAKE_FREQUENCY, 1000));  
           splitter.splitLog();
         } catch (OrphanHLogAfterSplitException e) {
           LOG.warn("Retrying splitting because of:", e);

src/main/java/org/apache/hadoop/hbase/util/Bytes.java

@@ -35,6 +35,7 @@ import java.math.BigInteger;
 import java.nio.ByteBuffer;
 import java.util.Comparator;
 import java.util.Iterator;
+import java.math.BigDecimal;
 
 /**
  * Utility class that handles byte arrays, conversions to/from other types,

src/main/java/org/apache/hadoop/hbase/util/FSUtils.java

@@ -138,6 +138,26 @@ public class FSUtils {
     throw io;
   }
 
+  /**
+   * Check whether dfs is in safemode. 
+   * @param conf 
+   * @return true if dfs is in safemode.
+   * @throws IOException
+   */
+  public static void checkDfsSafeMode(final Configuration conf) 
+  throws IOException {
+    boolean isInSafeMode = false;
+    FileSystem fs = FileSystem.get(conf);
+    if (fs instanceof DistributedFileSystem) {
+      DistributedFileSystem dfs = (DistributedFileSystem)fs;
+      // Check whether dfs is on safemode.
+      isInSafeMode = dfs.setSafeMode(FSConstants.SafeModeAction.SAFEMODE_GET);
+    }
+    if (isInSafeMode) {
+      throw new IOException("File system is in safemode, it can't be written now");
+    }
+  }
+  
   /**
    * Verifies current version of file system
    *