Merging changes r1083084:r1083902 from trunk to federation

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hdfs/branches/HDFS-1052@1089256 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -307,6 +307,9 @@ Trunk (unreleased changes)
     HDFS-1763. Replace hard-coded option strings with variables from
     DFSConfigKeys. (eli)
 
+    HDFS-1521. Persist transaction ID on disk between NN restarts. (Ivan Kelly
+    and Todd Lipcon via todd)
+
   OPTIMIZATIONS
 
     HDFS-1458. Improve checkpoint performance by avoiding unnecessary image
@@ -367,6 +370,8 @@ Trunk (unreleased changes)
 
     HDFS-1757. Don't compile fuse-dfs by default. (eli)
 
+    HDFS-1770. TestFiRename fails due to invalid block size. (eli)
+
 Release 0.22.0 - Unreleased
 
   NEW FEATURES

src/java/org/apache/hadoop/hdfs/protocol/FSConstants.java

@@ -88,7 +88,7 @@ public interface FSConstants {
   // Version is reflected in the data storage file.
   // Versions are negative.
   // Decrement LAYOUT_VERSION to define a new version.
-  public static final int LAYOUT_VERSION = -28;
+  public static final int LAYOUT_VERSION = -29;
   // Current version: 
-  // -28: Adding support for block pools and multiple namenodes
+  // -29: Adding support for block pools and multiple namenodes
 }

src/java/org/apache/hadoop/hdfs/server/common/Storage.java

