HBASE-433 HBASE-251 Region server should delete restore log after successful restore, Stuck replaying the edits of crashed machine.

HLog

- don't overwrite oldlogfile in splitLog if it already exists. Rename it and copy it into the new oldlogfile. Then delete it once it has been copied.
- use FileUtil.fullyDelete to delete region server log directory.

HRegion

- delete oldlogfile once it has been successfully processed


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

CHANGES.txt

@@ -33,6 +33,8 @@ Hbase Change Log
    HBASE-496   impossible state for createLease writes 400k lines in about 15mins
    HBASE-472   Passing on edits, we dump all to log
    HBASE-495   No server address listed in .META.
+   HBASE-433 HBASE-251 Region server should delete restore log after successful
+               restore, Stuck replaying the edits of crashed machine.
    
   IMPROVEMENTS
    HBASE-415   Rewrite leases to use DelayedBlockingQueue instead of polling

src/java/org/apache/hadoop/hbase/regionserver/HLog.java

@@ -19,7 +19,6 @@
  */
 package org.apache.hadoop.hbase.regionserver;
 
-import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.Collections;
@@ -49,6 +48,7 @@ import org.apache.hadoop.hbase.HStoreKey;
 import org.apache.hadoop.hbase.HTableDescriptor;
 import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.RemoteExceptionHandler;
 
 /**
  * HLog stores all the edits to the HStore.
@@ -304,7 +304,7 @@ public class HLog implements HConstants {
    * file-number.
    */
   Path computeFilename(final long fn) {
-    return new Path(dir, HLOG_DATFILE + new Long(fn).toString());
+    return new Path(dir, HLOG_DATFILE + fn);
   }
 
   /**
@@ -503,7 +503,15 @@ public class HLog implements HConstants {
    */
   public static void splitLog(Path rootDir, Path srcDir, FileSystem fs,
     Configuration conf) throws IOException {
+    if (!fs.exists(srcDir)) {
+      // Nothing to do
+      return;
+    }
     FileStatus logfiles[] = fs.listStatus(srcDir);
+    if (logfiles == null || logfiles.length == 0) {
+      // Nothing to do
+      return;
+    }
     LOG.info("splitting " + logfiles.length + " log(s) in " +
       srcDir.toString());
     Map<Text, SequenceFile.Writer> logWriters =
@@ -512,12 +520,12 @@ public class HLog implements HConstants {
       for (int i = 0; i < logfiles.length; i++) {
         if (LOG.isDebugEnabled()) {
           LOG.debug("Splitting " + i + " of " + logfiles.length + ": " +
-            logfiles[i]);
+            logfiles[i].getPath());
         }
         // Check for empty file.
         if (logfiles[i].getLen() <= 0) {
           LOG.info("Skipping " + logfiles[i].toString() +
-            " because zero length");
+              " because zero length");
           continue;
         }
         HLogKey key = new HLogKey();
@@ -538,6 +546,17 @@ public class HLog implements HConstants {
                   ),
                   HREGION_OLDLOGFILE_NAME
               );
+              
+              Path oldlogfile = null;
+              SequenceFile.Reader old = null;
+              if (fs.exists(logfile)) {
+                LOG.warn("Old log file " + logfile +
+                    " already exists. Copying existing file to new file");
+                oldlogfile = new Path(logfile.toString() + ".old");
+                fs.rename(logfile, oldlogfile);
+                old = new SequenceFile.Reader(fs, oldlogfile, conf);
+              }
+
               if (LOG.isDebugEnabled()) {
                 LOG.debug("Creating new log file writer for path " + logfile +
                   "; map content " + logWriters.toString());
@@ -547,8 +566,22 @@ public class HLog implements HConstants {
               // Use copy of regionName; regionName object is reused inside in
               // HStoreKey.getRegionName so its content changes as we iterate.
               logWriters.put(new Text(regionName), w);
+              
+              if (old != null) {
+                // Copy from existing log file
+                HLogKey oldkey = new HLogKey();
+                HLogEdit oldval = new HLogEdit();
+                for (; old.next(oldkey, oldval); count++) {
+                  if (LOG.isDebugEnabled() && count > 0 && count % 10000 == 0) {
+                    LOG.debug("Copied " + count + " edits");
+                  }
+                  w.append(oldkey, oldval);
+                }
+                old.close();
+                fs.delete(oldlogfile);
+              }
             }
-            if (count % 10000 == 0 && count > 0 && LOG.isDebugEnabled()) {
+            if (LOG.isDebugEnabled() && count > 0 && count % 10000 == 0) {
               LOG.debug("Applied " + count + " edits");
             }
             w.append(key, val);
@@ -566,13 +599,13 @@ public class HLog implements HConstants {
       }
     }
 
-    if (fs.exists(srcDir)) {
-      if (!fs.delete(srcDir)) {
-        LOG.error("Cannot delete: " + srcDir);
-        if (!FileUtil.fullyDelete(new File(srcDir.toString()))) {
-          throw new IOException("Cannot delete: " + srcDir);
-        }
-      }
+    try {
+      FileUtil.fullyDelete(fs, srcDir);
+    } catch (IOException e) {
+      e = RemoteExceptionHandler.checkIOException(e);
+      IOException io = new IOException("Cannot delete: " + srcDir);
+      io.initCause(e);
+      throw io;
     }
     LOG.info("log file splitting completed for " + srcDir.toString());
   }

src/java/org/apache/hadoop/hbase/regionserver/HRegion.java

@@ -314,6 +314,13 @@ public class HRegion implements HConstants {
         maxSeqId = storeSeqId;
       }
     }
+    if (fs.exists(oldLogFile)) {
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Deleting old log file: " + oldLogFile);
+      }
+      fs.delete(oldLogFile);
+    }
+    
     this.minSequenceId = maxSeqId;
     if (LOG.isDebugEnabled()) {
       LOG.debug("Next sequence id for region " + regionInfo.getRegionName() +