HDFS-2285. BackupNode should reject requests to modify namespace. Contributed by Konstantin Shvachko.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.22@1167608 13f79535-47bb-0310-9956-ffa450edef68

hdfs/CHANGES.txt

@@ -621,6 +621,8 @@ Release 0.22.0 - Unreleased
 
     HDFS-2280. BackupNode fails with MD5 checksum Exception. (shv)
 
+    HDFS-2285. BackupNode should reject requests to modify namespace. (shv)
+
 Release 0.21.1 - Unreleased
 
   IMPROVEMENTS

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

@@ -169,6 +169,11 @@ public class BackupNode extends NameNode {
     super.stop();
   }
 
+  @Override // NameNode
+  public boolean setSafeMode(SafeModeAction action) throws IOException {
+    throw new UnsupportedActionException("setSafeMode");
+  }
+
   /////////////////////////////////////////////////////
   // NamenodeProtocol implementation for backup node.
   /////////////////////////////////////////////////////

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

@@ -155,13 +155,13 @@ public class BackupStorage extends FSImage {
         throw new IOException("Could not locate checkpoint directories");
       StorageDirectory sdName = itImage.next();
       StorageDirectory sdEdits = itEdits.next();
-      getFSDirectoryRootLock().writeLock();
+      fsDir.writeLock();
       try { // load image under rootDir lock
         // make sure image checksum is verified against the expected value
         imageDigest = sig.imageDigest; 
         loadFSImage(FSImage.getImageFile(sdName, NameNodeFile.IMAGE));
       } finally {
-        getFSDirectoryRootLock().writeUnlock();
+        fsDir.writeUnlock();
       }
       loadFSEdits(sdEdits);
     }
@@ -170,6 +170,8 @@ public class BackupStorage extends FSImage {
     setStorageInfo(sig);
     imageDigest = sig.imageDigest;
     checkpointTime = sig.checkpointTime;
+    fsDir.setReady();
+    getFSNamesystem().setBlockTotal();
   }
 
   /**
@@ -180,10 +182,6 @@ public class BackupStorage extends FSImage {
     saveNamespace(false);
   }
 
-  private FSDirectory getFSDirectoryRootLock() {
-    return getFSNamesystem().dir;
-  }
-
   static File getJSpoolDir(StorageDirectory sd) {
     return new File(sd.getRoot(), STORAGE_JSPOOL_DIR);
   }

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

@@ -51,7 +51,6 @@ import org.apache.hadoop.hdfs.server.common.HdfsConstants.StartupOption;
 import org.apache.hadoop.hdfs.util.ByteArray;
 import static org.apache.hadoop.hdfs.server.common.Util.now;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.security.UserGroupInformation;
 
 /*************************************************
  * FSDirectory stores the filesystem directory state.
@@ -156,6 +155,11 @@ class FSDirectory implements Closeable {
       fsImage.close();
       throw e;
     }
+    setReady();
+  }
+
+  void setReady() {
+    if(ready) return;
     writeLock();
     try {
       this.ready = true;

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

@@ -311,10 +311,11 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
       LOG.info("Finished loading FSImage in " + timeTakenToLoadFSImage + " msecs");
       NameNode.getNameNodeMetrics().fsImageLoadTime.set(
                                 (int) timeTakenToLoadFSImage);
+      this.safeMode = new SafeModeInfo(conf);
     } else {
       this.dir = new FSDirectory(fsImage, this, conf);
+      this.safeMode = new SafeModeInfo(); // manual safe mode
     }
-    this.safeMode = new SafeModeInfo(conf);
     this.hostsReader = new HostsFileReader(conf.get("dfs.hosts",""),
                         conf.get("dfs.hosts.exclude",""));
     if (isBlockTokenEnabled) {

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

@@ -19,6 +19,7 @@ package org.apache.hadoop.hdfs.server.namenode;
 
 import java.io.File;
 import java.io.IOException;
+import java.net.InetSocketAddress;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -109,6 +110,7 @@ public class TestBackupNode extends TestCase {
   void testCheckpoint(StartupOption op) throws IOException {
     Path file1 = new Path("checkpoint.dat");
     Path file2 = new Path("checkpoint2.dat");
+    Path file3 = new Path("backup.dat");
 
     Configuration conf = new HdfsConfiguration();
     short replication = (short)conf.getInt("dfs.replication", 3);
@@ -178,6 +180,27 @@ public class TestBackupNode extends TestCase {
       //
       backup = startBackupNode(conf, op, 1);
       waitCheckpointDone(backup);
+
+      // Try BackupNode operations
+      InetSocketAddress add = backup.getNameNodeAddress();
+      // Write to BN
+      FileSystem bnFS = FileSystem.get(
+          new Path("hdfs://" + NameNode.getHostPortString(add)).toUri(), conf);
+      boolean canWrite = true;
+      try {
+        writeFile(bnFS, file3, replication);
+      } catch(IOException eio) {
+        LOG.info("Write to BN failed as expected: ", eio);
+        canWrite = false;
+      }
+      assertFalse("Write to BackupNode must be prohibited.", canWrite);
+
+      writeFile(fileSys, file3, replication);
+      checkFile(fileSys, file3, replication);
+      // should also be on BN right away
+      assertTrue("file3 does not exist on BackupNode",
+          op != StartupOption.BACKUP || bnFS.exists(file3));
+
     } catch(IOException e) {
       LOG.error("Error in TestBackupNode:", e);
       assertTrue(e.getLocalizedMessage(), false);