HDFS-2088. Move edits log archiving logic into FSEditLog/JournalManager. Contributed by Todd Lipcon.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/HDFS-1073@1139456 13f79535-47bb-0310-9956-ffa450edef68

hdfs/CHANGES.HDFS-1073.txt

@@ -59,4 +59,4 @@ HDFS-2077. Address checkpoint upload when one of the storage dirs is failed
            (todd)
 HDFS-2078. NameNode should not clear directory when restoring removed storage.
            (todd)
-           
+HDFS-2088. Move edits log archiving logic into FSEditLog/JournalManager (todd)

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

@@ -37,6 +37,7 @@ import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifie
 import org.apache.hadoop.hdfs.server.common.Storage.StorageDirectory;
 import static org.apache.hadoop.hdfs.server.common.Util.now;
 import org.apache.hadoop.hdfs.server.namenode.NNStorage.NameNodeDirType;
+import org.apache.hadoop.hdfs.server.namenode.NNStorageArchivalManager.StorageArchiver;
 import org.apache.hadoop.hdfs.server.namenode.metrics.NameNodeMetrics;
 import org.apache.hadoop.hdfs.server.protocol.NamenodeRegistration;
 import org.apache.hadoop.hdfs.server.protocol.RemoteEditLogManifest;
@@ -842,6 +843,25 @@ public class FSEditLog  {
     state = State.BETWEEN_LOG_SEGMENTS;
   }
 
+  /**
+   * Archive any log files that are older than the given txid.
+   */
+  public void archiveLogsOlderThan(
+      final long minTxIdToKeep, final StorageArchiver archiver) {
+    assert curSegmentTxId == FSConstants.INVALID_TXID || // on format this is no-op
+      minTxIdToKeep <= curSegmentTxId :
+      "cannot archive logs older than txid " + minTxIdToKeep +
+      " when current segment starts at " + curSegmentTxId;
+    
+    mapJournalsAndReportErrors(new JournalClosure() {
+      @Override
+      public void apply(JournalAndStream jas) throws IOException {
+        jas.manager.archiveLogsOlderThan(minTxIdToKeep, archiver);
+      }
+    }, "archiving logs older than " + minTxIdToKeep);
+  }
+
+  
   /**
    * The actual sync activity happens while not synchronized on this object.
    * Thus, synchronized activities that require that they are not concurrent

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

@@ -95,7 +95,8 @@ public class FSImage implements Closeable {
   /**
    * Can fs-image be rolled?
    */
