HDFS-1702. Federation: fix TestBackupNode and TestRefreshNamendoes failures. Contributed by Suresh Srinivas.

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

CHANGES.txt

@@ -154,7 +154,11 @@ Trunk (unreleased changes)
     HDFS-1682. Change Balancer CLI for multiple namenodes and balancing
     policy.  (szetszwo)
 
-    HDFS-1697. HDFS federation: fix testBlockRecovery (boryas)
+    HDFS-1697. Federation: fix TestBlockRecovery (boryas)
+
+    HDFS-1702. Federation: fix TestBackupNode and TestRefreshNamendoes
+    failures. (suresh)
+
   IMPROVEMENTS
 
     HDFS-1510. Added test-patch.properties required by test-patch.sh (nigel)

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

@@ -68,7 +68,11 @@ public class BackupNode extends NameNode {
   String nnHttpAddress;
   /** Checkpoint manager */
   Checkpointer checkpointManager;
-
+  /** ClusterID to which BackupNode belongs to */
+  String clusterId;
+  /** Block pool Id of the peer namenode of this BackupNode */
+  String blockPoolId;
+  
   BackupNode(Configuration conf, NamenodeRole role) throws IOException {
     super(conf, role);
   }
@@ -141,6 +145,10 @@ public class BackupNode extends NameNode {
     // therefore lease hard limit should never expire.
     namesystem.leaseManager.setLeasePeriod(
         FSConstants.LEASE_SOFTLIMIT_PERIOD, Long.MAX_VALUE);
+    
+    clusterId = nsInfo.getClusterID();
+    blockPoolId = nsInfo.getBlockPoolID();
+
     // register with the active name-node 
     registerWith(nsInfo);
     // Checkpoint daemon should start after the rpc server started
@@ -374,4 +382,12 @@ public class BackupNode extends NameNode {
       + FSConstants.LAYOUT_VERSION + " actual "+ nsInfo.getLayoutVersion();
     return nsInfo;
   }
+  
+  String getBlockPoolId() {
+    return blockPoolId;
+  }
+  
+  String getClusterId() {
+    return clusterId;
+  }
 }

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

@@ -19,7 +19,6 @@ package org.apache.hadoop.hdfs.server.namenode;
 
 import java.io.IOException;
 import java.io.File;
-import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.Collection;
 
@@ -251,6 +250,8 @@ class Checkpointer extends Daemon {
     }
 
     BackupStorage bnImage = getFSImage();
+    bnImage.blockpoolID = backupNode.getBlockPoolId();
+    bnImage.clusterID = backupNode.getClusterId();
     bnImage.loadCheckpoint(sig);
     sig.validateStorageInfo(bnImage);
     bnImage.saveCheckpoint();

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

@@ -1588,7 +1588,7 @@ public class MiniDFSCluster {
     NameNodeInfo[] newlist = new NameNodeInfo[numNameNodes];
     System.arraycopy(nameNodes, 0, newlist, 0, nameNodes.length);
     nameNodes = newlist;
-    String nameserviceId = NAMESERVICE_ID_PREFIX + nnIndex;
+    String nameserviceId = NAMESERVICE_ID_PREFIX + (nnIndex + 1);
     
     String nameserviceIds = conf.get(DFSConfigKeys.DFS_FEDERATION_NAMESERVICES);
     nameserviceIds += "," + nameserviceId;
@@ -1601,9 +1601,7 @@ public class MiniDFSCluster {
     // Refresh datanodes with the newly started namenode
     for (DataNodeProperties dn : dataNodes) {
       DataNode datanode = dn.datanode;
-      Configuration dnConf = datanode.getConf();
-      dnConf.set(DFSConfigKeys.DFS_FEDERATION_NAMESERVICES, nameserviceIds);
-      datanode.refreshNamenodes(dnConf);
+      datanode.refreshNamenodes(conf);
     }
 
     // Wait for new namenode to get registrations from all the datanodes

src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestRefreshNamenodes.java

@@ -24,6 +24,7 @@ import java.io.IOException;
 import java.net.InetSocketAddress;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.server.datanode.DataNode.BPOfferService;
 import org.junit.Test;
@@ -43,6 +44,7 @@ public class TestRefreshNamenodes {
     Configuration conf = new Configuration();
     MiniDFSCluster cluster = null;
     try {
+      conf.set(DFSConfigKeys.DFS_FEDERATION_NAMESERVICES, "namesServerId1");
       cluster = new MiniDFSCluster.Builder(conf).numNameNodes(1)
           .nameNodePort(nnPort1).build();