@@ -79,7 +79,7 @@ public abstract class Storage extends StorageInfo {
   public static final int PRE_RBW_LAYOUT_VERSION = -19;
   
   // last layout version that is before federation
-  public static final int LAST_PRE_FEDERATION_LAYOUT_VERSION = -27;
+  public static final int LAST_PRE_FEDERATION_LAYOUT_VERSION = -28;
   
   private   static final String STORAGE_FILE_LOCK     = "in_use.lock";
   protected static final String STORAGE_FILE_VERSION  = "VERSION";

src/java/org/apache/hadoop/hdfs/server/namenode/BackupImage.java

@@ -55,6 +55,7 @@ public class BackupImage extends FSImage {
 
   /** Is journal spooling in progress */
   volatile JSpoolState jsState;
+  private long lastAppliedTxId = 0;
 
   static enum JSpoolState {
     OFF,
@@ -153,6 +154,11 @@ public class BackupImage extends FSImage {
     if(!editLog.isOpen())
       editLog.open();
 
+    // set storage fields
+    storage.setStorageInfo(sig);
+    storage.setImageDigest(sig.getImageDigest());
+    storage.setCheckpointTime(sig.checkpointTime);
+
     FSDirectory fsDir = getFSNamesystem().dir;
     if(fsDir.isEmpty()) {
       Iterator<StorageDirectory> itImage
@@ -163,6 +169,7 @@ public class BackupImage extends FSImage {
         throw new IOException("Could not locate checkpoint directories");
       StorageDirectory sdName = itImage.next();
       StorageDirectory sdEdits = itEdits.next();
+
       getFSDirectoryRootLock().writeLock();
       try { // load image under rootDir lock
         loadFSImage(NNStorage.getStorageFile(sdName, NameNodeFile.IMAGE));
@@ -170,12 +177,8 @@ public class BackupImage extends FSImage {
         getFSDirectoryRootLock().writeUnlock();
       }
       loadFSEdits(sdEdits);
+      lastAppliedTxId = getEditLog().getLastWrittenTxId();
     }
-
-    // set storage fields
-    storage.setStorageInfo(sig);
-    storage.setImageDigest(sig.imageDigest);
-    storage.setCheckpointTime(sig.checkpointTime);
   }
 
   /**
@@ -227,7 +230,8 @@ public class BackupImage extends FSImage {
           backupInputStream.setBytes(data);
           FSEditLogLoader logLoader = new FSEditLogLoader(namesystem);
           logLoader.loadEditRecords(storage.getLayoutVersion(),
-                    backupInputStream.getDataInputStream(), true);
+              backupInputStream.getDataInputStream(), true,
+              lastAppliedTxId + 1);
           getFSNamesystem().dir.updateCountForINodeWithQuota(); // inefficient!
           break;
         case INPROGRESS:
@@ -348,12 +352,18 @@ public class BackupImage extends FSImage {
       EditLogFileInputStream edits = new EditLogFileInputStream(jSpoolFile);
       DataInputStream in = edits.getDataInputStream();
       FSEditLogLoader logLoader = new FSEditLogLoader(namesystem);
-      numEdits += logLoader.loadFSEdits(in, false);
+      int loaded = logLoader.loadFSEdits(in, false, lastAppliedTxId + 1);
+
+      lastAppliedTxId += loaded;
+      numEdits += loaded;
 
       // first time reached the end of spool
       jsState = JSpoolState.WAIT;
-      numEdits += logLoader.loadEditRecords(storage.getLayoutVersion(),
-                                            in, true);
+      loaded = logLoader.loadEditRecords(storage.getLayoutVersion(),
+                                         in, true, lastAppliedTxId + 1);
+      numEdits += loaded;
+      lastAppliedTxId += loaded;
+
       getFSNamesystem().dir.updateCountForINodeWithQuota();
       edits.close();
     }

src/java/org/apache/hadoop/hdfs/server/namenode/EditLogBackupOutputStream.java

@@ -51,15 +51,18 @@ class EditLogBackupOutputStream extends EditLogOutputStream {
 
   static class JournalRecord {
     byte op;
+    long txid;
     Writable[] args;
 
-    JournalRecord(byte op, Writable ... writables) {
+    JournalRecord(byte op, long txid, Writable ... writables) {
       this.op = op;
+      this.txid = txid;
       this.args = writables;
     }
 
     void write(DataOutputStream out) throws IOException {
       out.write(op);
+      out.writeLong(txid);
       if(args == null)
         return;
       for(Writable w : args)
@@ -105,8 +108,8 @@ class EditLogBackupOutputStream extends EditLogOutputStream {
   }
 
   @Override // EditLogOutputStream
-  void write(byte op, Writable ... writables) throws IOException {
-    bufCurrent.add(new JournalRecord(op, writables));
+  void write(byte op, long txid, Writable ... writables) throws IOException {
+    bufCurrent.add(new JournalRecord(op, txid, writables));
   }
 
   /**

src/java/org/apache/hadoop/hdfs/server/namenode/EditLogFileOutputStream.java

@@ -81,10 +81,18 @@ class EditLogFileOutputStream extends EditLogOutputStream {
     bufCurrent.write(b);
   }
 
-  /** {@inheritDoc} */
+  /**
+   * Write a transaction to the stream. The serialization format is:
+   * <ul>
+   *   <li>the opcode (byte)</li>
+   *   <li>the transaction id (long)</li>
+   *   <li>the actual Writables for the transaction</li>
+   * </ul>
+   * */
   @Override
-  void write(byte op, Writable... writables) throws IOException {
+  void write(byte op, long txid, Writable... writables) throws IOException {
     write(op);
+    bufCurrent.writeLong(txid);
     for (Writable w : writables) {
       w.write(bufCurrent);
     }

src/java/org/apache/hadoop/hdfs/server/namenode/EditLogOutputStream.java

@@ -46,10 +46,12 @@ implements JournalStream {
    * an array of Writable arguments.
    * 
    * @param op operation
+   * @param txid the transaction ID of this operation
    * @param writables array of Writable arguments
    * @throws IOException
    */
-  abstract void write(byte op, Writable ... writables) throws IOException;
+  abstract void write(byte op, long txid, Writable ... writables)
+  throws IOException;
 
   /**
    * Create and initialize underlying persistent edits log storage.

src/java/org/apache/hadoop/hdfs/server/namenode/FSEditLog.java

@@ -306,12 +306,19 @@ public class FSEditLog implements NNStorageListener {
       if(getNumEditStreams() == 0)
         throw new java.lang.IllegalStateException(NO_JOURNAL_STREAMS_WARNING);
       ArrayList<EditLogOutputStream> errorStreams = null;
+
+      // Only start a new transaction for OPs which will be persisted to disk.
+      // Obviously this excludes control op codes.
       long start = now();
+      if (opCode.getOpCode() < FSEditLogOpCodes.OP_JSPOOL_START.getOpCode()) {
+        start = beginTransaction();
+      }
+
       for(EditLogOutputStream eStream : editStreams) {
         if(!eStream.isOperationSupported(opCode.getOpCode()))
           continue;
         try {
-          eStream.write(opCode.getOpCode(), writables);
+          eStream.write(opCode.getOpCode(), txid, writables);
         } catch (IOException ie) {
           LOG.error("logEdit: removing "+ eStream.getName(), ie);
           if(errorStreams == null)
@@ -320,7 +327,7 @@ public class FSEditLog implements NNStorageListener {
         }
       }
       disableAndReportErrorOnStreams(errorStreams);
-      recordTransaction(start);
+      endTransaction(start);
       
       // check if it is time to schedule an automatic sync
       if (!shouldForceSync()) {
@@ -371,7 +378,8 @@ public class FSEditLog implements NNStorageListener {
     return false;
   }
   
-  private void recordTransaction(long start) {
+  private long beginTransaction() {
+    assert Thread.holdsLock(this);
     // get a new transactionId
     txid++;
 
@@ -380,7 +388,12 @@ public class FSEditLog implements NNStorageListener {
     //
     TransactionId id = myTransactionId.get();
     id.txid = txid;
-
+    return now();
+  }
+  
+  private void endTransaction(long start) {
+    assert Thread.holdsLock(this);
+    
     // update statistics
     long end = now();
     numTransactions++;
@@ -389,6 +402,21 @@ public class FSEditLog implements NNStorageListener {
       metrics.transactions.inc((end-start));
   }
 
+  /**
+   * Return the transaction ID of the last transaction written to the log.
+   */
+  synchronized long getLastWrittenTxId() {
+    return txid;
+  }
+  
+  /**
+   * Set the transaction ID to use for the next transaction written.
+   */
+  synchronized void setNextTxId(long nextTxid) {
+    assert synctxid <= txid;
+    txid = nextTxid - 1;
+  }
+  
   /**
    * Blocks until all ongoing edits have been synced to disk.
    * This differs from logSync in that it waits for edits that have been
@@ -784,6 +812,8 @@ public class FSEditLog implements NNStorageListener {
   
   /**
    * Closes the current edit log and opens edits.new. 
+   * @return the transaction id that will be used as the first transaction
+   *         in the new log
    */
   synchronized void rollEditLog() throws IOException {
     waitForSyncToFinish();
@@ -1014,7 +1044,7 @@ public class FSEditLog implements NNStorageListener {
     if(getNumEditStreams() == 0)
       throw new java.lang.IllegalStateException(NO_JOURNAL_STREAMS_WARNING);
     ArrayList<EditLogOutputStream> errorStreams = null;
-    long start = now();
+    long start = beginTransaction();
     for(EditLogOutputStream eStream : editStreams) {
       try {
         eStream.write(data, 0, length);
@@ -1026,7 +1056,7 @@ public class FSEditLog implements NNStorageListener {
       }
     }
     disableAndReportErrorOnStreams(errorStreams);
-    recordTransaction(start);
+    endTransaction(start);
   }
 
   /**

src/java/org/apache/hadoop/hdfs/server/namenode/FSEditLogLoader.java

@@ -53,17 +53,20 @@ public class FSEditLogLoader {
    * This is where we apply edits that we've been writing to disk all
    * along.
    */
-  int loadFSEdits(EditLogInputStream edits) throws IOException {
+  int loadFSEdits(EditLogInputStream edits, long expectedStartingTxId)
+  throws IOException {
     DataInputStream in = edits.getDataInputStream();
     long startTime = now();
-    int numEdits = loadFSEdits(in, true);
+    int numEdits = loadFSEdits(in, true, expectedStartingTxId);
     FSImage.LOG.info("Edits file " + edits.getName() 
         + " of size " + edits.length() + " edits # " + numEdits 
         + " loaded in " + (now()-startTime)/1000 + " seconds.");
     return numEdits;
   }
 
-  int loadFSEdits(DataInputStream in, boolean closeOnExit) throws IOException {
+  int loadFSEdits(DataInputStream in, boolean closeOnExit,
+      long expectedStartingTxId)
+  throws IOException {
     int numEdits = 0;
     int logVersion = 0;
 
@@ -89,19 +92,19 @@ public class FSEditLogLoader {
       }
       assert logVersion <= Storage.LAST_UPGRADABLE_LAYOUT_VERSION :
                             "Unsupported version " + logVersion;
-      numEdits = loadEditRecords(logVersion, in, false);
+      
+      numEdits = loadEditRecords(logVersion, in, false, expectedStartingTxId);
     } finally {
       if(closeOnExit)
         in.close();
     }
-    if (logVersion != FSConstants.LAYOUT_VERSION) // other version
-      numEdits++; // save this image asap
+    
     return numEdits;
   }
 
   @SuppressWarnings("deprecation")
   int loadEditRecords(int logVersion, DataInputStream in,
-      boolean closeOnExit) throws IOException {
+      boolean closeOnExit, long expectedStartingTxId) throws IOException {
     FSDirectory fsDir = fsNamesys.dir;
     int numEdits = 0;
     String clientName = null;
@@ -116,6 +119,8 @@ public class FSEditLogLoader {
         numOpUpdateMasterKey = 0, numOpOther = 0;
 
     try {
+      long txId = expectedStartingTxId - 1;
+
       while (true) {
         long timestamp = 0;
         long mtime = 0;
@@ -133,6 +138,17 @@ public class FSEditLogLoader {
         } catch (EOFException e) {
           break; // no more transactions
         }
+
+        if (logVersion <= -28) {
+          // Read the txid
+          long thisTxId = in.readLong();
+          if (thisTxId != txId + 1) {
+            throw new IOException("Expected transaction ID " +
+                (txId + 1) + " but got " + thisTxId);
+          }
+          txId = thisTxId;
+        }
+        
         numEdits++;
         switch (opCode) {
         case OP_ADD:

src/java/org/apache/hadoop/hdfs/server/namenode/FSImage.java

@@ -167,7 +167,7 @@ public class FSImage implements NNStorageListener, Closeable {
       storage.setUpgradeManager(ns.upgradeManager);
     }
   }
-
+ 
   void setCheckpointDirectories(Collection<URI> dirs,
                                 Collection<URI> editsDirs) {
     checkpointDirs = dirs;
@@ -661,16 +661,22 @@ public class FSImage implements NNStorageListener, Closeable {
     //
     // Load in bits
     //
+    latestEditsSD.read();
+    long editsVersion = storage.getLayoutVersion();
     latestNameSD.read();
-    needToSave |= loadFSImage(NNStorage.getStorageFile(latestNameSD,
-                                                       NameNodeFile.IMAGE));
+    long imageVersion = storage.getLayoutVersion();
+
+    loadFSImage(NNStorage.getStorageFile(latestNameSD, NameNodeFile.IMAGE));
     
     // Load latest edits
-    if (latestNameCheckpointTime > latestEditsCheckpointTime)
+    if (latestNameCheckpointTime > latestEditsCheckpointTime) {
       // the image is already current, discard edits
       needToSave |= true;
-    else // latestNameCheckpointTime == latestEditsCheckpointTime
-      needToSave |= (loadFSEdits(latestEditsSD) > 0);
+    } else { // latestNameCheckpointTime == latestEditsCheckpointTime
+      needToSave |= loadFSEdits(latestEditsSD);
+    }
+    needToSave |= (editsVersion != FSConstants.LAYOUT_VERSION 
+                    || imageVersion != FSConstants.LAYOUT_VERSION);
     
     return needToSave;
   }
@@ -680,7 +686,7 @@ public class FSImage implements NNStorageListener, Closeable {
    * filenames and blocks.  Return whether we should
    * "re-save" and consolidate the edit-logs
    */
-  boolean loadFSImage(File curFile) throws IOException {
+  void loadFSImage(File curFile) throws IOException {
     FSImageFormat.Loader loader = new FSImageFormat.Loader(
         conf, getFSNamesystem());
     loader.load(curFile);
@@ -697,44 +703,42 @@ public class FSImage implements NNStorageListener, Closeable {
           " is corrupt with MD5 checksum of " + readImageMd5 +
           " but expecting " + storage.getImageDigest());
     }
-
-    storage.namespaceID = loader.getLoadedNamespaceID();
-    storage.layoutVersion = loader.getLoadedImageVersion();
-
-    boolean needToSave =
-      loader.getLoadedImageVersion() != FSConstants.LAYOUT_VERSION;
-    return needToSave;
+    storage.setCheckpointTxId(loader.getLoadedImageTxId());
   }
 
   /**
    * Load and merge edits from two edits files
    * 
    * @param sd storage directory
-   * @return number of edits loaded
+   * @return true if the image should be re-saved
    * @throws IOException
    */
-  int loadFSEdits(StorageDirectory sd) throws IOException {
+  boolean loadFSEdits(StorageDirectory sd) throws IOException {
     FSEditLogLoader loader = new FSEditLogLoader(namesystem);
     
-    int numEdits = 0;
     EditLogFileInputStream edits =
       new EditLogFileInputStream(NNStorage.getStorageFile(sd,
                                                           NameNodeFile.EDITS));
-    
-    numEdits = loader.loadFSEdits(edits);
+    long startingTxId = storage.getCheckpointTxId() + 1;
+    long numLoaded = loader.loadFSEdits(edits, startingTxId);
+    startingTxId += numLoaded;
+
     edits.close();
     File editsNew = NNStorage.getStorageFile(sd, NameNodeFile.EDITS_NEW);
     
     if (editsNew.exists() && editsNew.length() > 0) {
       edits = new EditLogFileInputStream(editsNew);
-      numEdits += loader.loadFSEdits(edits);
+      numLoaded += loader.loadFSEdits(edits, startingTxId);
       edits.close();
     }
     
     // update the counts.
     getFSNamesystem().dir.updateCountForINodeWithQuota();    
     
-    return numEdits;
+    // update the txid for the edit log
+    editLog.setNextTxId(storage.getCheckpointTxId() + numLoaded + 1);
+    
+    return numLoaded > 0;
   }
 
   /**
@@ -745,6 +749,7 @@ public class FSImage implements NNStorageListener, Closeable {
     FSImageCompression compression = FSImageCompression.createCompression(conf);
     saver.save(newFile, getFSNamesystem(), compression);
     storage.setImageDigest(saver.getSavedDigest());
+    storage.setCheckpointTxId(editLog.getLastWrittenTxId());
   }
 
   /**

src/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormat.java

@@ -40,6 +40,7 @@ import org.apache.hadoop.fs.permission.PermissionStatus;
 import org.apache.hadoop.hdfs.DFSUtil;
 import org.apache.hadoop.hdfs.protocol.FSConstants;
 import org.apache.hadoop.hdfs.server.common.GenerationStamp;
+import org.apache.hadoop.hdfs.server.common.InconsistentFSStateException;
 import org.apache.hadoop.io.MD5Hash;
 import org.apache.hadoop.io.Text;
 
@@ -67,10 +68,8 @@ class FSImageFormat {
     /** Set to true once a file has been loaded using this loader. */
     private boolean loaded = false;
 
-    /** The image version of the loaded file */
-    private int imgVersion;
-    /** The namespace ID of the loaded file */
-    private int imgNamespaceID;
+    /** The transaction ID of the last edit represented by the loaded file */
+    private long imgTxId;
     /** The MD5 sum of the loaded file */
     private MD5Hash imgDigest;
 
@@ -79,15 +78,6 @@ class FSImageFormat {
       this.namesystem = namesystem;
     }
 
-    /**
-     * Return the version number of the image that has been loaded.
-     * @throws IllegalStateException if load() has not yet been called.
-     */
-    int getLoadedImageVersion() {
-      checkLoaded();
-      return imgVersion;
-    }
-    
     /**
      * Return the MD5 checksum of the image that has been loaded.
      * @throws IllegalStateException if load() has not yet been called.
@@ -97,13 +87,9 @@ class FSImageFormat {
       return imgDigest;
     }
 
-    /**
-     * Return the namespace ID of the image that has been loaded.
-     * @throws IllegalStateException if load() has not yet been called.
-     */
-    int getLoadedNamespaceID() {
+    long getLoadedImageTxId() {
       checkLoaded();
-      return imgNamespaceID;
+      return imgTxId;
     }
 
     /**
@@ -152,10 +138,14 @@ class FSImageFormat {
          * it should not contain version and namespace fields
          */
         // read image version: first appeared in version -1
-        imgVersion = in.readInt();
+        long imgVersion = in.readInt();
+        if(getLayoutVersion() != imgVersion)
+          throw new InconsistentFSStateException(curFile, 
+              "imgVersion " + imgVersion +
+              " expected to be " + getLayoutVersion());
 
         // read namespaceID: first appeared in version -2
-        imgNamespaceID = in.readInt();
+        in.readInt();
 
         // read number of files
         long numFiles = readNumFiles(in);
@@ -165,6 +155,15 @@ class FSImageFormat {
           long genstamp = in.readLong();
           namesystem.setGenerationStamp(genstamp); 
         }
+        
+        // read the transaction ID of the last edit represented by
+        // this image
+        if (imgVersion <= -28) {
+          imgTxId = in.readLong();
+        } else {
+          imgTxId = 0;
+        }
+        
 
         // read compression related info
         FSImageCompression compression;
@@ -256,11 +255,12 @@ class FSImageFormat {
    * @return an inode
    */
   private INode loadINode(DataInputStream in)
-  throws IOException {
+      throws IOException {
     long modificationTime = 0;
     long atime = 0;
     long blockSize = 0;
     
+    long imgVersion = getLayoutVersion();
     short replication = in.readShort();
     replication = namesystem.adjustReplication(replication);
     modificationTime = in.readLong();
@@ -326,7 +326,10 @@ class FSImageFormat {
           modificationTime, atime, nsQuota, dsQuota, blockSize);
     }
 
-    private void loadDatanodes(DataInputStream in) throws IOException {
+    private void loadDatanodes(DataInputStream in)
+        throws IOException {
+      long imgVersion = getLayoutVersion();
+
       if (imgVersion > -3) // pre datanode image version
         return;
       if (imgVersion <= -12) {
@@ -342,6 +345,7 @@ class FSImageFormat {
     private void loadFilesUnderConstruction(DataInputStream in)
     throws IOException {
       FSDirectory fsDir = namesystem.dir;
+      long imgVersion = getLayoutVersion();
       if (imgVersion > -13) // pre lease image version
         return;
       int size = in.readInt();
@@ -367,7 +371,10 @@ class FSImageFormat {
       }
     }
 
-    private void loadSecretManagerState(DataInputStream in) throws IOException {
+    private void loadSecretManagerState(DataInputStream in)
+        throws IOException {
+      long imgVersion = getLayoutVersion();
+
       if (imgVersion > -23) {
         //SecretManagerState is not available.
         //This must not happen if security is turned on.
@@ -376,8 +383,14 @@ class FSImageFormat {
       namesystem.loadSecretManagerState(in);
     }
 
+    private long getLayoutVersion() {
+      return namesystem.getFSImage().getStorage().getLayoutVersion();
+    }
+
+    private long readNumFiles(DataInputStream in)
+        throws IOException {
+      long imgVersion = getLayoutVersion();
 
-    private long readNumFiles(DataInputStream in) throws IOException {
       if (imgVersion <= -16) {
         return in.readLong();
       } else {
@@ -472,9 +485,12 @@ class FSImageFormat {
       DataOutputStream out = new DataOutputStream(fos);
       try {
         out.writeInt(FSConstants.LAYOUT_VERSION);
-        out.writeInt(sourceNamesystem.getFSImage().getStorage().getNamespaceID()); // TODO bad dependency
+        out.writeInt(sourceNamesystem.getFSImage()
+                     .getStorage().getNamespaceID()); // TODO bad dependency
         out.writeLong(fsDir.rootDir.numItemsInTree());
         out.writeLong(sourceNamesystem.getGenerationStamp());
+        long txid = sourceNamesystem.getEditLog().getLastWrittenTxId();
+        out.writeLong(txid);
 
         // write compression info and set up compressed stream
         out = compression.writeHeaderAndWrapStream(fos);

src/java/org/apache/hadoop/hdfs/server/namenode/NNStorage.java

@@ -67,6 +67,7 @@ public class NNStorage extends Storage implements Closeable {
   private static final Log LOG = LogFactory.getLog(NNStorage.class.getName());
 
   static final String MESSAGE_DIGEST_PROPERTY = "imageMD5Digest";
+  static final String CHECKPOINT_TXID_PROPERTY = "checkpointTxId";
 
   //
   // The filenames used for storing the images
@@ -151,6 +152,13 @@ public class NNStorage extends Storage implements Closeable {
 
   private long checkpointTime = -1L;  // The age of the image
 
+  /**
+   * TxId of the last transaction that was included in the most
+   * recent fsimage file. This does not include any transactions
+   * that have since been written to the edit log.
+   */
+  protected long checkpointTxId;
+
   /**
    * list of failed (and thus removed) storages
    */
@@ -492,6 +500,20 @@ public class NNStorage extends Storage implements Closeable {
     checkpointTime = newCpT;
   }
 
+  /**
+   * Set the transaction ID of the last checkpoint
+   */
+  void setCheckpointTxId(long checkpointTxId) {
+    this.checkpointTxId = checkpointTxId;
+  }
+
+  /**
+   * Return the transaction ID of the last checkpoint.
+   */
+  long getCheckpointTxId() {
+    return checkpointTxId;
+  }
+
   /**
    * Set the current checkpoint time. Writes the new checkpoint
    * time to all available storage directories.
@@ -701,6 +723,21 @@ public class NNStorage extends Storage implements Closeable {
           " has image MD5 digest when version is " + layoutVersion);
     }
 
+    String sCheckpointId = props.getProperty(CHECKPOINT_TXID_PROPERTY);
+    if (layoutVersion <= -28) {
+      if (sCheckpointId == null) {
+        throw new InconsistentFSStateException(sd.getRoot(),
+            "file " + STORAGE_FILE_VERSION
+            + " does not have the checkpoint transaction id set.");
+      }
+      this.checkpointTxId = Long.valueOf(sCheckpointId);
+    } else if (sCheckpointId != null) {
+      throw new InconsistentFSStateException(sd.getRoot(),
+          "file " + STORAGE_FILE_VERSION +
+          " has checkpoint transaction id when version is " 
+          + layoutVersion);
+    }
+
     this.setCheckpointTime(readCheckpointTime(sd));
   }
 
@@ -736,7 +773,7 @@ public class NNStorage extends Storage implements Closeable {
     }
 
     props.setProperty(MESSAGE_DIGEST_PROPERTY, imageDigest.toString());
-
+    props.setProperty(CHECKPOINT_TXID_PROPERTY, String.valueOf(checkpointTxId));
     writeCheckpointTime(sd);
   }
 

src/java/org/apache/hadoop/hdfs/server/namenode/SecondaryNameNode.java

@@ -711,9 +711,10 @@ public class SecondaryNameNode implements Runnable {
         sdEdits = it.next();
       if (sdEdits == null)
         throw new IOException("Could not locate checkpoint edits");
+      
+      this.getStorage().setStorageInfo(sig);
+      this.getStorage().setImageDigest(sig.getImageDigest());
       if (loadImage) {
-        // to avoid assert in loadFSImage()
-        this.getStorage().layoutVersion = -1;
         loadFSImage(NNStorage.getStorageFile(sdName, NameNodeFile.IMAGE));
       }
       loadFSEdits(sdEdits);

src/java/org/apache/hadoop/hdfs/tools/offlineEditsViewer/EditsElement.java

@@ -31,6 +31,7 @@ public enum EditsElement {
   EDITS_VERSION,
   RECORD,
   OPCODE,
+  TRANSACTION_ID,
   DATA,
     // elements in the data part of the editLog records
     LENGTH,

src/java/org/apache/hadoop/hdfs/tools/offlineEditsViewer/EditsLoaderCurrent.java

@@ -77,7 +77,9 @@ class EditsLoaderCurrent implements EditsLoader {
    * Visit OP_INVALID
    */
   private void visit_OP_INVALID() throws IOException {
-    ; // nothing to do, this op code has no data
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
   }
 
   /**
@@ -101,6 +103,9 @@ class EditsLoaderCurrent implements EditsLoader {
    */
   private void visit_OP_ADD_or_OP_CLOSE(FSEditLogOpCodes editsOpCode)
     throws IOException {
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
 
     IntToken opAddLength = v.visitInt(EditsElement.LENGTH);
     // this happens if the edits is not properly ended (-1 op code),
@@ -144,6 +149,10 @@ class EditsLoaderCurrent implements EditsLoader {
    * Visit OP_RENAME_OLD
    */
   private void visit_OP_RENAME_OLD() throws IOException {
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
+
     v.visitInt(        EditsElement.LENGTH);
     v.visitStringUTF8( EditsElement.SOURCE);
     v.visitStringUTF8( EditsElement.DESTINATION);
@@ -154,6 +163,10 @@ class EditsLoaderCurrent implements EditsLoader {
    * Visit OP_DELETE
    */
   private void visit_OP_DELETE() throws IOException {
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
+
     v.visitInt(        EditsElement.LENGTH);
     v.visitStringUTF8( EditsElement.PATH);
     v.visitStringUTF8( EditsElement.TIMESTAMP);
@@ -163,6 +176,10 @@ class EditsLoaderCurrent implements EditsLoader {
    * Visit OP_MKDIR
    */
   private void visit_OP_MKDIR() throws IOException {
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
+
     v.visitInt(        EditsElement.LENGTH);
     v.visitStringUTF8( EditsElement.PATH);
     v.visitStringUTF8( EditsElement.TIMESTAMP);
@@ -181,6 +198,10 @@ class EditsLoaderCurrent implements EditsLoader {
    * Visit OP_SET_REPLICATION
    */
   private void visit_OP_SET_REPLICATION() throws IOException {
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
+
     v.visitStringUTF8(EditsElement.PATH);
     v.visitStringUTF8(EditsElement.REPLICATION);
   }
@@ -189,6 +210,10 @@ class EditsLoaderCurrent implements EditsLoader {
    * Visit OP_SET_PERMISSIONS
    */
   private void visit_OP_SET_PERMISSIONS() throws IOException {
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
+
     v.visitStringUTF8( EditsElement.PATH);
     v.visitShort(      EditsElement.FS_PERMISSIONS);
   }
@@ -197,6 +222,10 @@ class EditsLoaderCurrent implements EditsLoader {
    * Visit OP_SET_OWNER
    */
   private void visit_OP_SET_OWNER() throws IOException {
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
+
     v.visitStringUTF8(EditsElement.PATH);
     v.visitStringUTF8(EditsElement.USERNAME);
     v.visitStringUTF8(EditsElement.GROUPNAME);
@@ -206,6 +235,10 @@ class EditsLoaderCurrent implements EditsLoader {
    * Visit OP_SET_GENSTAMP
    */
   private void visit_OP_SET_GENSTAMP() throws IOException {
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
+
     v.visitLong(EditsElement.GENERATION_STAMP);
   }
 
@@ -213,6 +246,10 @@ class EditsLoaderCurrent implements EditsLoader {
    * Visit OP_TIMES
    */
   private void visit_OP_TIMES() throws IOException {
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
+
     v.visitInt(        EditsElement.LENGTH);
     v.visitStringUTF8( EditsElement.PATH);
     v.visitStringUTF8( EditsElement.MTIME);
@@ -223,6 +260,10 @@ class EditsLoaderCurrent implements EditsLoader {
    * Visit OP_SET_QUOTA
    */
   private void visit_OP_SET_QUOTA() throws IOException {
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
+
     v.visitStringUTF8( EditsElement.PATH);
     v.visitLong(       EditsElement.NS_QUOTA);
     v.visitLong(       EditsElement.DS_QUOTA);
@@ -232,6 +273,10 @@ class EditsLoaderCurrent implements EditsLoader {
    * Visit OP_RENAME
    */
   private void visit_OP_RENAME() throws IOException {
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
+
     if(editsVersion > -21) {
       throw new IOException("Unexpected op code " + FSEditLogOpCodes.OP_RENAME
         + " for edit log version " + editsVersion
@@ -248,6 +293,10 @@ class EditsLoaderCurrent implements EditsLoader {
    * Visit OP_CONCAT_DELETE
    */
   private void visit_OP_CONCAT_DELETE() throws IOException {
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
+
     if(editsVersion > -22) {
       throw new IOException("Unexpected op code "
         + FSEditLogOpCodes.OP_CONCAT_DELETE
@@ -268,6 +317,10 @@ class EditsLoaderCurrent implements EditsLoader {
    * Visit OP_SYMLINK
    */
   private void visit_OP_SYMLINK() throws IOException {
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
+
     v.visitInt(        EditsElement.LENGTH);
     v.visitStringUTF8( EditsElement.SOURCE);
     v.visitStringUTF8( EditsElement.DESTINATION);
@@ -287,21 +340,25 @@ class EditsLoaderCurrent implements EditsLoader {
    * Visit OP_GET_DELEGATION_TOKEN
    */
   private void visit_OP_GET_DELEGATION_TOKEN() throws IOException {
-      if(editsVersion > -24) {
-        throw new IOException("Unexpected op code "
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
+    
+    if(editsVersion > -24) {
+      throw new IOException("Unexpected op code "
           + FSEditLogOpCodes.OP_GET_DELEGATION_TOKEN
           + " for edit log version " + editsVersion
           + " (op code 18 only expected for 24 and later)");
-      }
-      v.visitByte(       EditsElement.T_VERSION);
-      v.visitStringText( EditsElement.T_OWNER);
-      v.visitStringText( EditsElement.T_RENEWER);
-      v.visitStringText( EditsElement.T_REAL_USER);
-      v.visitVLong(      EditsElement.T_ISSUE_DATE);
-      v.visitVLong(      EditsElement.T_MAX_DATE);
-      v.visitVInt(       EditsElement.T_SEQUENCE_NUMBER);
-      v.visitVInt(       EditsElement.T_MASTER_KEY_ID);
-      v.visitStringUTF8( EditsElement.T_EXPIRY_TIME);
+    }
+    v.visitByte(       EditsElement.T_VERSION);
+    v.visitStringText( EditsElement.T_OWNER);
+    v.visitStringText( EditsElement.T_RENEWER);
+    v.visitStringText( EditsElement.T_REAL_USER);
+    v.visitVLong(      EditsElement.T_ISSUE_DATE);
+    v.visitVLong(      EditsElement.T_MAX_DATE);
+    v.visitVInt(       EditsElement.T_SEQUENCE_NUMBER);
+    v.visitVInt(       EditsElement.T_MASTER_KEY_ID);
+    v.visitStringUTF8( EditsElement.T_EXPIRY_TIME);
   }
 
   /**
@@ -309,22 +366,25 @@ class EditsLoaderCurrent implements EditsLoader {
    */
   private void visit_OP_RENEW_DELEGATION_TOKEN()
     throws IOException {
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
 
-      if(editsVersion > -24) {
-        throw new IOException("Unexpected op code "
+    if(editsVersion > -24) {
+      throw new IOException("Unexpected op code "
           + FSEditLogOpCodes.OP_RENEW_DELEGATION_TOKEN
           + " for edit log version " + editsVersion
           + " (op code 19 only expected for 24 and later)");
-      }
-      v.visitByte(       EditsElement.T_VERSION);
-      v.visitStringText( EditsElement.T_OWNER);
-      v.visitStringText( EditsElement.T_RENEWER);
-      v.visitStringText( EditsElement.T_REAL_USER);
-      v.visitVLong(      EditsElement.T_ISSUE_DATE);
-      v.visitVLong(      EditsElement.T_MAX_DATE);
-      v.visitVInt(       EditsElement.T_SEQUENCE_NUMBER);
-      v.visitVInt(       EditsElement.T_MASTER_KEY_ID);
-      v.visitStringUTF8( EditsElement.T_EXPIRY_TIME);
+    }
+    v.visitByte(       EditsElement.T_VERSION);
+    v.visitStringText( EditsElement.T_OWNER);
+    v.visitStringText( EditsElement.T_RENEWER);
+    v.visitStringText( EditsElement.T_REAL_USER);
+    v.visitVLong(      EditsElement.T_ISSUE_DATE);
+    v.visitVLong(      EditsElement.T_MAX_DATE);
+    v.visitVInt(       EditsElement.T_SEQUENCE_NUMBER);
+    v.visitVInt(       EditsElement.T_MASTER_KEY_ID);
+    v.visitStringUTF8( EditsElement.T_EXPIRY_TIME);
   }
 
   /**
@@ -332,21 +392,24 @@ class EditsLoaderCurrent implements EditsLoader {
    */
   private void visit_OP_CANCEL_DELEGATION_TOKEN()
     throws IOException {
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
 
-      if(editsVersion > -24) {
-        throw new IOException("Unexpected op code "
+    if(editsVersion > -24) {
+      throw new IOException("Unexpected op code "
           + FSEditLogOpCodes.OP_CANCEL_DELEGATION_TOKEN
           + " for edit log version " + editsVersion
           + " (op code 20 only expected for 24 and later)");
-      }
-      v.visitByte(       EditsElement.T_VERSION);
-      v.visitStringText( EditsElement.T_OWNER);
-      v.visitStringText( EditsElement.T_RENEWER);
-      v.visitStringText( EditsElement.T_REAL_USER);
-      v.visitVLong(      EditsElement.T_ISSUE_DATE);
-      v.visitVLong(      EditsElement.T_MAX_DATE);
-      v.visitVInt(       EditsElement.T_SEQUENCE_NUMBER);
-      v.visitVInt(       EditsElement.T_MASTER_KEY_ID);
+    }
+    v.visitByte(       EditsElement.T_VERSION);
+    v.visitStringText( EditsElement.T_OWNER);
+    v.visitStringText( EditsElement.T_RENEWER);
+    v.visitStringText( EditsElement.T_REAL_USER);
+    v.visitVLong(      EditsElement.T_ISSUE_DATE);
+    v.visitVLong(      EditsElement.T_MAX_DATE);
+    v.visitVInt(       EditsElement.T_SEQUENCE_NUMBER);
+    v.visitVInt(       EditsElement.T_MASTER_KEY_ID);
   }
 
   /**
@@ -354,17 +417,20 @@ class EditsLoaderCurrent implements EditsLoader {
    */
   private void visit_OP_UPDATE_MASTER_KEY()
     throws IOException {
-
-      if(editsVersion > -24) {
-        throw new IOException("Unexpected op code "
+    if(editsVersion <= -28) {
+      v.visitLong(EditsElement.TRANSACTION_ID);
+    }
+    
+    if(editsVersion > -24) {
+      throw new IOException("Unexpected op code "
           + FSEditLogOpCodes.OP_UPDATE_MASTER_KEY
           + " for edit log version " + editsVersion
           + "(op code 21 only expected for 24 and later)");
-      }
-      v.visitVInt(  EditsElement.KEY_ID);
-      v.visitVLong( EditsElement.KEY_EXPIRY_DATE);
-      VIntToken blobLengthToken = v.visitVInt(EditsElement.KEY_LENGTH);
-      v.visitBlob(EditsElement.KEY_BLOB, blobLengthToken.value);
+    }
+    v.visitVInt(  EditsElement.KEY_ID);
+    v.visitVLong( EditsElement.KEY_EXPIRY_DATE);
+    VIntToken blobLengthToken = v.visitVInt(EditsElement.KEY_LENGTH);
+    v.visitBlob(EditsElement.KEY_BLOB, blobLengthToken.value);
   }
 
   private void visitOpCode(FSEditLogOpCodes editsOpCode)

src/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/ImageLoaderCurrent.java

@@ -156,6 +156,10 @@ class ImageLoaderCurrent implements ImageLoader {
 
       v.visit(ImageElement.GENERATION_STAMP, in.readLong());
 
+      if (imageVersion <= -28) {
+        v.visit(ImageElement.TRANSACTION_ID, in.readLong());
+      }
+
       if (imageVersion <= -25) {
         boolean isCompressed = in.readBoolean();
         v.visit(ImageElement.IS_COMPRESSED, imageVersion);

src/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/ImageVisitor.java

@@ -71,7 +71,8 @@ abstract class ImageVisitor {
     NUM_DELEGATION_TOKENS,
     DELEGATION_TOKENS,
     DELEGATION_TOKEN_IDENTIFIER,
-    DELEGATION_TOKEN_EXPIRY_TIME
+    DELEGATION_TOKEN_EXPIRY_TIME,
+    TRANSACTION_ID
   }
   
   /**

src/test/aop/org/apache/hadoop/fs/TestFiRename.java

@@ -24,6 +24,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Options.Rename;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
 import org.junit.After;
@@ -60,7 +61,6 @@ public class TestFiRename {
   
   private static Configuration CONF = new Configuration();
   static {
-    CONF.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, 1);
     CONF.setInt("io.bytes.per.checksum", 1);
   }
 

src/test/hdfs/org/apache/hadoop/hdfs/server/namenode/TestBackupNode.java

@@ -171,6 +171,14 @@ public class TestBackupNode extends TestCase {
       //
       backup = startBackupNode(conf, op, 1);
       waitCheckpointDone(backup);
+
+      for (int i = 0; i < 10; i++) {
+        writeFile(fileSys, new Path("file_" + i), replication);
+      }
+
+      backup.doCheckpoint();
+      waitCheckpointDone(backup);
+
     } catch(IOException e) {
       LOG.error("Error in TestBackupNode:", e);
       assertTrue(e.getLocalizedMessage(), false);

src/test/hdfs/org/apache/hadoop/hdfs/server/namenode/TestEditLog.java

@@ -30,12 +30,15 @@ import org.apache.hadoop.fs.permission.*;
 
 import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
 import org.apache.hadoop.hdfs.server.namenode.EditLogFileInputStream;
 import org.apache.hadoop.hdfs.server.common.Storage.StorageDirectory;
 import org.apache.hadoop.hdfs.server.namenode.NNStorage.NameNodeDirType;
 import org.apache.hadoop.hdfs.server.namenode.NNStorage.NameNodeFile;
 import org.apache.hadoop.hdfs.server.namenode.metrics.NameNodeMetrics;
 import org.apache.hadoop.metrics.util.MetricsTimeVaryingInt;
+
+import org.apache.hadoop.util.StringUtils;
  
 import org.mockito.Mockito;
 
@@ -50,6 +53,26 @@ public class TestEditLog extends TestCase {
   static final int NUM_TRANSACTIONS = 100;
   static final int NUM_THREADS = 100;
 
+  /** An edits log with 3 edits from 0.20 - the result of
+   * a fresh namesystem followed by hadoop fs -touchz /myfile */
+  static final byte[] HADOOP20_SOME_EDITS =
+    StringUtils.hexStringToByte((
+        "ffff ffed 0a00 0000 0000 03fa e100 0000" +
+        "0005 0007 2f6d 7966 696c 6500 0133 000d" +
+        "3132 3932 3331 3634 3034 3138 3400 0d31" +
+        "3239 3233 3136 3430 3431 3834 0009 3133" +
+        "3432 3137 3732 3800 0000 0004 746f 6464" +
+        "0a73 7570 6572 6772 6f75 7001 a400 1544" +
+        "4653 436c 6965 6e74 5f2d 3136 3136 3535" +
+        "3738 3931 000b 3137 322e 3239 2e35 2e33" +
+        "3209 0000 0005 0007 2f6d 7966 696c 6500" +
+        "0133 000d 3132 3932 3331 3634 3034 3138" +
+        "3400 0d31 3239 3233 3136 3430 3431 3834" +
+        "0009 3133 3432 3137 3732 3800 0000 0004" +
+        "746f 6464 0a73 7570 6572 6772 6f75 7001" +
+        "a4ff 0000 0000 0000 0000 0000 0000 0000"
+    ).replace(" ",""));
+
   //
   // an object that does a bunch of transactions
   //
@@ -80,6 +103,51 @@ public class TestEditLog extends TestCase {
     }
   }
 
+  /**
+   * Test case for an empty edit log from a prior version of Hadoop.
+   */
+  public void testPreTxIdEditLogNoEdits() throws Exception {
+    FSNamesystem namesys = Mockito.mock(FSNamesystem.class);
+    int numEdits = testLoad(
+        StringUtils.hexStringToByte("ffffffed"), // just version number
+        namesys);
+    assertEquals(0, numEdits);
+  }
+  
+  /**
+   * Test case for loading a very simple edit log from a format
+   * prior to the inclusion of edit transaction IDs in the log.
+   */
+  public void testPreTxidEditLogWithEdits() throws Exception {
+    Configuration conf = new HdfsConfiguration();
+    MiniDFSCluster cluster = null;
+
+    try {
+      cluster = new MiniDFSCluster.Builder(conf).numDataNodes(0).build();
+      cluster.waitActive();
+      final FSNamesystem namesystem = cluster.getNamesystem();
+
+      int numEdits = testLoad(HADOOP20_SOME_EDITS, namesystem);
+      assertEquals(3, numEdits);
+      // Sanity check the edit
+      HdfsFileStatus fileInfo = namesystem.getFileInfo("/myfile", false);
+      assertEquals("supergroup", fileInfo.getGroup());
+      assertEquals(3, fileInfo.getReplication());
+    } finally {
+      cluster.shutdown();
+    }
+  }
+  
+  private int testLoad(byte[] data, FSNamesystem namesys) throws IOException {
+    FSEditLogLoader loader = new FSEditLogLoader(namesys);
+    EditLogInputStream mockStream = Mockito.mock(EditLogInputStream.class);
+    ByteArrayInputStream bais = new ByteArrayInputStream(data);
+    Mockito.doReturn(new DataInputStream(bais))
+      .when(mockStream).getDataInputStream();
+    
+    return loader.loadFSEdits(mockStream, 1);
+  }
+
   /**
    * Tests transaction logging in dfs.
    */
@@ -145,13 +213,13 @@ public class TestEditLog extends TestCase {
       // If there were any corruptions, it is likely that the reading in
       // of these transactions will throw an exception.
       //
-      FSEditLogLoader loader = new FSEditLogLoader(namesystem);
       for (Iterator<StorageDirectory> it = 
               fsimage.getStorage().dirIterator(NameNodeDirType.EDITS); it.hasNext();) {
+        FSEditLogLoader loader = new FSEditLogLoader(namesystem);
         File editFile = NNStorage.getStorageFile(it.next(), NameNodeFile.EDITS);
         System.out.println("Verifying file: " + editFile);
         int numEdits = loader.loadFSEdits(
-                                  new EditLogFileInputStream(editFile));
+            new EditLogFileInputStream(editFile), 1);
         int numLeases = namesystem.leaseManager.countLease();
         System.out.println("Number of outstanding leases " + numLeases);
         assertEquals(0, numLeases);

src/test/hdfs/org/apache/hadoop/hdfs/server/namenode/TestEditLogRace.java

@@ -187,20 +187,27 @@ public class TestEditLogRace {
 
       startTransactionWorkers(namesystem, caughtErr);
 
+      long previousLogTxId = 1;
+
       for (int i = 0; i < NUM_ROLLS && caughtErr.get() == null; i++) {
         try {
           Thread.sleep(20);
         } catch (InterruptedException e) {}
 
         LOG.info("Starting roll " + i + ".");
-        editLog.rollEditLog();
-        LOG.info("Roll complete " + i + ".");
+        long nextTxId = 0L;
+        synchronized (editLog) {
+          editLog.rollEditLog();
+          nextTxId = editLog.getLastWrittenTxId() + 1;
+        }
+        LOG.info("Roll " + i + " complete before txid " + nextTxId);
 
-        verifyEditLogs(namesystem, fsimage);
+        verifyEditLogs(namesystem, fsimage, previousLogTxId);
 
         LOG.info("Starting purge " + i + ".");
         editLog.purgeEditLog();
         LOG.info("Complete purge " + i + ".");
+        previousLogTxId = nextTxId;
       }
     } finally {
       stopTransactionWorkers();
@@ -213,7 +220,8 @@ public class TestEditLogRace {
     }
   }
 
-  private void verifyEditLogs(FSNamesystem namesystem, FSImage fsimage)
+  private void verifyEditLogs(FSNamesystem namesystem, FSImage fsimage, 
+                              long startingTxId)
     throws IOException {
     // Verify that we can read in all the transactions that we have written.
     // If there were any corruptions, it is likely that the reading in
@@ -222,8 +230,9 @@ public class TestEditLogRace {
            fsimage.getStorage().dirIterator(NameNodeDirType.EDITS); it.hasNext();) {
       File editFile = fsimage.getStorage().getStorageFile(it.next(), NameNodeFile.EDITS);
       System.out.println("Verifying file: " + editFile);
-      int numEdits = new FSEditLogLoader(namesystem).loadFSEdits(
-        new EditLogFileInputStream(editFile));
+      FSEditLogLoader loader = new FSEditLogLoader(namesystem);
+      int numEdits = loader.loadFSEdits(new EditLogFileInputStream(editFile), 
+                                        startingTxId);
       System.out.println("Number of edits: " + numEdits);
     }
   }
@@ -265,14 +274,19 @@ public class TestEditLogRace {
         namesystem.enterSafeMode();
 
         // Verify edit logs before the save
-        verifyEditLogs(namesystem, fsimage);
+        // They should start with the first edit after the checkpoint
+        verifyEditLogs(namesystem, fsimage, 
+                       fsimage.getStorage().getCheckpointTxId() + 1);
 
         LOG.info("Save " + i + ": saving namespace");
         namesystem.saveNamespace();
         LOG.info("Save " + i + ": leaving safemode");
 
         // Verify that edit logs post save are also not corrupt
-        verifyEditLogs(namesystem, fsimage);
+        verifyEditLogs(namesystem, fsimage, 
+                       fsimage.getStorage().getCheckpointTxId() + 1);
+        assertEquals(fsimage.getStorage().getCheckpointTxId(), 
+                     editLog.getLastWrittenTxId());
 
         namesystem.leaveSafeMode(false);
         LOG.info("Save " + i + ": complete");
@@ -393,7 +407,9 @@ public class TestEditLogRace {
       doAnEditThread.join();
       assertNull(deferredException.get());
 
-      verifyEditLogs(namesystem, fsimage);
+      // Since we did one transaction before the saveNamespace, the new
+      // log starts at txid 2
+      verifyEditLogs(namesystem, fsimage, 2);
     } finally {
       LOG.info("Closing namesystem");
       if(namesystem != null) namesystem.close();
@@ -478,7 +494,9 @@ public class TestEditLogRace {
       doAnEditThread.join();
       assertNull(deferredException.get());
 
-      verifyEditLogs(namesystem, fsimage);
+      // Since we did one transaction before the saveNamespace, the new
+      // log starts at txid 2
+      verifyEditLogs(namesystem, fsimage, 2);
     } finally {
       LOG.info("Closing namesystem");
       if(namesystem != null) namesystem.close();

src/test/hdfs/org/apache/hadoop/hdfs/server/namenode/TestSaveNamespace.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hdfs.server.namenode;
 
 import static org.apache.hadoop.hdfs.server.common.Util.fileAsURI;
+import static org.junit.Assert.*;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.anyObject;
@@ -30,6 +31,7 @@ import java.io.IOException;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.commons.logging.impl.Log4JLogger;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.permission.FsPermission;
@@ -41,6 +43,7 @@ import org.apache.hadoop.hdfs.protocol.FSConstants.SafeModeAction;
 import org.apache.hadoop.hdfs.server.common.HdfsConstants.NamenodeRole;
 import org.apache.hadoop.hdfs.server.common.Storage.StorageDirectory;
 import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.log4j.Level;
 import org.junit.Test;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
@@ -57,6 +60,10 @@ import org.mockito.stubbing.Answer;
  * </ol>
  */
 public class TestSaveNamespace {
+  static {
+    ((Log4JLogger)FSImage.LOG).getLogger().setLevel(Level.ALL);
+  }
+  
   private static final Log LOG = LogFactory.getLog(TestSaveNamespace.class);
 
   private static class FaultySaveImage implements Answer<Void> {
@@ -307,7 +314,37 @@ public class TestSaveNamespace {
       }
     }
   }
+  
+  @Test
+  public void testTxIdPersistence() throws Exception {
+    Configuration conf = getConf();
+    NameNode.initMetrics(conf, NamenodeRole.ACTIVE);
+    NameNode.format(conf);
+    FSNamesystem fsn = new FSNamesystem(conf);
 
+    try {
+      assertEquals(0, fsn.getEditLog().getLastWrittenTxId());
+      doAnEdit(fsn, 1);
+      assertEquals(1, fsn.getEditLog().getLastWrittenTxId());
+      
+      fsn.setSafeMode(SafeModeAction.SAFEMODE_ENTER);
+      fsn.saveNamespace();
+      
+      // Shut down and restart
+      fsn.getFSImage().close();
+      fsn.close();
+      fsn = null;
+      
+      fsn = new FSNamesystem(conf);
+      assertEquals(1, fsn.getEditLog().getLastWrittenTxId());
+      
+    } finally {
+      if (fsn != null) {
+        fsn.close();
+      }
+    }
+  }
+  
   private void doAnEdit(FSNamesystem fsn, int id) throws IOException {
     // Make an edit
     fsn.mkdirs(

src/test/hdfs/org/apache/hadoop/hdfs/server/namenode/TestSecurityTokenEditLog.java

@@ -134,15 +134,15 @@ public class TestSecurityTokenEditLog extends TestCase {
       // If there were any corruptions, it is likely that the reading in
       // of these transactions will throw an exception.
       //
-      FSEditLogLoader loader = new FSEditLogLoader(namesystem);
       namesystem.getDelegationTokenSecretManager().stopThreads();
       int numKeys = namesystem.getDelegationTokenSecretManager().getNumberOfKeys();
       for (Iterator<StorageDirectory> it = 
              fsimage.getStorage().dirIterator(NameNodeDirType.EDITS); it.hasNext();) {
+        FSEditLogLoader loader = new FSEditLogLoader(namesystem);
         File editFile = fsimage.getStorage().getStorageFile(it.next(), NameNodeFile.EDITS);
         System.out.println("Verifying file: " + editFile);
         int numEdits = loader.loadFSEdits(
-                                  new EditLogFileInputStream(editFile));
+            new EditLogFileInputStream(editFile), 1);
         assertTrue("Verification for " + editFile + " failed. " +
                    "Expected " + (NUM_THREADS * opsPerTrans * NUM_TRANSACTIONS + numKeys) + " transactions. "+
                    "Found " + numEdits + " transactions.",