-  volatile protected CheckpointStates ckptState = FSImage.CheckpointStates.START; 
+  volatile protected CheckpointStates ckptState = FSImage.CheckpointStates.START;
+  private final NNStorageArchivalManager archivalManager; 
 
   /**
    * Construct an FSImage.
@@ -152,6 +153,8 @@ public class FSImage implements Closeable {
 
     this.editLog = new FSEditLog(storage);
     setFSNamesystem(ns);
+    
+    archivalManager = new NNStorageArchivalManager(conf, storage, editLog);
   }
 
   protected FSNamesystem getFSNamesystem() {
@@ -825,7 +828,19 @@ public class FSImage implements Closeable {
     
     // Since we now have a new checkpoint, we can clean up some
     // old edit logs and checkpoints.
-    storage.archiveOldStorage();
+    archiveOldStorage();
+  }
+
+  /**
+   * Archive any files in the storage directories that are no longer
+   * necessary.
+   */
+  public void archiveOldStorage() {
+    try {
+      archivalManager.archiveOldStorage();
+    } catch (Exception e) {
+      LOG.warn("Unable to archive old storage", e);
+    }
   }
 
   /**

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

@@ -43,6 +43,7 @@ import org.apache.hadoop.hdfs.server.namenode.NNStorage.NameNodeFile;
 import org.apache.hadoop.hdfs.server.protocol.RemoteEditLog;
 import org.apache.hadoop.hdfs.server.protocol.RemoteEditLogManifest;
 
+import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
 
@@ -105,45 +106,7 @@ class FSImageTransactionalStorageInspector extends FSImageStorageInspector {
                    "not configured to contain images.");
         }
       }
-      
-      // Check for edits
-      Matcher editsMatch = EDITS_REGEX.matcher(name);
-      if (editsMatch.matches()) {
-        if (sd.getStorageDirType().isOfType(NameNodeDirType.EDITS)) {
-          try {
-            long startTxId = Long.valueOf(editsMatch.group(1));
-            long endTxId = Long.valueOf(editsMatch.group(2));
-            addEditLog(new FoundEditLog(sd, f, startTxId, endTxId));
-          } catch (NumberFormatException nfe) {
-            LOG.error("Edits file " + f + " has improperly formatted " +
-                      "transaction ID");
-            // skip
-          }          
-        } else {
-          LOG.warn("Found edits file at " + f + " but storage directory is " +
-                   "not configured to contain edits.");
-        }
-      }
-      
-      // Check for in-progress edits
-      Matcher inProgressEditsMatch = EDITS_INPROGRESS_REGEX.matcher(name);
-      if (inProgressEditsMatch.matches()) {
-        if (sd.getStorageDirType().isOfType(NameNodeDirType.EDITS)) {
-          try {
-            long startTxId = Long.valueOf(inProgressEditsMatch.group(1));
-            addEditLog(
-              new FoundEditLog(sd, f, startTxId, FoundEditLog.UNKNOWN_END));
-          } catch (NumberFormatException nfe) {
-            LOG.error("In-progress edits file " + f + " has improperly " +
-                      "formatted transaction ID");
-            // skip
-          }          
-        } else {
-          LOG.warn("Found in-progress edits file at " + f + " but " +
-                   "storage directory is not configured to contain edits.");
-        }
-      }
-      
+
       // Check for a seen_txid file, which marks a minimum transaction ID that
       // must be included in our load plan.
       try {
@@ -151,13 +114,58 @@ class FSImageTransactionalStorageInspector extends FSImageStorageInspector {
       } catch (IOException ioe) {
         LOG.warn("Unable to determine the max transaction ID seen by " + sd, ioe);
       }
-      
     }
-
+    
+    List<FoundEditLog> editLogs = matchEditLogs(filesInStorage);
+    if (sd.getStorageDirType().isOfType(NameNodeDirType.EDITS)) {
+      for (FoundEditLog log : editLogs) {
+        addEditLog(log);
+      }
+    } else if (!editLogs.isEmpty()){
+      LOG.warn("Found the following edit log file(s) in " + sd +
+          " even though it was not configured to store edits:\n" +
+          "  " + Joiner.on("\n  ").join(editLogs));
+          
+    }
+    
     // set finalized flag
     isUpgradeFinalized = isUpgradeFinalized && !sd.getPreviousDir().exists();
   }
 
+  static List<FoundEditLog> matchEditLogs(File[] filesInStorage) {
+    List<FoundEditLog> ret = Lists.newArrayList();
+    for (File f : filesInStorage) {
+      String name = f.getName();
+      // Check for edits
+      Matcher editsMatch = EDITS_REGEX.matcher(name);
+      if (editsMatch.matches()) {
+        try {
+          long startTxId = Long.valueOf(editsMatch.group(1));
+          long endTxId = Long.valueOf(editsMatch.group(2));
+          ret.add(new FoundEditLog(f, startTxId, endTxId));
+        } catch (NumberFormatException nfe) {
+          LOG.error("Edits file " + f + " has improperly formatted " +
+                    "transaction ID");
+          // skip
+        }          
+      }
+      
+      // Check for in-progress edits
+      Matcher inProgressEditsMatch = EDITS_INPROGRESS_REGEX.matcher(name);
+      if (inProgressEditsMatch.matches()) {
+        try {
+          long startTxId = Long.valueOf(inProgressEditsMatch.group(1));
+          ret.add(
+            new FoundEditLog(f, startTxId, FoundEditLog.UNKNOWN_END));
+        } catch (NumberFormatException nfe) {
+          LOG.error("In-progress edits file " + f + " has improperly " +
+                    "formatted transaction ID");
+          // skip
+        }          
+      }
+    }
+    return ret;
+  }
 
   private void addEditLog(FoundEditLog foundEditLog) {
     foundEditLogs.add(foundEditLog);
@@ -484,7 +492,6 @@ class FSImageTransactionalStorageInspector extends FSImageStorageInspector {
    * Record of an edit log that has been located and had its filename parsed.
    */
   static class FoundEditLog {
-    final StorageDirectory sd;
     File file;
     final long startTxId;
     long lastTxId;
@@ -494,13 +501,12 @@ class FSImageTransactionalStorageInspector extends FSImageStorageInspector {
     
     static final long UNKNOWN_END = -1;
     
-    FoundEditLog(StorageDirectory sd, File file,
+    FoundEditLog(File file,
         long startTxId, long endTxId) {
       assert endTxId == UNKNOWN_END || endTxId >= startTxId;
       assert startTxId > 0;
       assert file != null;
       
-      this.sd = sd;
       this.startTxId = startTxId;
       this.lastTxId = endTxId;
       this.file = file;

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

@@ -22,8 +22,12 @@ import org.apache.commons.logging.LogFactory;
 
 import java.io.File;
 import java.io.IOException;
+import java.util.List;
 
+import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.hdfs.server.common.Storage.StorageDirectory;
+import org.apache.hadoop.hdfs.server.namenode.FSImageTransactionalStorageInspector.FoundEditLog;
+import org.apache.hadoop.hdfs.server.namenode.NNStorageArchivalManager.StorageArchiver;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
@@ -85,4 +89,19 @@ public class FileJournalManager implements JournalManager {
   public void setOutputBufferCapacity(int size) {
     this.outputBufferCapacity = size;
   }
+
+  @Override
+  public void archiveLogsOlderThan(long minTxIdToKeep, StorageArchiver archiver)
+      throws IOException {
+    File[] files = FileUtil.listFiles(sd.getCurrentDir());
+    List<FoundEditLog> editLogs = 
+      FSImageTransactionalStorageInspector.matchEditLogs(files);
+    for (FoundEditLog log : editLogs) {
+      if (log.getStartTxId() < minTxIdToKeep &&
+          log.getLastTxId() < minTxIdToKeep) {
+        archiver.archiveLog(log);
+      }
+    }
+  }
+
 }

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

@@ -150,7 +150,7 @@ public class GetImageServlet extends HttpServlet {
               
               // Now that we have a new checkpoint, we might be able to
               // remove some old ones.
-              nnImage.getStorage().archiveOldStorage();
+              nnImage.archiveOldStorage();
             } finally {
               currentlyDownloadingCheckpoints.remove(txid);
             }

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

@@ -19,6 +19,8 @@ package org.apache.hadoop.hdfs.server.namenode;
 
 import java.io.IOException;
 
+import org.apache.hadoop.hdfs.server.namenode.NNStorageArchivalManager.StorageArchiver;
+
 /**
  * A JournalManager is responsible for managing a single place of storing
  * edit logs. It may correspond to multiple files, a backup node, etc.
@@ -38,4 +40,16 @@ public interface JournalManager {
    * Set the amount of memory that this stream should use to buffer edits
    */
   void setOutputBufferCapacity(int size);
+
+  /**
+   * The JournalManager may archive/purge any logs for transactions less than
+   * or equal to minImageTxId.
+   *
+   * @param minTxIdToKeep the earliest txid that must be retained after purging
+   *                      old logs
+   * @param archiver the archival implementation to use
+   * @throws IOException if purging fails
+   */
+  void archiveLogsOlderThan(long minTxIdToKeep, StorageArchiver archiver)
+    throws IOException;
 }

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

@@ -123,8 +123,6 @@ public class NNStorage extends Storage implements Closeable {
   private UpgradeManager upgradeManager = null;
   protected String blockpoolID = ""; // id of the block pool
   
-  private final NNStorageArchivalManager archivalManager;
-
   /**
    * flag that controls if we try to restore failed storages
    */
@@ -167,8 +165,6 @@ public class NNStorage extends Storage implements Closeable {
     storageDirs = new CopyOnWriteArrayList<StorageDirectory>();
     
     setStorageDirectories(imageDirs, editsDirs);
-    
-    archivalManager = new NNStorageArchivalManager(conf, this);
   }
 
   @Override // Storage
@@ -530,19 +526,6 @@ public class NNStorage extends Storage implements Closeable {
     }
   }
 
-  /**
-   * Archive any files in the storage directories that are no longer
-   * necessary.
-   */
-  public void archiveOldStorage() {
-    try {
-      archivalManager.archiveOldStorage();
-    } catch (Exception e) {
-      LOG.warn("Unable to archive old storage", e);
-    }
-  }
-
-
   /**
    * Generate new namespaceID.
    *

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

@@ -48,20 +48,24 @@ public class NNStorageArchivalManager {
   private static final Log LOG = LogFactory.getLog(NNStorageArchivalManager.class);
   private final NNStorage storage;
   private final StorageArchiver archiver;
+  private final FSEditLog editLog;
   
   public NNStorageArchivalManager(
       Configuration conf,
       NNStorage storage,
+      FSEditLog editLog,
       StorageArchiver archiver) {
     this.numCheckpointsToRetain = conf.getInt(
         DFSConfigKeys.DFS_NAMENODE_NUM_CHECKPOINTS_RETAINED_KEY,
         DFSConfigKeys.DFS_NAMENODE_NUM_CHECKPOINTS_RETAINED_DEFAULT);
     this.storage = storage;
+    this.editLog = editLog;
     this.archiver = archiver;
   }
   
-  public NNStorageArchivalManager(Configuration conf, NNStorage storage) {
-    this(conf, storage, new DeletionStorageArchiver());
+  public NNStorageArchivalManager(Configuration conf, NNStorage storage,
+      FSEditLog editLog) {
+    this(conf, storage, editLog, new DeletionStorageArchiver());
   }
 
   public void archiveOldStorage() throws IOException {
@@ -71,20 +75,12 @@ public class NNStorageArchivalManager {
 
     long minImageTxId = getImageTxIdToRetain(inspector);
     archiveCheckpointsOlderThan(inspector, minImageTxId);
-    archiveLogsOlderThan(inspector, minImageTxId);
+    // If fsimage_N is the image we want to keep, then we need to keep
+    // all txns > N. We can remove anything < N+1, since fsimage_N
+    // reflects the state up to and including N.
+    editLog.archiveLogsOlderThan(minImageTxId + 1, archiver);
   }
   
-  private void archiveLogsOlderThan(
-      FSImageTransactionalStorageInspector inspector,
-      long minImageTxId) {
-    for (FoundEditLog log : inspector.getFoundEditLogs()) {
-      if (log.getStartTxId() < minImageTxId) {
-        LOG.info("Purging old edit log " + log);
-        archiver.archiveLog(log);
-      }
-    }
-  }
-
   private void archiveCheckpointsOlderThan(
       FSImageTransactionalStorageInspector inspector,
       long minTxId) {

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

@@ -508,7 +508,7 @@ public class SecondaryNameNode implements Runnable {
     
     // Since we've successfully checkpointed, we can remove some old
     // image files
-    checkpointImage.getStorage().archiveOldStorage();
+    checkpointImage.archiveOldStorage();
     
     return loadImage;
   }

hdfs/src/test/hdfs/org/apache/hadoop/hdfs/MiniDFSCluster.java

@@ -58,6 +58,7 @@ import org.apache.hadoop.hdfs.server.datanode.DataNodeTestUtils;
 import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.hdfs.server.namenode.NameNodeAdapter;
+import org.apache.hadoop.hdfs.server.namenode.SecondaryNameNode;
 import org.apache.hadoop.hdfs.server.protocol.DatanodeProtocol;
 import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration;
 import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocol;

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

@@ -0,0 +1,133 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.namenode;
+
+import java.io.File;
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.HdfsConfiguration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.protocol.FSConstants.SafeModeAction;
+import org.apache.hadoop.test.GenericTestUtils;
+import org.junit.Test;
+
+import com.google.common.base.Joiner;
+
+import static org.apache.hadoop.test.GenericTestUtils.assertGlobEquals;
+
+
+/**
+ * Functional tests for NNStorageArchivalManager. This differs from
+ * {@link TestNNStorageArchivalManager} in that the other test suite
+ * is only unit/mock-based tests whereas this suite starts miniclusters,
+ * etc.
+ */
+public class TestNNStorageArchivalFunctional {
+
+  private static File TEST_ROOT_DIR =
+    new File(MiniDFSCluster.getBaseDirectory());
+  private static Log LOG = LogFactory.getLog(
+      TestNNStorageArchivalFunctional.class);
+
+ /**
+  * Test case where two directories are configured as NAME_AND_EDITS
+  * and one of them fails to save storage. Since the edits and image
+  * failure states are decoupled, the failure of image saving should
+  * not prevent the archival of logs from that dir.
+  */
+  @Test
+  public void testArchivingWithNameEditsDirAfterFailure()
+      throws IOException {
+    MiniDFSCluster cluster = null;    
+    Configuration conf = new HdfsConfiguration();
+
+    File sd0 = new File(TEST_ROOT_DIR, "nn0");
+    File sd1 = new File(TEST_ROOT_DIR, "nn1");
+    File cd0 = new File(sd0, "current");
+    File cd1 = new File(sd1, "current");
+    conf.set(DFSConfigKeys.DFS_NAMENODE_NAME_DIR_KEY,
+        Joiner.on(",").join(sd0, sd1));
+
+    try {
+      cluster = new MiniDFSCluster.Builder(conf)
+        .numDataNodes(0)
+        .manageNameDfsDirs(false)
+        .format(true).build();
+  
+      NameNode nn = cluster.getNameNode();
+
+      doSaveNamespace(nn);
+      LOG.info("After first save, images 0 and 2 should exist in both dirs");
+      assertGlobEquals(cd0, "fsimage_\\d*", "fsimage_0", "fsimage_2");
+      assertGlobEquals(cd1, "fsimage_\\d*", "fsimage_0", "fsimage_2");
+      assertGlobEquals(cd0, "edits_.*", "edits_1-2", "edits_inprogress_3");
+      assertGlobEquals(cd1, "edits_.*", "edits_1-2", "edits_inprogress_3");
+      
+      doSaveNamespace(nn);
+      LOG.info("After second save, image 0 should be archived, " +
+          "and image 4 should exist in both.");
+      assertGlobEquals(cd0, "fsimage_\\d*", "fsimage_2", "fsimage_4");
+      assertGlobEquals(cd1, "fsimage_\\d*", "fsimage_2", "fsimage_4");
+      assertGlobEquals(cd0, "edits_.*", "edits_3-4", "edits_inprogress_5");
+      assertGlobEquals(cd1, "edits_.*", "edits_3-4", "edits_inprogress_5");
+      
+      LOG.info("Failing first storage dir by chmodding it");
+      sd0.setExecutable(false);
+      doSaveNamespace(nn);      
+      LOG.info("Restoring accessibility of first storage dir");      
+      sd0.setExecutable(true);
+
+      LOG.info("nothing should have been archived in first storage dir");
+      assertGlobEquals(cd0, "fsimage_\\d*", "fsimage_2", "fsimage_4");
+      assertGlobEquals(cd0, "edits_.*", "edits_3-4", "edits_inprogress_5");
+
+      LOG.info("fsimage_2 should be archived in second storage dir");
+      assertGlobEquals(cd1, "fsimage_\\d*", "fsimage_4", "fsimage_6");
+      assertGlobEquals(cd1, "edits_.*", "edits_5-6", "edits_inprogress_7");
+
+      LOG.info("On next save, we should archive logs from the failed dir," +
+          " but not images, since the image directory is in failed state.");
+      doSaveNamespace(nn);
+      assertGlobEquals(cd1, "fsimage_\\d*", "fsimage_6", "fsimage_8");
+      assertGlobEquals(cd1, "edits_.*", "edits_7-8", "edits_inprogress_9");
+      assertGlobEquals(cd0, "fsimage_\\d*", "fsimage_2", "fsimage_4");
+      assertGlobEquals(cd0, "edits_.*", "edits_inprogress_9");
+      
+
+
+    } finally {
+      sd0.setExecutable(true);
+
+      LOG.info("Shutting down...");
+      if (cluster != null) {
+        cluster.shutdown();
+      }
+    }
+  }
+
+  private static void doSaveNamespace(NameNode nn) throws IOException {
+    LOG.info("Saving namespace...");
+    nn.setSafeMode(SafeModeAction.SAFEMODE_ENTER);
+    nn.saveNamespace();
+    nn.setSafeMode(SafeModeAction.SAFEMODE_LEAVE);
+  }
+}

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

@@ -169,7 +169,8 @@ public class TestNNStorageArchivalManager {
       ArgumentCaptor.forClass(FoundEditLog.class);    
 
     // Ask the manager to archive files we don't need any more
-    new NNStorageArchivalManager(conf, tc.mockStorage(), mockArchiver)
+    new NNStorageArchivalManager(conf,
+        tc.mockStorage(), tc.mockEditLog(), mockArchiver)
       .archiveOldStorage();
     
     // Verify that it asked the archiver to remove the correct files
@@ -208,6 +209,12 @@ public class TestNNStorageArchivalManager {
         this.type = type;
         files = Lists.newArrayList();
       }
+
+      StorageDirectory mockStorageDir() {
+        return TestFSImageStorageInspector.mockDirectory(
+            type, false,
+            files.toArray(new String[0]));
+      }
     }
 
     void addRoot(String root, NameNodeDirType dir) {
@@ -239,13 +246,40 @@ public class TestNNStorageArchivalManager {
     NNStorage mockStorage() throws IOException {
       List<StorageDirectory> sds = Lists.newArrayList();
       for (FakeRoot root : dirRoots.values()) {
-        StorageDirectory mockDir = TestFSImageStorageInspector.mockDirectory(
-            root.type, false,
-            root.files.toArray(new String[0]));
-        sds.add(mockDir);
+        sds.add(root.mockStorageDir());
       }
       return mockStorageForDirs(sds.toArray(new StorageDirectory[0]));
     }
+    
+    public FSEditLog mockEditLog() {
+      final List<JournalManager> jms = Lists.newArrayList();
+      for (FakeRoot root : dirRoots.values()) {
+        if (!root.type.isOfType(NameNodeDirType.EDITS)) continue;
+        
+        FileJournalManager fjm = new FileJournalManager(
+            root.mockStorageDir());
+        jms.add(fjm);
+      }
+
+      FSEditLog mockLog = Mockito.mock(FSEditLog.class);
+      Mockito.doAnswer(new Answer<Void>() {
+
+        @Override
+        public Void answer(InvocationOnMock invocation) throws Throwable {
+          Object[] args = invocation.getArguments();
+          assert args.length == 2;
+          long txId = (Long) args[0];
+          StorageArchiver archiver = (StorageArchiver) args[1];
+          
+          for (JournalManager jm : jms) {
+            jm.archiveLogsOlderThan(txId, archiver);
+          }
+          return null;
+        }
+      }).when(mockLog).archiveLogsOlderThan(
+          Mockito.anyLong(), (StorageArchiver) Mockito.anyObject());
+      return mockLog;
+    }
   }
 
   private static NNStorage mockStorageForDirs(final StorageDirectory ... mockDirs)

hdfs/src/test/hdfs/org/apache/hadoop/test/GenericTestUtils.java

@@ -19,9 +19,13 @@ package org.apache.hadoop.test;
 
 import java.io.File;
 import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 
 import org.apache.commons.logging.Log;
+import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocol;
 import org.apache.hadoop.util.StringUtils;
 import org.junit.Assert;
@@ -29,6 +33,9 @@ import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
+import com.google.common.base.Joiner;
+import com.google.common.collect.Sets;
+
 /**
  * Test provides some very generic helpers which might be used across the tests
  */
@@ -48,7 +55,28 @@ public abstract class GenericTestUtils {
   public static void assertExists(File f) {
     Assert.assertTrue("File " + f + " should exist", f.exists());
   }
-
+    
+  /**
+   * List all of the files in 'dir' that match the regex 'pattern'.
+   * Then check that this list is identical to 'expectedMatches'.
+   * @throws IOException if the dir is inaccessible
+   */
+  public static void assertGlobEquals(File dir, String pattern,
+      String ... expectedMatches) throws IOException {
+    
+    Set<String> found = Sets.newTreeSet();
+    for (File f : FileUtil.listFiles(dir)) {
+      if (f.getName().matches(pattern)) {
+        found.add(f.getName());
+      }
+    }
+    Set<String> expectedSet = Sets.newTreeSet(
+        Arrays.asList(expectedMatches));
+    Assert.assertEquals("Bad files matching " + pattern + " in " + dir,
+        Joiner.on(",").join(found),
+        Joiner.on(",").join(expectedSet));
+  }
+  
   public static void assertExceptionContains(String string, IOException ioe) {
     String msg = ioe.getMessage();
     Assert.assertTrue(