HDFS-1718. Federation: MiniDFSCluster#waitActive() bug causes some tests to fail. Contributed by Suresh Srinivas.

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

CHANGES.txt

@@ -197,6 +197,9 @@ Trunk (unreleased changes)
     HDFS-1722. Federation: Add flag to MiniDFSCluser to differentiate between
     federation and non-federation modes. (boryas via suresh)
 
+    HDFS-1718. Federation: MiniDFSCluster#waitActive() bug causes some tests
+    to fail. (suresh)
+
   IMPROVEMENTS
 
     HDFS-1510. Added test-patch.properties required by test-patch.sh (nigel)

src/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java

@@ -244,6 +244,10 @@ public class DataNode extends Configured
       return nameNodeThreads.values().toArray(bposArray);
     }
     
+    synchronized BPOfferService get(InetSocketAddress addr) {
+      return nameNodeThreads.get(addr);
+    }
+    
     synchronized BPOfferService get(String bpid) {
       return bpMapping.get(bpid);
     }
@@ -668,6 +672,10 @@ public class DataNode extends Configured
       return initialized;
     }
     
+    public boolean isAlive() {
+      return shouldServiceRun && bpThread.isAlive();
+    }
+    
     public String getBlockPoolId() {
       return blockPoolId;
     }
@@ -1968,7 +1976,7 @@ public class DataNode extends Configured
    */
   public boolean isDatanodeUp() {
     for (BPOfferService bp : blockPoolManager.getAllNamenodeThreads()) {
-      if (bp.bpThread.isAlive()) {
+      if (bp.isAlive()) {
         return true;
       }
     }
@@ -2567,16 +2575,22 @@ public class DataNode extends Configured
     refreshNamenodes(conf);
   }
 
+  /**
+   * @param addr rpc address of the namenode
+   * @return true - if BPOfferService corresponding to the namenode is alive
+   */
+  public boolean isBPServiceAlive(InetSocketAddress addr) {
+    BPOfferService bp = blockPoolManager.get(addr);
+    return bp != null ? bp.isAlive() : false;
+  }
+  
   /**
    * @param bpid block pool Id
    * @return true - if BPOfferService thread is alive
    */
   public boolean isBPServiceAlive(String bpid) {
     BPOfferService bp = blockPoolManager.get(bpid);
-    if (bp != null) {
-      return bp.shouldServiceRun && bp.bpThread.isAlive();
-    }
-    return false;
+    return bp != null ? bp.isAlive() : false;
   }
 
   /**
@@ -2587,7 +2601,7 @@ public class DataNode extends Configured
    */
   public boolean isDatanodeFullyStarted() {
     for (BPOfferService bp : blockPoolManager.getAllNamenodeThreads()) {
-      if (!bp.initialized() || !bp.shouldServiceRun || !bp.bpThread.isAlive()) {
+      if (!bp.initialized() || !bp.isAlive()) {
         return false;
       }
     }

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

@@ -1360,11 +1360,20 @@ public class NameNode implements NamenodeProtocols, FSConstants {
 
   /**
    * Returns the address on which the NameNodes is listening to.
-   * @return the address on which the NameNodes is listening to.
+   * @return namenode rpc address
    */
   public InetSocketAddress getNameNodeAddress() {
     return rpcAddress;
   }
+  
+  /**
+   * Returns namenode service rpc address, if set. Otherwise returns
+   * namenode rpc address.
+   * @return namenode service rpc address used by datanodes
+   */
+  public InetSocketAddress getServiceRpcAddress() {
+    return serviceRPCAddress != null ? serviceRPCAddress : rpcAddress;
+  }
 
   /**
    * Returns the address of the NameNodes http server, 

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

@@ -1362,14 +1362,13 @@ public class MiniDFSCluster {
     if (nameNodes.length == 0 || nameNodes[nnIndex] == null) {
       return;
     }
-    String bpid = nameNodes[nnIndex].nameNode.getFSImage().getBlockPoolID();
-    InetSocketAddress addr = new InetSocketAddress("localhost",
-        getNameNodePort(nnIndex));
+    InetSocketAddress addr = nameNodes[nnIndex].nameNode.getServiceRpcAddress();
     DFSClient client = new DFSClient(addr, conf);
 
     // ensure all datanodes have registered and sent heartbeat to the namenode
-    while (shouldWait(client.datanodeReport(DatanodeReportType.LIVE), bpid)) {
+    while (shouldWait(client.datanodeReport(DatanodeReportType.LIVE), addr)) {
       try {
+        LOG.info("Waiting for cluster to become active");
         Thread.sleep(100);
       } catch (InterruptedException e) {
       }
@@ -1387,29 +1386,26 @@ public class MiniDFSCluster {
     }
   }
   
-  private synchronized boolean shouldWait(DatanodeInfo[] dnInfo, String bpid) {
+  private synchronized boolean shouldWait(DatanodeInfo[] dnInfo,
+      InetSocketAddress addr) {
+    // If a datanode failed to start, then do not wait
     for (DataNodeProperties dn : dataNodes) {
-      // If any one of the datanode is down, then do not continue to wait
-      // since the subsequent checks to stop waiting in this method are never
-      // going to be be met resulting in waiting forever.
-      if (!dn.datanode.isDatanodeUp()) {
-        return false;
-      }
-
-      // if registration of this datanode with the namenode failed -  don't wait
-      if(!dn.datanode.isBPServiceAlive(bpid)) {
+      // the datanode thread communicating with the namenode should be alive
+      if (!dn.datanode.isBPServiceAlive(addr)) {
+        LOG.warn("One or more BPOfferService failed to start in datanode " + dn
+            + " for namenode" + addr);
         return false;
       }
     }
     
+    // Wait for expected number of datanodes to start
     if (dnInfo.length != numDataNodes) {
       return true;
     }
     
-    // check if all data nodes are fully started
+    // if one of the data nodes is not fully started, continue to wait
     for (DataNodeProperties dn : dataNodes) {
       if (!dn.datanode.isDatanodeFullyStarted()) {
-        // if one of the data nodes is not fully started, continue to wait
         return true;
       }
     }

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

@@ -830,10 +830,10 @@ public class TestFileCreation extends junit.framework.TestCase {
       dfs = (DistributedFileSystem)cluster.getFileSystem();
 
       // create a new file.
-      final String f = DIR + "dhrubashutdown";
+      final String f = DIR + "testFsCloseAfterClusterShutdown";
       final Path fpath = new Path(f);
       FSDataOutputStream out = TestFileCreation.createFile(dfs, fpath, DATANODE_NUM);
-      out.write("something_dhruba".getBytes());
+      out.write("something_test".getBytes());
       out.hflush();    // ensure that block is allocated
 
       // shutdown last datanode in pipeline.

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

@@ -161,9 +161,7 @@ public class TestDataNodeMultipleRegistrations {
       // try block report
       BPOfferService bpos1 = dn.getAllBpOs()[0];
       bpos1.lastBlockReport = 0;
-      DatanodeCommand cmd = bpos1.blockReport();
-
-      assertNotNull("cmd is null", cmd);
+      bpos1.blockReport();
 
       assertEquals("wrong nn address", bpos1.nnAddr,
           nn1.getNameNodeAddress());

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

@@ -45,8 +45,8 @@ public class TestRefreshNamenodes {
     MiniDFSCluster cluster = null;
     try {
       conf.set(DFSConfigKeys.DFS_FEDERATION_NAMESERVICES, "namesServerId1");
-      cluster = new MiniDFSCluster.Builder(conf).numNameNodes(1)
-          .nameNodePort(nnPort1).build();
+      cluster = new MiniDFSCluster.Builder(conf).federation(true).
+          numNameNodes(1).nameNodePort(nnPort1).build();
 
       DataNode dn = cluster.getDataNodes().get(0);
       assertEquals(1, dn.getAllBpOs().length);
@@ -75,7 +75,9 @@ public class TestRefreshNamenodes {
         assertTrue("NameNode address " + addr + " is not found.", found);
       }
     } finally {
-      cluster.shutdown();
+      if (cluster != null) {
+        cluster.shutdown();
+      }
     }
   }
 }