HDFS-2026. SecondaryNameNode should properly handle the case where the NameNode is reformatted. Contributed by Todd Lipcon.

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

hdfs/CHANGES.HDFS-1073.txt

@@ -53,3 +53,5 @@ HDFS-2027. Image inspector should return finalized logs before unfinalized
 HDFS-2074. Determine edit log validity by truly reading and validating
            transactions. (todd)
 HDFS-2085. Finalize in-progress edit logs at startup. (todd)
+HDFS-2026. SecondaryNameNode should properly handle the case where the
+           NameNode is reformatted. (todd)

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

@@ -25,6 +25,8 @@ import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableUtils;
 
+import com.google.common.base.Joiner;
+
 /**
  * Common class for storage information.
  * 
@@ -105,4 +107,9 @@ public class StorageInfo implements Writable {
     .append(";nsid=").append(namespaceID).append(";c=").append(cTime);
     return sb.toString();
   }
+  
+  public String toColonSeparatedString() {
+    return Joiner.on(":").join(
+        layoutVersion, namespaceID, cTime, clusterID);
+  }
 }

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

@@ -197,7 +197,8 @@ class Checkpointer extends Daemon {
    * Copy the new image into remote name-node.
    */
   private void uploadCheckpoint(CheckpointSignature sig) throws IOException {
-    // Use the exact http addr as specified in config to deal with ip aliasing
+    // TODO: checkpoint node disabled in 1073 branch
+/*    // Use the exact http addr as specified in config to deal with ip aliasing
     InetSocketAddress httpSocAddr = backupNode.getHttpAddress();
     int httpPort = httpSocAddr.getPort();
     String fileid = "putimage=1&port=" + httpPort +
@@ -206,6 +207,7 @@ class Checkpointer extends Daemon {
     LOG.info("Posted URL " + backupNode.nnHttpAddress + fileid);
     TransferFsImage.getFileClient(backupNode.nnHttpAddress, 
         fileid, null, false);
+        */
   }
 
   /**

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

@@ -875,22 +875,6 @@ public class FSImage implements Closeable {
     return new CheckpointSignature(this);
   }
 
-  /**
-   * This is called just before a new checkpoint is uploaded to the
-   * namenode.
-   */
-  void validateCheckpointUpload(CheckpointSignature sig) throws IOException { 
-    // verify token
-    long curTxId = getEditLog().getLastWrittenTxId();
-    if (sig.curSegmentTxId > curTxId) {
-      throw new IOException("Namenode has already reached txid " +
-          curTxId + " but new checkpoint was created using editlog " +
-          "starting at txid " + sig.curSegmentTxId + ". Checkpoint Aborted.");
-    }
-
-    sig.validateStorageInfo(this);
-  }
-
   /**
    * Start checkpoint.
    * <p>

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

@@ -20,6 +20,7 @@ package org.apache.hadoop.hdfs.server.namenode;
 import java.security.PrivilegedExceptionAction;
 import java.util.*;
 import java.io.*;
+import java.net.InetSocketAddress;
 
 import javax.servlet.ServletContext;
 import javax.servlet.ServletException;
@@ -35,6 +36,7 @@ import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.server.common.JspHelper;
+import org.apache.hadoop.hdfs.server.common.StorageInfo;
 import org.apache.hadoop.hdfs.server.protocol.RemoteEditLog;
 import org.apache.hadoop.hdfs.util.DataTransferThrottler;
 import org.apache.hadoop.hdfs.util.MD5FileUtils;
@@ -58,6 +60,7 @@ public class GetImageServlet extends HttpServlet {
   private static final String TXID_PARAM = "txid";
   private static final String START_TXID_PARAM = "startTxId";
   private static final String END_TXID_PARAM = "endTxId";
+  private static final String STORAGEINFO_PARAM = "storageInfo";
   
   private static Set<Long> currentlyDownloadingCheckpoints =
     Collections.<Long>synchronizedSet(new HashSet<Long>());
@@ -81,12 +84,27 @@ public class GetImageServlet extends HttpServlet {
         return;
       }
       
+      String myStorageInfoString = nnImage.getStorage().toColonSeparatedString();
+      String theirStorageInfoString = parsedParams.getStorageInfoString();
+      if (theirStorageInfoString != null &&
+          !myStorageInfoString.equals(theirStorageInfoString)) {
+        response.sendError(HttpServletResponse.SC_FORBIDDEN,
+            "This namenode has storage info " + myStorageInfoString + 
+            " but the secondary expected " + theirStorageInfoString);
+        LOG.warn("Received an invalid request file transfer request " +
+            "from a secondary with storage info " + theirStorageInfoString);
+        return;
+      }
+      
       UserGroupInformation.getCurrentUser().doAs(new PrivilegedExceptionAction<Void>() {
         @Override
         public Void run() throws Exception {
           if (parsedParams.isGetImage()) {
             long txid = parsedParams.getTxId();
             File imageFile = nnImage.getStorage().getFsImageName(txid);
+            if (imageFile == null) {
+              throw new IOException("Could not find image with txid " + txid);
+            }
             setVerificationHeaders(response, imageFile);
             // send fsImage
             TransferFsImage.getFileServer(response.getOutputStream(), imageFile,
@@ -119,7 +137,6 @@ public class GetImageServlet extends HttpServlet {
               }
               
               // issue a HTTP get request to download the new fsimage 
-              nnImage.validateCheckpointUpload(parsedParams.getToken());
               MD5Hash downloadImageDigest = reloginIfNecessary().doAs(
                   new PrivilegedExceptionAction<MD5Hash>() {
                   @Override
@@ -225,23 +242,42 @@ public class GetImageServlet extends HttpServlet {
     }
   }
 
-  static String getParamStringForImage(long txid) {
-    return "getimage=1&" + TXID_PARAM + "=" + txid;
+  static String getParamStringForImage(long txid,
+      StorageInfo remoteStorageInfo) {
+    return "getimage=1&" + TXID_PARAM + "=" + txid
+      + "&" + STORAGEINFO_PARAM + "=" +
+      remoteStorageInfo.toColonSeparatedString();
+    
   }
 
-  static String getParamStringForLog(RemoteEditLog log) {
+  static String getParamStringForLog(RemoteEditLog log,
+      StorageInfo remoteStorageInfo) {
     return "getedit=1&" + START_TXID_PARAM + "=" + log.getStartTxId()
-        + "&" + END_TXID_PARAM + "=" + log.getEndTxId();
+        + "&" + END_TXID_PARAM + "=" + log.getEndTxId()
+        + "&" + STORAGEINFO_PARAM + "=" +
+          remoteStorageInfo.toColonSeparatedString();
   }
   
+  static String getParamStringToPutImage(long txid,
+      InetSocketAddress imageListenAddress, NNStorage storage) {
+    
+    return "putimage=1" +
+      "&" + TXID_PARAM + "=" + txid +
+      "&port=" + imageListenAddress.getPort() +
+      "&machine=" + imageListenAddress.getHostName()
+      + "&" + STORAGEINFO_PARAM + "=" +
+      storage.toColonSeparatedString();
+  }
+
+  
   static class GetImageParams {
     private boolean isGetImage;
     private boolean isGetEdit;
     private boolean isPutImage;
     private int remoteport;
     private String machineName;
-    private CheckpointSignature token;
     private long startTxId, endTxId, txId;
+    private String storageInfoString;
 
     /**
      * @param request the object from which this servlet reads the url contents
@@ -256,7 +292,6 @@ public class GetImageServlet extends HttpServlet {
       isGetImage = isGetEdit = isPutImage = false;
       remoteport = 0;
       machineName = null;
-      token = null;
 
       for (Iterator<String> it = pmap.keySet().iterator(); it.hasNext();) {
         String key = it.next();
@@ -274,8 +309,8 @@ public class GetImageServlet extends HttpServlet {
           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 CheckpointSignature(pmap.get("token")[0]);
+        } else if (key.equals(STORAGEINFO_PARAM)) {
+          storageInfoString = pmap.get(key)[0];
         }
       }
 
@@ -285,6 +320,10 @@ public class GetImageServlet extends HttpServlet {
       }
     }
 
+    public String getStorageInfoString() {
+      return storageInfoString;
+    }
+
     public long getTxId() {
       Preconditions.checkState(isGetImage || isPutImage);
       return txId;
@@ -311,10 +350,6 @@ public class GetImageServlet extends HttpServlet {
     boolean isPutImage() {
       return isPutImage;
     }
-
-    CheckpointSignature getToken() {
-      return token;
-    }
     
     String getInfoServer() throws IOException{
       if (machineName == null || remoteport == 0) {

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

@@ -776,7 +776,9 @@ public class NNStorage extends Storage implements Closeable {
    * @param uVersion the new version.
    */
   private void setDistributedUpgradeState(boolean uState, int uVersion) {
-    upgradeManager.setUpgradeState(uState, uVersion);
+    if (upgradeManager != null) {
+      upgradeManager.setUpgradeState(uState, uVersion);
+    }
   }
 
   /**

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

@@ -28,6 +28,8 @@ import java.util.Date;
 import java.util.Iterator;
 import java.util.List;
 
+import javax.jws.soap.SOAPBinding.Use;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -232,8 +234,8 @@ public class SecondaryNameNode implements Runnable {
             System.setProperty("https.cipherSuites", 
                 Krb5AndCertsSslSocketConnector.KRB5_CIPHER_SUITES.get(0));
             InetSocketAddress secInfoSocAddr = 
-              NetUtils.createSocketAddr(infoBindAddress + ":"+ conf.get(
-                "dfs.secondary.https.port", infoBindAddress + ":" + 0));
+              NetUtils.createSocketAddr(infoBindAddress + ":"+ conf.getInt(
+                "dfs.secondary.https.port", 443));
             imagePort = secInfoSocAddr.getPort();
             infoServer.addSslListener(secInfoSocAddr, conf, false, true);
           }
@@ -255,8 +257,10 @@ public class SecondaryNameNode implements Runnable {
 
     // The web-server port can be ephemeral... ensure we have the correct info
     infoPort = infoServer.getPort();
-    if(!UserGroupInformation.isSecurityEnabled())
+    if (!UserGroupInformation.isSecurityEnabled()) {
       imagePort = infoPort;
+    }
+    
     conf.set(DFSConfigKeys.DFS_NAMENODE_SECONDARY_HTTP_ADDRESS_KEY, infoBindAddress + ":" +infoPort); 
     LOG.info("Secondary Web-server up at: " + infoBindAddress + ":" +infoPort);
     LOG.info("Secondary image servlet up at: " + infoBindAddress + ":" + imagePort);
@@ -406,17 +410,6 @@ public class SecondaryNameNode implements Runnable {
     return nameNodeAddr;
   }
 
-  /**
-   * Copy the new fsimage into the NameNode
-   */
-  private void putFSImage(CheckpointSignature sig, long txid) throws IOException {
-    String fileid = "putimage=1&txid=" + txid + "&port=" + imagePort +
-      "&machine=" + infoBindAddress + 
-      "&token=" + sig.toString();
-    LOG.info("Posted URL " + fsName + fileid);
-    TransferFsImage.getFileClient(fsName, fileid, null, false);
-  }
-
   /**
    * Returns the Jetty server that the Namenode is listening on.
    */
