HADOOP-1076. Periodic checkpoint can continue even if an earlier

checkpoint encountered an error.  (Dhruba Borthakur via dhruba)



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

CHANGES.txt

@@ -93,6 +93,9 @@ Trunk (unreleased changes)
     HADOOP-1500. Fix typographical errors in the DFS WebUI.
     (Nigel Daley via dhruba)
 
+    HADOOP-1076. Periodic checkpoint can continue even if an earlier
+    checkpoint encountered an error.  (Dhruba Borthakur via dhruba)
+
     HADOOP-1887. The Namenode encounters an ArrayIndexOutOfBoundsException
     while listing a directory that had a file that was
     being actively written to.  (Dhruba Borthakur via dhruba)

src/java/org/apache/hadoop/dfs/ClientProtocol.java

@@ -34,8 +34,9 @@ interface ClientProtocol extends VersionedProtocol {
    * 16 : removed deprecated obtainLock() and releaseLock(). 
    * 17 : getBlockSize replaced by getPreferredBlockSize
    * 18 : datanodereport returns dead, live or all nodes.
+   * 19 : rollEditLog() returns a token to uniquely identify the editfile.
    */
-  public static final long versionID = 18L;
+  public static final long versionID = 19L;
   
   ///////////////////////////////////////
   // File contents
@@ -326,11 +327,11 @@ interface ClientProtocol extends VersionedProtocol {
 
   /**
    * Closes the current edit log and opens a new one. The 
-   * call fails if there are already two or more edits log files or
-   * if the file system is in SafeMode.
+   * call fails if the file system is in SafeMode.
+   * Returns a unique token to identify this transaction.
    * @throws IOException
    */
-  public void rollEditLog() throws IOException;
+  public long rollEditLog() throws IOException;
 
   /**
    * Rolls the fsImage log. It removes the old fsImage, copies the

src/java/org/apache/hadoop/dfs/FSConstants.java

@@ -133,7 +133,10 @@ public interface FSConstants {
 
   // type of the datanode report
   public static enum DatanodeReportType {ALL, LIVE, DEAD }
-  
+
+  // checkpoint states
+  public enum CheckpointStates{ START, ROLLED_EDITS, UPLOAD_START, UPLOAD_DONE; }
+
   /**
    * Type of the node
    */

src/java/org/apache/hadoop/dfs/FSEditLog.java

@@ -393,7 +393,6 @@ class FSEditLog {
                                     + " for version " + logVersion);
             DatanodeID nodeID = new DatanodeID();
             nodeID.readFields(in);
-            DatanodeDescriptor node = fsNamesys.getDatanode(nodeID);
             //Datanodes are not persistent any more.
             break;
           }
@@ -577,13 +576,21 @@ class FSEditLog {
  
   /**
    * Closes the current edit log and opens edits.new. 
+   * Returns the lastModified time of the edits log.
    */
   synchronized void rollEditLog() throws IOException {
     //
-    // If edits.new already exists, then return error.
+    // If edits.new already exists in some directory, verify it
+    // exists in all directories.
     //
     if (existsNew()) {
-      throw new IOException("Attempt to roll edit log but edits.new exists");
+      for (int idx = 0; idx < getNumStorageDirs(); idx++) {
+        if (!getEditNewFile(idx).exists()) { 
+          throw new IOException("Inconsistent existance of edits.new " +
+                                getEditNewFile(idx));
+        }
+      }
+      return; // nothing to do, edits.new exists!
     }
 
     close();                     // close existing edit log
@@ -645,4 +652,11 @@ class FSEditLog {
   synchronized File getFsEditName() throws IOException {
     return getEditFile(0);
   }
+
+  /**
+   * Returns the timestamp of the edit log
+   */
+  synchronized long getFsEditTime() throws IOException {
+    return getEditFile(0).lastModified();
+  }
 }

src/java/org/apache/hadoop/dfs/FSNamesystem.java

@@ -33,6 +33,7 @@ import java.io.IOException;
 import java.io.PrintWriter;
 import java.util.*;
 import java.util.Map.Entry;
+import java.text.SimpleDateFormat;
 
 /***************************************************
  * FSNamesystem does the actual bookkeeping work for the
@@ -195,6 +196,13 @@ class FSNamesystem implements FSConstants {
   private HostsFileReader hostsReader; 
   private Daemon dnthread = null;
 
+  // can fs-image be rolled?
+  volatile private CheckpointStates ckptState = CheckpointStates.START; 
+
+  private static final SimpleDateFormat DATE_FORM =
+    new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
+
   /**
    * dirs is a list of directories where the filesystem directory state 
    * is stored
@@ -3447,13 +3455,15 @@ class FSNamesystem implements FSConstants {
     return getEditLog().getEditLogSize();
   }
 
-  synchronized void rollEditLog() throws IOException {
+  synchronized long rollEditLog() throws IOException {
     if (isInSafeMode()) {
       throw new SafeModeException("Checkpoint not created",
                                   safeMode);
     }
     LOG.info("Roll Edit Log from " + Server.getRemoteAddress());
     getEditLog().rollEditLog();
+    ckptState = CheckpointStates.ROLLED_EDITS;
+    return getEditLog().getFsEditTime();
   }
 
   synchronized void rollFSImage() throws IOException {
@@ -3462,13 +3472,46 @@ class FSNamesystem implements FSConstants {
       throw new SafeModeException("Checkpoint not created",
                                   safeMode);
     }
+    if (ckptState != CheckpointStates.UPLOAD_DONE) {
+      throw new IOException("Cannot roll fsImage before rolling edits log.");
+    }
     dir.fsImage.rollFSImage();
+    ckptState = CheckpointStates.START;
   }
 
   File getFsEditName() throws IOException {
     return getEditLog().getFsEditName();
   }
-    
+
+  /*
+   * This is called just before a new checkpoint is uploaded to the
+   * namenode.
+   */
+  synchronized void validateCheckpointUpload(long token) throws IOException {
+    if (ckptState != CheckpointStates.ROLLED_EDITS) {
+      throw new IOException("Namenode is not expecting an new image " +
+                             ckptState);
+    } 
+    // verify token
+    long modtime = getEditLog().getFsEditTime();
+    if (token != modtime) {
+      throw new IOException("Namenode has an edit log with timestamp of " +
+                            DATE_FORM.format(new Date(modtime)) +
+                            " but new checkpoint was created using editlog " +
+                            " with timestamp " + 
+                            DATE_FORM.format(new Date(token)) + 
+                            ". Checkpoint Aborted.");
+    }
+    ckptState = CheckpointStates.UPLOAD_START;
+  }
+
+  /*
+   * This is called when a checkpoint upload finishes successfully.
+   */
+  synchronized void checkpointUploadDone() {
+    ckptState = CheckpointStates.UPLOAD_DONE;
+  }
+
   /**
    * Returns whether the given block is one pointed-to by a file.
    */

src/java/org/apache/hadoop/dfs/GetImageServlet.java

@@ -20,8 +20,6 @@ package org.apache.hadoop.dfs;
 import java.util.*;
 import java.io.*;
 import org.apache.hadoop.util.StringUtils;
-import org.apache.hadoop.conf.*;
-import org.apache.hadoop.mapred.StatusHttpServer;
 import org.apache.commons.logging.*;
 import javax.servlet.ServletContext;
 import javax.servlet.ServletException;
@@ -58,8 +56,10 @@ public class GetImageServlet extends HttpServlet {
                                       nn.getFsEditName());
       } else if (ff.putImage()) {
         // issue a HTTP get request to download the new fsimage 
+        nn.validateCheckpointUpload(ff.getToken());
         TransferFsImage.getFileClient(ff.getInfoServer(), "getimage=1", 
                                       nn.getFsImageNameCheckpoint());
+        nn.checkpointUploadDone();
       }
     } catch (IOException ie) {
       StringUtils.stringifyException(ie);

src/java/org/apache/hadoop/dfs/NameNode.java

@@ -521,8 +521,8 @@ public class NameNode implements ClientProtocol, DatanodeProtocol, FSConstants {
   /**
    * Roll the edit log.
    */
-  public void rollEditLog() throws IOException {
-    namesystem.rollEditLog();
+  public long rollEditLog() throws IOException {
+    return namesystem.rollEditLog();
   }
 
   /**
@@ -711,6 +711,20 @@ public class NameNode implements ClientProtocol, DatanodeProtocol, FSConstants {
     return getFSImage().getFsImageNameCheckpoint();
   }
 
+  /**
+   * Validates that this is a valid checkpoint upload request
+   */
+  public void validateCheckpointUpload(long token) throws IOException {
+    namesystem.validateCheckpointUpload(token);
+  }
+
+  /**
+   * Indicates that a new checkpoint has been successfully uploaded.
+   */
+  public void checkpointUploadDone() {
+    namesystem.checkpointUploadDone();
+  }
+
   /**
    * Returns the name of the edits file
    */

src/java/org/apache/hadoop/dfs/SecondaryNameNode.java

@@ -223,11 +223,12 @@ public class SecondaryNameNode implements FSConstants, Runnable {
   /**
    * Copy the new fsimage into the NameNode
    */
-  private void putFSImage() throws IOException {
+  private void putFSImage(long token) throws IOException {
     String fsName = getInfoServer();
     String fileid = "putimage=1&port=" + infoPort +
       "&machine=" +
-      InetAddress.getLocalHost().getHostAddress();
+      InetAddress.getLocalHost().getHostAddress() +
+      "&token=" + token;
     LOG.info("Posted URL " + fsName + fileid);
     TransferFsImage.getFileClient(fsName, fileid, (File[])null);
   }
@@ -257,8 +258,9 @@ public class SecondaryNameNode implements FSConstants, Runnable {
 
     //
     // Tell the namenode to start logging transactions in a new edit file
+    // Retuns a token that would be used to upload the merged image.
     //
-    namenode.rollEditLog();
+    long token = namenode.rollEditLog();
 
     //
     // error simulation code for junit test
@@ -276,7 +278,7 @@ public class SecondaryNameNode implements FSConstants, Runnable {
     // Upload the new image into the NameNode. Then tell the Namenode
     // to make this new uploaded image as the most current image.
     //
-    putFSImage();
+    putFSImage(token);
 
     //
     // error simulation code for junit test

src/java/org/apache/hadoop/dfs/TransferFsImage.java

@@ -34,6 +34,7 @@ class TransferFsImage implements FSConstants {
   private boolean isPutImage;
   private int remoteport;
   private String machineName;
+  private long token;
   
   /**
    * File downloader.
@@ -50,6 +51,7 @@ class TransferFsImage implements FSConstants {
     isGetImage = isGetEdit = isPutImage = false;
     remoteport = 0;
     machineName = null;
+    token = 0;
 
     for (Iterator<String> it = pmap.keySet().iterator(); it.hasNext();) {
       String key = it.next();
@@ -63,6 +65,8 @@ class TransferFsImage implements FSConstants {
         remoteport = new Integer(pmap.get("port")[0]).intValue();
       } else if (key.equals("machine")) { 
         machineName = pmap.get("machine")[0];
+      } else if (key.equals("token")) { 
+        token = new Long(pmap.get("token")[0]).longValue();
       }
     }
     if ((isGetImage && isGetEdit) ||
@@ -83,6 +87,10 @@ class TransferFsImage implements FSConstants {
     return isPutImage;
   }
 
+  long getToken() {
+    return token;
+  }
+
   String getInfoServer() throws IOException{
     if (machineName == null || remoteport == 0) {
       throw new IOException ("MachineName and port undefined");

src/test/org/apache/hadoop/dfs/TestCheckpoint.java

@@ -190,8 +190,7 @@ public class TestCheckpoint extends TestCase {
     try {
       assertTrue(!fileSys.exists(file1));
       //
-      // Make the checkpoint fail after rolling the
-      // edit log.
+      // Make the checkpoint fail after uploading the new fsimage.
       //
       SecondaryNameNode secondary = new SecondaryNameNode(conf);
       secondary.initializeErrorSimulationEvent(2);
@@ -235,6 +234,72 @@ public class TestCheckpoint extends TestCase {
     }
   }
 
+  /*
+   * Simulate a secondary namenode crash after rolling the edit log.
+   */
+  private void testSecondaryNamenodeError3(Configuration conf)
+    throws IOException {
+    System.out.println("Starting testSecondaryNamenodeError 31");
+    Path file1 = new Path("checkpointzz.dat");
+    MiniDFSCluster cluster = new MiniDFSCluster(conf, numDatanodes, 
+                                                false, null);
+    cluster.waitActive();
+    FileSystem fileSys = cluster.getFileSystem();
+    try {
+      assertTrue(!fileSys.exists(file1));
+      //
+      // Make the checkpoint fail after rolling the edit log.
+      //
+      SecondaryNameNode secondary = new SecondaryNameNode(conf);
+      secondary.initializeErrorSimulationEvent(2);
+      secondary.setErrorSimulation(0);
+
+      try {
+        secondary.doCheckpoint();  // this should fail
+        assertTrue(false);      
+      } catch (IOException e) {
+      }
+      secondary.shutdown(); // secondary namenode crash!
+
+      // start new instance of secondary and verify that 
+      // a new rollEditLog suceedes inspite of the fact that 
+      // edits.new already exists.
+      //
+      secondary = new SecondaryNameNode(conf);
+      secondary.doCheckpoint();  // this should work correctly
+      secondary.shutdown();
+
+      //
+      // Create a new file
+      //
+      writeFile(fileSys, file1, replication);
+      checkFile(fileSys, file1, replication);
+    } finally {
+      fileSys.close();
+      cluster.shutdown();
+    }
+
+    //
+    // Restart cluster and verify that file exists.
+    // Then take another checkpoint to verify that the 
+    // namenode restart accounted for the twice-rolled edit logs.
+    //
+    System.out.println("Starting testSecondaryNamenodeError 32");
+    cluster = new MiniDFSCluster(conf, numDatanodes, false, null);
+    cluster.waitActive();
+    fileSys = cluster.getFileSystem();
+    try {
+      checkFile(fileSys, file1, replication);
+      cleanupFile(fileSys, file1);
+      SecondaryNameNode secondary = new SecondaryNameNode(conf);
+      secondary.doCheckpoint();
+      secondary.shutdown();
+    } finally {
+      fileSys.close();
+      cluster.shutdown();
+    }
+  }
+
   /**
    * Tests checkpoint in DFS.
    */
@@ -322,6 +387,7 @@ public class TestCheckpoint extends TestCase {
 
     testSecondaryNamenodeError1(conf);
     testSecondaryNamenodeError2(conf);
+    testSecondaryNamenodeError3(conf);
     testNamedirError(conf, namedirs);
   }
 }