@@ -441,6 +434,14 @@ public class SecondaryNameNode implements Runnable {
       return configuredAddress;
     }
   }
+  
+  /**
+   * Return the host:port of where this SecondaryNameNode is listening
+   * for image transfers
+   */
+  private InetSocketAddress getImageListenAddress() {
+    return new InetSocketAddress(infoBindAddress, imagePort);
+  }
 
   /**
    * Create a new checkpoint
@@ -452,6 +453,17 @@ public class SecondaryNameNode implements Runnable {
     // Tell the namenode to start logging transactions in a new edit file
     // Returns a token that would be used to upload the merged image.
     CheckpointSignature sig = namenode.rollEditLog();
+    
+    // Make sure we're talking to the same NN!
+    if (checkpointImage.getNamespaceID() != 0) {
+      // If the image actually has some data, make sure we're talking
+      // to the same NN as we did before.
+      sig.validateStorageInfo(checkpointImage);
+    } else {
+      // if we're a fresh 2NN, just take the storage info from the server
+      // we first talk to.
+      checkpointImage.getStorage().setStorageInfo(sig);
+    }
 
     // error simulation code for junit test
     if (ErrorSimulator.getErrorSimulation(0)) {
@@ -482,7 +494,8 @@ public class SecondaryNameNode implements Runnable {
     // to make this new uploaded image as the most current image.
     //
     long txid = checkpointImage.getStorage().getMostRecentCheckpointTxId();
-    putFSImage(sig, txid);
+    TransferFsImage.uploadImageFromStorage(fsName, getImageListenAddress(),
+        checkpointImage.getStorage(), txid);
 
     // error simulation code for junit test
     if (ErrorSimulator.getErrorSimulation(1)) {
@@ -671,6 +684,11 @@ public class SecondaryNameNode implements Runnable {
           case NOT_FORMATTED:
             break;  // it's ok since initially there is no current and VERSION
           case NORMAL:
+            // Read the VERSION file. This verifies that:
+            // (a) the VERSION file for each of the directories is the same,
+            // and (b) when we connect to a NN, we can verify that the remote
+            // node matches the same namespace that we ran on previously.
+            sd.read();
             break;
           default:  // recovery is possible
             sd.doRecover(curState);

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

@@ -34,6 +34,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.hdfs.protocol.FSConstants;
+import org.apache.hadoop.hdfs.server.common.StorageInfo;
 import org.apache.hadoop.hdfs.server.namenode.NNStorage.NameNodeDirType;
 import org.apache.hadoop.hdfs.server.protocol.RemoteEditLog;
 import org.apache.hadoop.hdfs.util.DataTransferThrottler;
@@ -58,7 +59,8 @@ class TransferFsImage implements FSConstants {
   static MD5Hash downloadImageToStorage(
       String fsName, long imageTxId, NNStorage dstStorage, boolean needDigest)
       throws IOException {
-    String fileid = GetImageServlet.getParamStringForImage(imageTxId);
+    String fileid = GetImageServlet.getParamStringForImage(
+        imageTxId, dstStorage);
     String fileName = NNStorage.getCheckpointImageFileName(imageTxId);
     
     List<File> dstFiles = dstStorage.getFiles(
@@ -74,18 +76,42 @@ class TransferFsImage implements FSConstants {
   }
   
   static void downloadEditsToStorage(String fsName, RemoteEditLog log,
-      NNStorage storage) throws IOException {
-    String fileid = GetImageServlet.getParamStringForLog(log);
+      NNStorage dstStorage) throws IOException {
+    String fileid = GetImageServlet.getParamStringForLog(
+        log, dstStorage);
     String fileName = NNStorage.getFinalizedEditsFileName(
         log.getStartTxId(), log.getEndTxId());
 
-    List<File> dstFiles = storage.getFiles(NameNodeDirType.EDITS, fileName);
+    List<File> dstFiles = dstStorage.getFiles(NameNodeDirType.EDITS, fileName);
     assert !dstFiles.isEmpty() : "No checkpoint targets.";
 
     getFileClient(fsName, fileid, dstFiles, false);
     LOG.info("Downloaded file " + dstFiles.get(0).getName() + " size " +
         dstFiles.get(0).length() + " bytes.");
   }
+ 
+  /**
+   * Requests that the NameNode download an image from this node.
+   *
+   * @param fsName the http address for the remote NN
+   * @param imageListenAddress the host/port where the local node is running an
+   *                           HTTPServer hosting GetImageServlet
+   * @param storage the storage directory to transfer the image from
+   * @param txid the transaction ID of the image to be uploaded
+   */
+  static void uploadImageFromStorage(String fsName,
+      InetSocketAddress imageListenAddress,
+      NNStorage storage, long txid) throws IOException {
+    
+    String fileid = GetImageServlet.getParamStringToPutImage(
+        txid, imageListenAddress, storage);
+    // this doesn't directly upload an image, but rather asks the NN
+    // to connect back to the 2NN to download the specified image.
+    TransferFsImage.getFileClient(fsName, fileid, null, false);
+    LOG.info("Uploaded image with txid " + txid + " to namenode at " +
+    		fsName);
+  }
+
   
   /**
    * A server-side method to respond to a getfile http request
@@ -249,4 +275,5 @@ class TransferFsImage implements FSConstants {
     String header = connection.getHeaderField(MD5_HEADER);
     return (header != null) ? new MD5Hash(header) : null;
   }
+
 }

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

@@ -95,6 +95,7 @@ public class MiniDFSCluster {
    */
   public static class Builder {
     private int nameNodePort = 0;
+    private int nameNodeHttpPort = 0;
     private final Configuration conf;
     private int numNameNodes = 1;
     private int numDataNodes = 1;
@@ -130,6 +131,14 @@ public class MiniDFSCluster {
       this.nameNodePort = val;
       return this;
     }
+    
+    /**
+     * Default: 0
+     */
+    public Builder nameNodeHttpPort(int val) {
+      this.nameNodeHttpPort = val;
+      return this;
+    }
 
     /**
      * Default: 1
@@ -247,6 +256,7 @@ public class MiniDFSCluster {
       builder.federation = true;
       
     initMiniDFSCluster(builder.nameNodePort,
+                       builder.nameNodeHttpPort,
                        builder.conf,
                        builder.numDataNodes,
                        builder.format,
@@ -463,12 +473,13 @@ public class MiniDFSCluster {
                         String[] racks, String hosts[],
                         long[] simulatedCapacities) throws IOException {
     this.nameNodes = new NameNodeInfo[1]; // Single namenode in the cluster
-    initMiniDFSCluster(nameNodePort, conf, numDataNodes, format,
+    initMiniDFSCluster(nameNodePort, 0, conf, numDataNodes, format,
         manageNameDfsDirs, manageDataDfsDirs, operation, racks, hosts,
         simulatedCapacities, null, true, false, false);
   }
 
-  private void initMiniDFSCluster(int nameNodePort, Configuration conf,
+  private void initMiniDFSCluster(int nameNodePort, int nameNodeHttpPort,
+      Configuration conf,
       int numDataNodes, boolean format, boolean manageNameDfsDirs,
       boolean manageDataDfsDirs, StartupOption operation, String[] racks,
       String[] hosts, long[] simulatedCapacities, String clusterId,
@@ -516,7 +527,8 @@ public class MiniDFSCluster {
   
     if (!federation) {
       conf.set(DFSConfigKeys.FS_DEFAULT_NAME_KEY, "127.0.0.1:" + nameNodePort);
-      conf.set(DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY, "127.0.0.1:0");
+      conf.set(DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY, "127.0.0.1:"
+          + nameNodeHttpPort);
       NameNode nn = createNameNode(0, conf, numDataNodes, manageNameDfsDirs,
           format, operation, clusterId);
       nameNodes[0] = new NameNodeInfo(nn, conf);

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

@@ -31,6 +31,7 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.commons.logging.Log;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.DFSUtil;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileContext;
 import org.apache.hadoop.fs.FileSystem;
@@ -45,10 +46,14 @@ import org.apache.hadoop.hdfs.protocol.FSConstants.SafeModeAction;
 import org.apache.hadoop.hdfs.server.common.HdfsConstants.StartupOption;
 import org.apache.hadoop.hdfs.server.common.Storage;
 import org.apache.hadoop.hdfs.server.common.Storage.StorageDirectory;
+import org.apache.hadoop.hdfs.server.common.StorageInfo;
 import org.apache.hadoop.hdfs.server.namenode.NNStorage.NameNodeDirType;
 import org.apache.hadoop.hdfs.server.namenode.SecondaryNameNode.CheckpointStorage;
 import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocol;
+import org.apache.hadoop.hdfs.server.protocol.RemoteEditLog;
+import org.apache.hadoop.hdfs.server.protocol.RemoteEditLogManifest;
 import org.apache.hadoop.hdfs.tools.DFSAdmin;
+import org.apache.hadoop.ipc.WritableRpcEngine;
 import org.apache.hadoop.test.GenericTestUtils;
 import org.apache.hadoop.test.GenericTestUtils.DelayAnswer;
 import org.apache.hadoop.util.StringUtils;
@@ -1254,6 +1259,126 @@ public class TestCheckpoint extends TestCase {
     // (i.e that both checkpoints went through)
     assertNNHasCheckpoints(cluster, ImmutableList.of(6,8));
   }
+  
+  /**
+   * Test case where the name node is reformatted while the secondary namenode
+   * is running. The secondary should shut itself down if if talks to a NN
+   * with the wrong namespace.
+   */
+  @SuppressWarnings("deprecation")
+  public void testReformatNNBetweenCheckpoints() throws IOException {
+    MiniDFSCluster cluster = null;
+    SecondaryNameNode secondary = null;
+    
+    Configuration conf = new HdfsConfiguration();
+    conf.setInt(CommonConfigurationKeysPublic.IPC_CLIENT_CONNECTION_MAXIDLETIME_KEY,
+        1);
+
+    try {
+      cluster = new MiniDFSCluster.Builder(conf).numDataNodes(0)
+          .format(true).build();
+      int origPort = cluster.getNameNodePort();
+      int origHttpPort = cluster.getNameNode().getHttpAddress().getPort();
+      secondary = startSecondaryNameNode(conf);
+
+      // secondary checkpoints once
+      secondary.doCheckpoint();
+
+      // we reformat primary NN
+      cluster.shutdown();
+      cluster = null;
+
+      // Brief sleep to make sure that the 2NN's IPC connection to the NN
+      // is dropped.
+      try {
+        Thread.sleep(100);
+      } catch (InterruptedException ie) {
+      }
+      
+      // Start a new NN with the same host/port.
+      cluster = new MiniDFSCluster.Builder(conf)
+        .numDataNodes(0)
+        .nameNodePort(origPort)
+        .nameNodeHttpPort(origHttpPort)
+        .format(true).build();
+
+      try {
+        secondary.doCheckpoint();
+        fail("Should have failed checkpoint against a different namespace");
+      } catch (IOException ioe) {
+        LOG.info("Got expected failure", ioe);
+        assertTrue(ioe.toString().contains("Inconsistent checkpoint"));
+      }
+    } finally {
+      if (secondary != null) {
+        secondary.shutdown();
+      }
+      if (cluster != null) {
+        cluster.shutdown();
+      }
+    }  
+  }
+  
+  /**
+   * Test that the primary NN will not serve any files to a 2NN who doesn't
+   * share its namespace ID, and also will not accept any files from one.
+   */
+  @SuppressWarnings("deprecation")
+  public void testNamespaceVerifiedOnFileTransfer() throws IOException {
+    MiniDFSCluster cluster = null;
+    
+    Configuration conf = new HdfsConfiguration();
+    try {
+      cluster = new MiniDFSCluster.Builder(conf).numDataNodes(0)
+          .format(true).build();
+      
+      NameNode nn = cluster.getNameNode();
+      String fsName = NameNode.getHostPortString(nn.getHttpAddress());
+
+
+      // Make a finalized log on the server side. 
+      nn.rollEditLog();
+      RemoteEditLogManifest manifest = nn.getEditLogManifest(0);
+      RemoteEditLog log = manifest.getLogs().get(0);
+      
+      NNStorage dstImage = Mockito.mock(NNStorage.class);
+      Mockito.doReturn(Lists.newArrayList(new File("/wont-be-written")))
+        .when(dstImage).getFiles(
+            Mockito.<NameNodeDirType>anyObject(), Mockito.anyString());
+      
+      Mockito.doReturn(new StorageInfo(1, 1, "X", 1).toColonSeparatedString())
+        .when(dstImage).toColonSeparatedString();
+
+      try {
+        TransferFsImage.downloadImageToStorage(fsName, 0, dstImage, false);
+        fail("Storage info was not verified");
+      } catch (IOException ioe) {
+        String msg = StringUtils.stringifyException(ioe);
+        assertTrue(msg, msg.contains("but the secondary expected"));
+      }
+
+      try {
+        TransferFsImage.downloadEditsToStorage(fsName, log, dstImage);
+        fail("Storage info was not verified");
+      } catch (IOException ioe) {
+        String msg = StringUtils.stringifyException(ioe);
+        assertTrue(msg, msg.contains("but the secondary expected"));
+      }
+
+      try {
+        InetSocketAddress fakeAddr = new InetSocketAddress(1);
+        TransferFsImage.uploadImageFromStorage(fsName, fakeAddr, dstImage, 0);
+        fail("Storage info was not verified");
+      } catch (IOException ioe) {
+        String msg = StringUtils.stringifyException(ioe);
+        assertTrue(msg, msg.contains("but the secondary expected"));
+      }
+    } finally {
+      if (cluster != null) {
+        cluster.shutdown();
+      }
+    }  
+  }
 
 
   @SuppressWarnings("deprecation")