hadoop-hdfs-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From t...@apache.org
Subject svn commit: r1099687 [10/15] - in /hadoop/hdfs/branches/HDFS-1073: ./ bin/ src/c++/libhdfs/ src/contrib/hdfsproxy/ src/contrib/hdfsproxy/src/java/org/apache/hadoop/hdfsproxy/ src/java/ src/java/org/apache/hadoop/hdfs/ src/java/org/apache/hadoop/hdfs/pr...
Date Thu, 05 May 2011 05:40:13 GMT
Modified: hadoop/hdfs/branches/HDFS-1073/src/test/aop/org/apache/hadoop/hdfs/server/datanode/DataTransferProtocolAspects.aj
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1073/src/test/aop/org/apache/hadoop/hdfs/server/datanode/DataTransferProtocolAspects.aj?rev=1099687&r1=1099686&r2=1099687&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1073/src/test/aop/org/apache/hadoop/hdfs/server/datanode/DataTransferProtocolAspects.aj (original)
+++ hadoop/hdfs/branches/HDFS-1073/src/test/aop/org/apache/hadoop/hdfs/server/datanode/DataTransferProtocolAspects.aj Thu May  5 05:40:07 2011
@@ -49,7 +49,7 @@ public aspect DataTransferProtocolAspect
 
   after(DataXceiver dataxceiver) returning(Op op): receiverOp(dataxceiver) {
     LOG.info("FI: receiverOp " + op + ", datanode="
-        + dataxceiver.getDataNode().getDatanodeRegistration().getName());    
+        + dataxceiver.getDataNode().getDatanodeId().getName());    
   }
 
   pointcut statusRead(DataXceiver dataxceiver):
@@ -59,10 +59,10 @@ public aspect DataTransferProtocolAspect
       ) throws IOException: statusRead(dataxceiver) {
     final DataNode d = dataxceiver.getDataNode();
     LOG.info("FI: statusRead " + status + ", datanode="
-        + d.getDatanodeRegistration().getName());    
+        + d.getDatanodeId().getName());    
     DataTransferTest dtTest = DataTransferTestUtil.getDataTransferTest();
     if (dtTest != null) 
-      dtTest.fiStatusRead.run(d.getDatanodeRegistration());
+      dtTest.fiStatusRead.run(d.getDatanodeId());
   }
 
   pointcut receiverOpWriteBlock(DataXceiver dataxceiver):
@@ -74,6 +74,6 @@ public aspect DataTransferProtocolAspect
     DataTransferTest dtTest = DataTransferTestUtil.getDataTransferTest();
     if (dtTest != null)
       dtTest.fiReceiverOpWriteBlock.run(
-          dataxceiver.getDataNode().getDatanodeRegistration());
+          dataxceiver.getDataNode().getDatanodeId());
   }
 }

Modified: hadoop/hdfs/branches/HDFS-1073/src/test/aop/org/apache/hadoop/hdfs/server/datanode/TestFiDataTransferProtocol.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1073/src/test/aop/org/apache/hadoop/hdfs/server/datanode/TestFiDataTransferProtocol.java?rev=1099687&r1=1099686&r2=1099687&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1073/src/test/aop/org/apache/hadoop/hdfs/server/datanode/TestFiDataTransferProtocol.java (original)
+++ hadoop/hdfs/branches/HDFS-1073/src/test/aop/org/apache/hadoop/hdfs/server/datanode/TestFiDataTransferProtocol.java Thu May  5 05:40:07 2011
@@ -306,4 +306,4 @@ public class TestFiDataTransferProtocol 
     final String methodName = FiTestUtil.getMethodName();
     runCallWritePacketToDisk(methodName, 2, new DoosAction(methodName, 2));
   }
-}
\ No newline at end of file
+}

Propchange: hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Thu May  5 05:40:07 2011
@@ -1,5 +1,6 @@
 /hadoop/core/branches/branch-0.19/hdfs/src/test/hdfs:713112
 /hadoop/core/trunk/src/test/hdfs:776175-785643
+/hadoop/hdfs/branches/HDFS-1052/src/test/hdfs:987665-1095512
 /hadoop/hdfs/branches/HDFS-265/src/test/hdfs:796829-820463
 /hadoop/hdfs/branches/branch-0.21/src/test/hdfs:820487
-/hadoop/hdfs/trunk/src/test/hdfs:1086482-1097628
+/hadoop/hdfs/trunk/src/test/hdfs:1086482-1099686

Modified: hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestFcHdfsCreateMkdir.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestFcHdfsCreateMkdir.java?rev=1099687&r1=1099686&r2=1099687&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestFcHdfsCreateMkdir.java (original)
+++ hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestFcHdfsCreateMkdir.java Thu May  5 05:40:07 2011
@@ -43,7 +43,7 @@ public class TestFcHdfsCreateMkdir exten
                                     throws IOException, LoginException, URISyntaxException  {
     Configuration conf = new HdfsConfiguration();
     cluster = new MiniDFSCluster.Builder(conf).numDataNodes(2).build();
-    fc = FileContext.getFileContext(cluster.getURI(), conf);
+    fc = FileContext.getFileContext(cluster.getURI(0), conf);
     defaultWorkingDirectory = fc.makeQualified( new Path("/user/" + 
         UserGroupInformation.getCurrentUser().getShortUserName()));
     fc.mkdir(defaultWorkingDirectory, FileContext.DEFAULT_PERM, true);

Modified: hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestFcHdfsPermission.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestFcHdfsPermission.java?rev=1099687&r1=1099686&r2=1099687&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestFcHdfsPermission.java (original)
+++ hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestFcHdfsPermission.java Thu May  5 05:40:07 2011
@@ -43,7 +43,7 @@ public class TestFcHdfsPermission extend
                                     throws IOException, LoginException, URISyntaxException  {
     Configuration conf = new HdfsConfiguration();
     cluster = new MiniDFSCluster.Builder(conf).numDataNodes(2).build();
-    fc = FileContext.getFileContext(cluster.getURI(), conf);
+    fc = FileContext.getFileContext(cluster.getURI(0), conf);
     defaultWorkingDirectory = fc.makeQualified( new Path("/user/" + 
         UserGroupInformation.getCurrentUser().getShortUserName()));
     fc.mkdir(defaultWorkingDirectory, FileContext.DEFAULT_PERM, true);

Modified: hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestFcHdfsSymlink.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestFcHdfsSymlink.java?rev=1099687&r1=1099686&r2=1099687&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestFcHdfsSymlink.java (original)
+++ hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestFcHdfsSymlink.java Thu May  5 05:40:07 2011
@@ -56,7 +56,7 @@ public class TestFcHdfsSymlink extends F
   }
 
   protected URI testURI() {
-    return cluster.getURI();
+    return cluster.getURI(0);
   }
 
   @Override
@@ -73,7 +73,7 @@ public class TestFcHdfsSymlink extends F
     conf.setBoolean(DFSConfigKeys.DFS_PERMISSIONS_ENABLED_KEY, true);
     conf.set(FsPermission.UMASK_LABEL, "000");
     cluster = new MiniDFSCluster.Builder(conf).build();
-    fc = FileContext.getFileContext(cluster.getURI());
+    fc = FileContext.getFileContext(cluster.getURI(0));
   }
   
   @AfterClass
@@ -130,7 +130,7 @@ public class TestFcHdfsSymlink extends F
     // Ditto when using another file context since the file system
     // for the slash is resolved according to the link's parent.
     FileContext localFc = FileContext.getLocalFSFileContext();
-    Path linkQual = new Path(cluster.getURI().toString(), fileViaLink); 
+    Path linkQual = new Path(cluster.getURI(0).toString(), fileViaLink); 
     assertEquals(fileSize, localFc.getFileStatus(linkQual).getLen());    
   }
   

Modified: hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestHDFSFileContextMainOperations.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestHDFSFileContextMainOperations.java?rev=1099687&r1=1099686&r2=1099687&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestHDFSFileContextMainOperations.java (original)
+++ hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/fs/TestHDFSFileContextMainOperations.java Thu May  5 05:40:07 2011
@@ -50,7 +50,7 @@ public class TestHDFSFileContextMainOper
       LoginException, URISyntaxException {
     cluster = new MiniDFSCluster.Builder(CONF).numDataNodes(2).build();
     cluster.waitClusterUp();
-    fc = FileContext.getFileContext(cluster.getURI(), CONF);
+    fc = FileContext.getFileContext(cluster.getURI(0), CONF);
     defaultWorkingDirectory = fc.makeQualified( new Path("/user/" + 
         UserGroupInformation.getCurrentUser().getShortUserName()));
     fc.mkdir(defaultWorkingDirectory, FileContext.DEFAULT_PERM, true);
@@ -64,7 +64,7 @@ public class TestHDFSFileContextMainOper
     cluster = new MiniDFSCluster.Builder(CONF).numDataNodes(1)
                                               .format(false).build();
     cluster.waitClusterUp();
-    fc = FileContext.getFileContext(cluster.getURI(), CONF);
+    fc = FileContext.getFileContext(cluster.getURI(0), CONF);
     defaultWorkingDirectory = fc.makeQualified( new Path("/user/" + 
         UserGroupInformation.getCurrentUser().getShortUserName()));
     fc.mkdir(defaultWorkingDirectory, FileContext.DEFAULT_PERM, true);

Modified: hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/BlockReaderTestUtil.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/BlockReaderTestUtil.java?rev=1099687&r1=1099686&r2=1099687&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/BlockReaderTestUtil.java (original)
+++ hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/BlockReaderTestUtil.java Thu May  5 05:40:07 2011
@@ -25,8 +25,8 @@ import java.util.Random;
 import java.util.List;
 import java.io.IOException;
 
-import org.apache.hadoop.hdfs.protocol.Block;
 import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
+import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
 import org.apache.hadoop.hdfs.protocol.LocatedBlock;
 import org.apache.hadoop.hdfs.server.common.HdfsConstants;
 import org.apache.hadoop.hdfs.server.datanode.DataNode;
@@ -125,7 +125,7 @@ public class BlockReaderTestUtil {
       throws IOException {
     InetSocketAddress targetAddr = null;
     Socket sock = null;
-    Block block = testBlock.getBlock();
+    ExtendedBlock block = testBlock.getBlock();
     DatanodeInfo[] nodes = testBlock.getLocations();
     targetAddr = NetUtils.createSocketAddr(nodes[0].getName());
     sock = new Socket();

Modified: hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/DFSTestUtil.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/DFSTestUtil.java?rev=1099687&r1=1099686&r2=1099687&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/DFSTestUtil.java (original)
+++ hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/DFSTestUtil.java Thu May  5 05:40:07 2011
@@ -50,7 +50,7 @@ import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileSystem.Statistics;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.DFSClient.DFSDataInputStream;
-import org.apache.hadoop.hdfs.protocol.Block;
+import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
 import org.apache.hadoop.hdfs.protocol.DataTransferProtocol;
 import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
 import org.apache.hadoop.hdfs.protocol.LocatedBlock;
@@ -58,6 +58,7 @@ import org.apache.hadoop.hdfs.protocol.L
 import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
 import org.apache.hadoop.hdfs.server.datanode.DataNode;
 import org.apache.hadoop.hdfs.server.datanode.TestTransferRbw;
+import org.apache.hadoop.hdfs.server.namenode.DatanodeDescriptor;
 import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
 import org.apache.hadoop.hdfs.server.namenode.NameNodeAdapter;
 import org.apache.hadoop.io.IOUtils;
@@ -238,7 +239,7 @@ public class DFSTestUtil {
   public static boolean allBlockReplicasCorrupt(MiniDFSCluster cluster,
       Path file, int blockNo) throws IOException {
     DFSClient client = new DFSClient(new InetSocketAddress("localhost",
-        cluster.getNameNodePort()), cluster.getConfiguration());
+        cluster.getNameNodePort()), cluster.getConfiguration(0));
     LocatedBlocks blocks;
     try {
        blocks = client.getNamenode().getBlockLocations(
@@ -254,7 +255,7 @@ public class DFSTestUtil {
    * the requested number of racks, with the requested number of
    * replicas, and the requested number of replicas still needed.
    */
-  public static void waitForReplication(MiniDFSCluster cluster, Block b,
+  public static void waitForReplication(MiniDFSCluster cluster, ExtendedBlock b,
       int racks, int replicas, int neededReplicas)
       throws IOException, TimeoutException, InterruptedException {
     int curRacks = 0;
@@ -265,7 +266,7 @@ public class DFSTestUtil {
 
     do {
       Thread.sleep(1000);
-      int []r = NameNodeAdapter.getReplicaInfo(cluster.getNameNode(), b);
+      int []r = NameNodeAdapter.getReplicaInfo(cluster.getNameNode(), b.getLocalBlock());
       curRacks = r[0];
       curReplicas = r[1];
       curNeededReplicas = r[2];
@@ -288,11 +289,11 @@ public class DFSTestUtil {
    * given block in the file contains the given number of corrupt replicas.
    */
   public static void waitCorruptReplicas(FileSystem fs, FSNamesystem ns,
-      Path file, Block b, int corruptRepls)
+      Path file, ExtendedBlock b, int corruptRepls)
       throws IOException, TimeoutException {
     int count = 0;
-    final int ATTEMPTS = 20;
-    int repls = ns.numCorruptReplicas(b);
+    final int ATTEMPTS = 50;
+    int repls = ns.numCorruptReplicas(b.getLocalBlock());
     while (repls != corruptRepls && count < ATTEMPTS) {
       try {
         IOUtils.copyBytes(fs.open(file), new IOUtils.NullOutputStream(),
@@ -301,7 +302,7 @@ public class DFSTestUtil {
         // Swallow exceptions
       }
       System.out.println("Waiting for "+corruptRepls+" corrupt replicas");
-      repls = ns.numCorruptReplicas(b);
+      repls = ns.numCorruptReplicas(b.getLocalBlock());
       count++;
     }
     if (count == ATTEMPTS) {
@@ -342,11 +343,11 @@ public class DFSTestUtil {
    * Returns the index of the first datanode which has a copy
    * of the given block, or -1 if no such datanode exists.
    */
-  public static int firstDnWithBlock(MiniDFSCluster cluster, Block b)
+  public static int firstDnWithBlock(MiniDFSCluster cluster, ExtendedBlock b)
       throws IOException {
     int numDatanodes = cluster.getDataNodes().size();
     for (int i = 0; i < numDatanodes; i++) {
-      String blockContent = cluster.readBlockOnDataNode(i, b.getBlockName());
+      String blockContent = cluster.readBlockOnDataNode(i, b);
       if (blockContent != null) {
         return i;
       }
@@ -354,6 +355,88 @@ public class DFSTestUtil {
     return -1;
   }
 
+  /*
+   * Return the total capacity of all live DNs.
+   */
+  public static long getLiveDatanodeCapacity(FSNamesystem ns) {
+    ArrayList<DatanodeDescriptor> live = new ArrayList<DatanodeDescriptor>();
+    ArrayList<DatanodeDescriptor> dead = new ArrayList<DatanodeDescriptor>();
+    ns.DFSNodesStatus(live, dead);
+    long capacity = 0;
+    for (final DatanodeDescriptor dn : live) {
+      capacity += dn.getCapacity();
+    }
+    return capacity;
+  }
+
+  /*
+   * Return the capacity of the given live DN.
+   */
+  public static long getDatanodeCapacity(FSNamesystem ns, int index) {
+    ArrayList<DatanodeDescriptor> live = new ArrayList<DatanodeDescriptor>();
+    ArrayList<DatanodeDescriptor> dead = new ArrayList<DatanodeDescriptor>();
+    ns.DFSNodesStatus(live, dead);
+    return live.get(index).getCapacity();
+  }
+
+  /*
+   * Wait for the given # live/dead DNs, total capacity, and # vol failures. 
+   */
+  public static void waitForDatanodeStatus(FSNamesystem ns, int expectedLive, 
+      int expectedDead, long expectedVolFails, long expectedTotalCapacity, 
+      long timeout) throws InterruptedException, TimeoutException {
+    ArrayList<DatanodeDescriptor> live = new ArrayList<DatanodeDescriptor>();
+    ArrayList<DatanodeDescriptor> dead = new ArrayList<DatanodeDescriptor>();
+    final int ATTEMPTS = 10;
+    int count = 0;
+    long currTotalCapacity = 0;
+    int volFails = 0;
+
+    do {
+      Thread.sleep(timeout);
+      live.clear();
+      dead.clear();
+      ns.DFSNodesStatus(live, dead);
+      currTotalCapacity = 0;
+      volFails = 0;
+      for (final DatanodeDescriptor dd : live) {
+        currTotalCapacity += dd.getCapacity();
+        volFails += dd.getVolumeFailures();
+      }
+      count++;
+    } while ((expectedLive != live.size() ||
+              expectedDead != dead.size() ||
+              expectedTotalCapacity != currTotalCapacity ||
+              expectedVolFails != volFails)
+             && count < ATTEMPTS);
+
+    if (count == ATTEMPTS) {
+      throw new TimeoutException("Timed out waiting for capacity."
+          + " Live = "+live.size()+" Expected = "+expectedLive
+          + " Dead = "+dead.size()+" Expected = "+expectedDead
+          + " Total capacity = "+currTotalCapacity
+          + " Expected = "+expectedTotalCapacity
+          + " Vol Fails = "+volFails+" Expected = "+expectedVolFails);
+    }
+  }
+
+  /*
+   * Wait for the given DN to consider itself dead.
+   */
+  public static void waitForDatanodeDeath(DataNode dn) 
+      throws InterruptedException, TimeoutException {
+    final int ATTEMPTS = 10;
+    int count = 0;
+    do {
+      Thread.sleep(1000);
+      count++;
+    } while (dn.isDatanodeUp() && count < ATTEMPTS);
+
+    if (count == ATTEMPTS) {
+      throw new TimeoutException("Timed out waiting for DN to die");
+    }
+  }
+  
   /** return list of filenames created as part of createFiles */
   public String[] getFileNames(String topDir) {
     if (nFiles == 0)
@@ -405,7 +488,7 @@ public class DFSTestUtil {
     files = null;
   }
   
-  public static Block getFirstBlock(FileSystem fs, Path path) throws IOException {
+  public static ExtendedBlock getFirstBlock(FileSystem fs, Path path) throws IOException {
     DFSDataInputStream in = 
       (DFSDataInputStream) ((DistributedFileSystem)fs).open(path);
     in.readByte();
@@ -553,7 +636,7 @@ public class DFSTestUtil {
   }
 
   /** For {@link TestTransferRbw} */
-  public static DataTransferProtocol.Status transferRbw(final Block b, 
+  public static DataTransferProtocol.Status transferRbw(final ExtendedBlock b, 
       final DFSClient dfsClient, final DatanodeInfo... datanodes) throws IOException {
     assertEquals(2, datanodes.length);
     final Socket s = DFSOutputStream.createSocketForPipeline(datanodes[0],

Modified: hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/MiniDFSCluster.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/MiniDFSCluster.java?rev=1099687&r1=1099686&r2=1099687&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/MiniDFSCluster.java (original)
+++ hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/MiniDFSCluster.java Thu May  5 05:40:07 2011
@@ -36,8 +36,6 @@ import java.util.Random;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
 import org.apache.hadoop.fs.FileSystem;
@@ -46,15 +44,20 @@ import org.apache.hadoop.hdfs.protocol.B
 import org.apache.hadoop.hdfs.protocol.BlockListAsLongs;
 import org.apache.hadoop.hdfs.protocol.ClientProtocol;
 import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
+import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
 import org.apache.hadoop.hdfs.protocol.FSConstants.DatanodeReportType;
+import org.apache.hadoop.hdfs.server.common.Storage;
 import org.apache.hadoop.hdfs.server.common.HdfsConstants.StartupOption;
 import org.apache.hadoop.hdfs.server.datanode.DataNode;
+import org.apache.hadoop.hdfs.server.datanode.DataStorage;
 import org.apache.hadoop.hdfs.server.datanode.FSDatasetInterface;
 import org.apache.hadoop.hdfs.server.datanode.SimulatedFSDataset;
+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.protocol.DatanodeProtocol;
+import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration;
 import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocol;
 import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocols;
 import org.apache.hadoop.hdfs.tools.DFSAdmin;
@@ -65,6 +68,7 @@ import org.apache.hadoop.security.Refres
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.authorize.ProxyUsers;
 import org.apache.hadoop.security.authorize.RefreshAuthorizationPolicyProtocol;
+import org.apache.hadoop.test.GenericTestUtils;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.ToolRunner;
 
@@ -73,16 +77,18 @@ import org.apache.hadoop.util.ToolRunner
  * The data directories for non-simulated DFS are under the testing directory.
  * For simulated data nodes, no underlying fs storage is used.
  */
-@InterfaceAudience.LimitedPrivate({"HBase", "HDFS", "Hive", "MapReduce", "Pig"})
-@InterfaceStability.Unstable
 public class MiniDFSCluster {
 
+  private static final String NAMESERVICE_ID_PREFIX = "nameserviceId";
+  private static final Log LOG = LogFactory.getLog(MiniDFSCluster.class);
+
   /**
    * Class to construct instances of MiniDFSClusters with specific options.
    */
   public static class Builder {
     private int nameNodePort = 0;
     private final Configuration conf;
+    private int numNameNodes = 1;
     private int numDataNodes = 1;
     private boolean format = true;
     private boolean manageNameDfsDirs = true;
@@ -91,15 +97,25 @@ public class MiniDFSCluster {
     private String[] racks = null; 
     private String [] hosts = null;
     private long [] simulatedCapacities = null;
-    // wait until namenode has left safe mode?
+    private String clusterId = null;
     private boolean waitSafeMode = true;
     private boolean setupHostsFile = false;
+    private boolean federation = false;
     
     public Builder(Configuration conf) {
       this.conf = conf;
     }
     
     /**
+     * default false - non federated cluster
+     * @param val
+     * @return Builder object
+     */
+    public Builder federation (boolean val){
+      this.federation = val;
+      return this;
+    }
+    /**
      * Default: 0
      */
     public Builder nameNodePort(int val) {
@@ -110,6 +126,14 @@ public class MiniDFSCluster {
     /**
      * Default: 1
      */
+    public Builder numNameNodes(int val) {
+      this.numNameNodes = val;
+      return this;
+    }
+
+    /**
+     * Default: 1
+     */
     public Builder numDataNodes(int val) {
       this.numDataNodes = val;
       return this;
@@ -180,6 +204,14 @@ public class MiniDFSCluster {
     }
     
     /**
+     * Default: null
+     */
+    public Builder clusterId(String cid) {
+      this.clusterId = cid;
+      return this;
+    }
+
+    /**
      * Default: false
      * When true the hosts file/include file for the cluster is setup
      */
@@ -200,6 +232,12 @@ public class MiniDFSCluster {
    * Used by builder to create and return an instance of MiniDFSCluster
    */
   private MiniDFSCluster(Builder builder) throws IOException {
+    LOG.info("starting cluster with " + builder.numNameNodes + " namenodes.");
+    nameNodes = new NameNodeInfo[builder.numNameNodes];
+    // try to determine if in federation mode
+    if(builder.numNameNodes > 1)
+      builder.federation = true;
+      
     initMiniDFSCluster(builder.nameNodePort,
                        builder.conf,
                        builder.numDataNodes,
@@ -210,8 +248,10 @@ public class MiniDFSCluster {
                        builder.racks,
                        builder.hosts,
                        builder.simulatedCapacities,
+                       builder.clusterId,
                        builder.waitSafeMode,
-                       builder.setupHostsFile);
+                       builder.setupHostsFile,
+                       builder.federation);
   }
   
   public class DataNodeProperties {
@@ -225,27 +265,35 @@ public class MiniDFSCluster {
       this.dnArgs = args;
     }
   }
-  private static final Log LOG = LogFactory.getLog(MiniDFSCluster.class);
 
-  private URI myUri = null;
   private Configuration conf;
-  private NameNode nameNode;
+  private NameNodeInfo[] nameNodes;
   private int numDataNodes;
   private ArrayList<DataNodeProperties> dataNodes = 
                          new ArrayList<DataNodeProperties>();
   private File base_dir;
   private File data_dir;
-
+  private boolean federation = false; 
   private boolean waitSafeMode = true;
   
-  public final static String FINALIZED_DIR_NAME = "/current/finalized/";
-  
+  /**
+   * Stores the information related to a namenode in the cluster
+   */
+  static class NameNodeInfo {
+    final NameNode nameNode;
+    final Configuration conf;
+    NameNodeInfo(NameNode nn, Configuration conf) {
+      this.nameNode = nn;
+      this.conf = conf;
+    }
+  }
   
   /**
    * This null constructor is used only when wishing to start a data node cluster
    * without a name node (ie when the name node is started elsewhere).
    */
   public MiniDFSCluster() {
+    nameNodes = new NameNodeInfo[0]; // No namenode in the cluster
   }
   
   /**
@@ -406,21 +454,22 @@ public class MiniDFSCluster {
                         StartupOption operation,
                         String[] racks, String hosts[],
                         long[] simulatedCapacities) throws IOException {
+    this.nameNodes = new NameNodeInfo[1]; // Single namenode in the cluster
     initMiniDFSCluster(nameNodePort, conf, numDataNodes, format,
         manageNameDfsDirs, manageDataDfsDirs, operation, racks, hosts,
-        simulatedCapacities, true, false);
+        simulatedCapacities, null, true, false, false);
   }
 
   private void initMiniDFSCluster(int nameNodePort, Configuration conf,
       int numDataNodes, boolean format, boolean manageNameDfsDirs,
       boolean manageDataDfsDirs, StartupOption operation, String[] racks,
-      String[] hosts, long[] simulatedCapacities, boolean waitSafeMode,
-      boolean setupHostsFile)
-    throws IOException {
+      String[] hosts, long[] simulatedCapacities, String clusterId,
+      boolean waitSafeMode, boolean setupHostsFile, boolean federation) 
+  throws IOException {
     this.conf = conf;
     base_dir = new File(getBaseDirectory());
     data_dir = new File(base_dir, "data");
-
+    this.federation = federation;
     this.waitSafeMode = waitSafeMode;
     
     // use alternate RPC engine if spec'd
@@ -444,73 +493,187 @@ public class MiniDFSCluster {
       conf.setBoolean(CommonConfigurationKeys.HADOOP_SECURITY_AUTHORIZATION,
                       false);
     }
-
-    // Setup the NameNode configuration
-    FileSystem.setDefaultUri(conf, "hdfs://localhost:"+ Integer.toString(nameNodePort));
-    conf.set(DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY, "127.0.0.1:0");  
-    if (manageNameDfsDirs) {
-      conf.set(DFSConfigKeys.DFS_NAMENODE_NAME_DIR_KEY,
-          fileAsURI(new File(base_dir, "name1"))+","+
-          fileAsURI(new File(base_dir, "name2")));
-      conf.set(DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_DIR_KEY,
-          fileAsURI(new File(base_dir, "namesecondary1"))+","+
-          fileAsURI(new File(base_dir, "namesecondary2")));
-    }
     
     int replication = conf.getInt(DFSConfigKeys.DFS_REPLICATION_KEY, 3);
     conf.setInt(DFSConfigKeys.DFS_REPLICATION_KEY, Math.min(replication, numDataNodes));
     conf.setInt(DFSConfigKeys.DFS_NAMENODE_SAFEMODE_EXTENSION_KEY, 0);
     conf.setInt(DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_INTERVAL_KEY, 3); // 3 second
+    conf.setClass(DFSConfigKeys.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY, 
+                   StaticMapping.class, DNSToSwitchMapping.class);
+    
+    Collection<String> nameserviceIds = DFSUtil.getNameServiceIds(conf);
+    if(nameserviceIds.size() > 1)  
+      federation = true;
+  
+    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");
+      NameNode nn = createNameNode(0, conf, numDataNodes, manageNameDfsDirs,
+          format, operation, clusterId);
+      nameNodes[0] = new NameNodeInfo(nn, conf);
+      FileSystem.setDefaultUri(conf, getURI(0));
+    } else {
+      if (nameserviceIds.isEmpty()) {
+        for (int i = 0; i < nameNodes.length; i++) {
+          nameserviceIds.add(NAMESERVICE_ID_PREFIX + i);
+        }
+      }
+      initFederationConf(conf, nameserviceIds, numDataNodes, nameNodePort);
+      createFederationNamenodes(conf, nameserviceIds, manageNameDfsDirs, format,
+          operation, clusterId);
+    }
     
-    // Format and clean out DataNode directories
     if (format) {
       if (data_dir.exists() && !FileUtil.fullyDelete(data_dir)) {
         throw new IOException("Cannot remove data directory: " + data_dir);
       }
-      NameNode.format(conf); 
     }
     
-    // Start the NameNode
-    String[] args = (operation == null ||
-                     operation == StartupOption.FORMAT ||
-                     operation == StartupOption.REGULAR) ?
-      new String[] {} : new String[] {operation.getName()};
-    conf.setClass(DFSConfigKeys.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY, 
-                   StaticMapping.class, DNSToSwitchMapping.class);
-    nameNode = NameNode.createNameNode(args, conf);
-    
     // Start the DataNodes
     startDataNodes(conf, numDataNodes, manageDataDfsDirs, operation, racks,
         hosts, simulatedCapacities, setupHostsFile);
     waitClusterUp();
-
     //make sure ProxyUsers uses the latest conf
     ProxyUsers.refreshSuperUserGroupsConfiguration(conf);
   }
   
+  /** Initialize configuration for federated cluster */
+  private static void initFederationConf(Configuration conf,
+      Collection<String> nameserviceIds, int numDataNodes, int nnPort) {
+    String nameserviceIdList = "";
+    for (String nameserviceId : nameserviceIds) {
+      // Create comma separated list of nameserviceIds
+      if (nameserviceIdList.length() > 0) {
+        nameserviceIdList += ",";
+      }
+      nameserviceIdList += nameserviceId;
+      initFederatedNamenodeAddress(conf, nameserviceId, nnPort);
+      nnPort = nnPort == 0 ? 0 : nnPort + 2;
+    }
+    conf.set(DFSConfigKeys.DFS_FEDERATION_NAMESERVICES, nameserviceIdList);
+  }
+
+  /* For federated namenode initialize the address:port */
+  private static void initFederatedNamenodeAddress(Configuration conf,
+      String nameserviceId, int nnPort) {
+    // Set nameserviceId specific key
+    String key = DFSUtil.getNameServiceIdKey(
+        DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY, nameserviceId);
+    conf.set(key, "127.0.0.1:0");
+
+    key = DFSUtil.getNameServiceIdKey(
+        DFSConfigKeys.DFS_NAMENODE_RPC_ADDRESS_KEY, nameserviceId);
+    conf.set(key, "127.0.0.1:" + nnPort);
+  }
+  
+  private void createFederationNamenodes(Configuration conf,
+      Collection<String> nameserviceIds, boolean manageNameDfsDirs,
+      boolean format, StartupOption operation, String clusterId)
+      throws IOException {
+    // Create namenodes in the cluster
+    int nnCounter = 0;
+    for (String nameserviceId : nameserviceIds) {
+      createFederatedNameNode(nnCounter++, conf, numDataNodes, manageNameDfsDirs,
+          format, operation, clusterId, nameserviceId);
+    }
+  }
+  
+  private NameNode createNameNode(int nnIndex, Configuration conf,
+      int numDataNodes, boolean manageNameDfsDirs, boolean format,
+      StartupOption operation, String clusterId)
+      throws IOException {
+    if (manageNameDfsDirs) {
+      conf.set(DFSConfigKeys.DFS_NAMENODE_NAME_DIR_KEY,
+          fileAsURI(new File(base_dir, "name" + (2*nnIndex + 1)))+","+
+          fileAsURI(new File(base_dir, "name" + (2*nnIndex + 2))));
+      conf.set(DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_DIR_KEY,
+          fileAsURI(new File(base_dir, "namesecondary" + (2*nnIndex + 1)))+","+
+          fileAsURI(new File(base_dir, "namesecondary" + (2*nnIndex + 2))));
+    }
+    
+    // Format and clean out DataNode directories
+    if (format) {
+      GenericTestUtils.formatNamenode(conf);
+    }
+    if (operation == StartupOption.UPGRADE){
+      operation.setClusterId(clusterId);
+    }
+    
+    // Start the NameNode
+    String[] args = (operation == null ||
+                     operation == StartupOption.FORMAT ||
+                     operation == StartupOption.REGULAR) ?
+      new String[] {} : new String[] {operation.getName()};
+    return NameNode.createNameNode(args, conf);
+  }
+  
+  private void createFederatedNameNode(int nnIndex, Configuration conf,
+      int numDataNodes, boolean manageNameDfsDirs, boolean format,
+      StartupOption operation, String clusterId, String nameserviceId)
+      throws IOException {
+    conf.set(DFSConfigKeys.DFS_FEDERATION_NAMESERVICE_ID, nameserviceId);
+    NameNode nn = createNameNode(nnIndex, conf, numDataNodes, manageNameDfsDirs,
+        format, operation, clusterId);
+    conf.set(DFSUtil.getNameServiceIdKey(
+        DFSConfigKeys.DFS_NAMENODE_RPC_ADDRESS_KEY, nameserviceId), NameNode
+        .getHostPortString(nn.getNameNodeAddress()));
+    conf.set(DFSUtil.getNameServiceIdKey(
+        DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY, nameserviceId), NameNode
+        .getHostPortString(nn.getHttpAddress()));
+    DFSUtil.setGenericConf(conf, nameserviceId, 
+        DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY);
+    nameNodes[nnIndex] = new NameNodeInfo(nn, new Configuration(conf));
+  }
+
   private void setRpcEngine(Configuration conf, Class<?> protocol, Class<?> engine) {
     conf.setClass("rpc.engine."+protocol.getName(), engine, Object.class);
   }
 
   /**
-   * 
-   * @return URI of this MiniDFSCluster
+   * @return URI of the namenode from a single namenode MiniDFSCluster
    */
   public URI getURI() {
-    return myUri;
+    checkSingleNameNode();
+    return getURI(0);
+  }
+  
+  /**
+   * @return URI of the given namenode in MiniDFSCluster
+   */
+  public URI getURI(int nnIndex) {
+    InetSocketAddress addr = nameNodes[nnIndex].nameNode.getNameNodeAddress();
+    String hostPort = NameNode.getHostPortString(addr);
+    URI uri = null;
+    try {
+      uri = new URI("hdfs://" + hostPort);
+    } catch (URISyntaxException e) {
+      NameNode.LOG.warn("unexpected URISyntaxException: " + e );
+    }
+    return uri;
   }
 
   /**
-   * Get configuration.
-   * @return Configuration of this MiniDFSCluster
+   * @return Configuration of for the given namenode
    */
-  public Configuration getConfiguration() {
-    return conf;
+  public Configuration getConfiguration(int nnIndex) {
+    return nameNodes[nnIndex].conf;
   }
 
   /**
-   * wait for the cluster to get out of 
-   * safemode.
+   * wait for the given namenode to get out of safemode.
+   */
+  public void waitNameNodeUp(int nnIndex) {
+    while (!isNameNodeUp(nnIndex)) {
+      try {
+        LOG.warn("Waiting for namenode at " + nnIndex + " to start...");
+        Thread.sleep(1000);
+      } catch (InterruptedException e) {
+      }
+    }
+  }
+  
+  /**
+   * wait for the cluster to get out of safemode.
    */
   public void waitClusterUp() {
     if (numDataNodes > 0) {
@@ -593,18 +756,6 @@ public class MiniDFSCluster {
     // If minicluster's name node is null assume that the conf has been
     // set with the right address:port of the name node.
     //
-    if (nameNode != null) { // set conf from the name node
-      InetSocketAddress nnAddr = nameNode.getNameNodeAddress(); 
-      int nameNodePort = nnAddr.getPort(); 
-      try {
-	  myUri = new URI("hdfs://"+ nnAddr.getHostName() + ":" +
-			  Integer.toString(nameNodePort));
-      } catch (URISyntaxException e) {
-	  throw new IOException("Couldn't parse own URI", e);
-      }
-      FileSystem.setDefaultUri(conf, myUri);
-    }
-
     if (racks != null && numDataNodes > racks.length ) {
       throw new IllegalArgumentException( "The length of racks [" + racks.length
           + "] is less than the number of datanodes [" + numDataNodes + "].");
@@ -615,7 +766,6 @@ public class MiniDFSCluster {
     }
     //Generate some hostnames if required
     if (racks != null && hosts == null) {
-      System.out.println("Generating host names for datanodes");
       hosts = new String[numDataNodes];
       for (int i = curDatanodesNum; i < curDatanodesNum + numDataNodes; i++) {
         hosts[i - curDatanodesNum] = "host" + i + ".foo.com";
@@ -639,8 +789,8 @@ public class MiniDFSCluster {
       // Set up datanode address
       setupDatanodeAddress(dnConf, setupHostsFile);
       if (manageDfsDirs) {
-        File dir1 = new File(data_dir, "data"+(2*i+1));
-        File dir2 = new File(data_dir, "data"+(2*i+2));
+        File dir1 = getStorageDir(i, 0);
+        File dir2 = getStorageDir(i, 1);
         dir1.mkdirs();
         dir2.mkdirs();
         if (!dir1.isDirectory() || !dir2.isDirectory()) { 
@@ -652,7 +802,7 @@ public class MiniDFSCluster {
         conf.set(DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY, dirs);
       }
       if (simulatedCapacities != null) {
-        dnConf.setBoolean("dfs.datanode.simulateddatastorage", true);
+        dnConf.setBoolean(SimulatedFSDataset.CONFIG_PROPERTY_SIMULATED, true);
         dnConf.setLong(SimulatedFSDataset.CONFIG_PROPERTY_CAPACITY,
             simulatedCapacities[i-curDatanodesNum]);
       }
@@ -689,7 +839,7 @@ public class MiniDFSCluster {
         StaticMapping.addNodeToRack(ipAddr + ":" + port,
                                   racks[i-curDatanodesNum]);
       }
-      DataNode.runDatanodeDaemon(dn);
+      dn.runDatanodeDaemon();
       dataNodes.add(new DataNodeProperties(dn, newconf, dnArgs));
     }
     curDatanodesNum += numDataNodes;
@@ -753,6 +903,30 @@ public class MiniDFSCluster {
                    simulatedCapacities, false);
     
   }
+
+  /**
+   * Finalize the namenode. Block pools corresponding to the namenode are
+   * finalized on the datanode.
+   */
+  private void finalizeNamenode(NameNode nn, Configuration conf) throws Exception {
+    if (nn == null) {
+      throw new IllegalStateException("Attempting to finalize "
+                                      + "Namenode but it is not running");
+    }
+    ToolRunner.run(new DFSAdmin(conf), new String[] {"-finalizeUpgrade"});
+  }
+  
+  /**
+   * Finalize cluster for the namenode at the given index 
+   * @see MiniDFSCluster#finalizeCluster(Configuration)
+   * @param nnIndex
+   * @param conf
+   * @throws Exception
+   */
+  public void finalizeCluster(int nnIndex, Configuration conf) throws Exception {
+    finalizeNamenode(nameNodes[nnIndex].nameNode, nameNodes[nnIndex].conf);
+  }
+
   /**
    * If the NameNode is running, attempt to finalize a previous upgrade.
    * When this method return, the NameNode should be finalized, but
@@ -761,18 +935,32 @@ public class MiniDFSCluster {
    * @throws IllegalStateException if the Namenode is not running.
    */
   public void finalizeCluster(Configuration conf) throws Exception {
-    if (nameNode == null) {
-      throw new IllegalStateException("Attempting to finalize "
-                                      + "Namenode but it is not running");
+    for (NameNodeInfo nnInfo : nameNodes) {
+      if (nnInfo == null) {
+        throw new IllegalStateException("Attempting to finalize "
+            + "Namenode but it is not running");
+      }
+      finalizeNamenode(nnInfo.nameNode, nnInfo.conf);
     }
-    ToolRunner.run(new DFSAdmin(conf), new String[] {"-finalizeUpgrade"});
+  }
+  
+  public int getNumNameNodes() {
+    return nameNodes.length;
   }
   
   /**
    * Gets the started NameNode.  May be null.
    */
   public NameNode getNameNode() {
-    return nameNode;
+    checkSingleNameNode();
+    return getNameNode(0);
+  }
+  
+  /**
+   * Gets the NameNode for the index.  May be null.
+   */
+  public NameNode getNameNode(int nnIndex) {
+    return nameNodes[nnIndex].nameNode;
   }
   
   /**
@@ -780,7 +968,12 @@ public class MiniDFSCluster {
    * @return {@link FSNamesystem} object.
    */
   public FSNamesystem getNamesystem() {
-    return NameNodeAdapter.getNamesystem(nameNode);
+    checkSingleNameNode();
+    return NameNodeAdapter.getNamesystem(nameNodes[0].nameNode);
+  }
+  
+  public FSNamesystem getNamesystem(int nnIndex) {
+    return NameNodeAdapter.getNamesystem(nameNodes[nnIndex].nameNode);
   }
 
   /**
@@ -808,21 +1001,34 @@ public class MiniDFSCluster {
   /**
    * Gets the rpc port used by the NameNode, because the caller 
    * supplied port is not necessarily the actual port used.
+   * Assumption: cluster has a single namenode
    */     
   public int getNameNodePort() {
-    return nameNode.getNameNodeAddress().getPort();
+    checkSingleNameNode();
+    return getNameNodePort(0);
   }
     
   /**
-   * Shut down the servers that are up.
+   * Gets the rpc port used by the NameNode at the given index, because the
+   * caller supplied port is not necessarily the actual port used.
+   */     
+  public int getNameNodePort(int nnIndex) {
+    return nameNodes[nnIndex].nameNode.getNameNodeAddress().getPort();
+  }
+    
+  /**
+   * Shutdown all the nodes in the cluster.
    */
   public void shutdown() {
     System.out.println("Shutting down the Mini HDFS Cluster");
     shutdownDataNodes();
-    if (nameNode != null) {
-      nameNode.stop();
-      nameNode.join();
-      nameNode = null;
+    for (NameNodeInfo nnInfo : nameNodes) {
+      NameNode nameNode = nnInfo.nameNode;
+      if (nameNode != null) {
+        nameNode.stop();
+        nameNode.join();
+        nameNode = null;
+      }
     }
   }
   
@@ -840,23 +1046,36 @@ public class MiniDFSCluster {
   }
 
   /**
-   * Shutdown namenode.
+   * Shutdown all the namenodes.
    */
-  public synchronized void shutdownNameNode() {
-    if (nameNode != null) {
+  public synchronized void shutdownNameNodes() {
+    for (int i = 0; i < nameNodes.length; i++) {
+      shutdownNameNode(i);
+    }
+  }
+  
+  /**
+   * Shutdown the namenode at a given index.
+   */
+  public synchronized void shutdownNameNode(int nnIndex) {
+    NameNode nn = nameNodes[nnIndex].nameNode;
+    if (nn != null) {
       System.out.println("Shutting down the namenode");
-      nameNode.stop();
-      nameNode.join();
-      nameNode = null;
+      nn.stop();
+      nn.join();
+      Configuration conf = nameNodes[nnIndex].conf;
+      nameNodes[nnIndex] = new NameNodeInfo(null, conf);
     }
   }
 
   /**
-   * Restart namenode.
+   * Restart namenode at a given index.
    */
-  public synchronized void restartNameNode() throws IOException {
-    shutdownNameNode();
-    nameNode = NameNode.createNameNode(new String[] {}, conf);
+  public synchronized void restartNameNode(int nnIndex) throws IOException {
+    Configuration conf = nameNodes[nnIndex].conf;
+    shutdownNameNode(nnIndex);
+    NameNode nn = NameNode.createNameNode(new String[] {}, conf);
+    nameNodes[nnIndex] = new NameNodeInfo(nn, conf);
     waitClusterUp();
     System.out.println("Restarted the namenode");
     int failedCount = 0;
@@ -881,79 +1100,64 @@ public class MiniDFSCluster {
   /**
    * Return the contents of the given block on the given datanode.
    *
-   * @param The index of the datanode
-   * @param The name of the block
+   * @param block block to be corrupted
    * @throws IOException on error accessing the file for the given block
-   * @return The contents of the block file, null if none found
    */
-  public String readBlockOnDataNode(int i, String blockName) throws IOException {
-    assert (i >= 0 && i < dataNodes.size()) : "Invalid datanode "+i;
-
-    // Each datanode has multiple data dirs, check each
-    for (int dn = i*2; dn < i*2+2; dn++) {
-      File dataDir = new File(getBaseDirectory() + "data");
-      File blockFile = new File(dataDir,
-          "data" + (dn+1) + FINALIZED_DIR_NAME + blockName);
-      if (blockFile.exists()) {
-        return DFSTestUtil.readFile(blockFile);
-      }
-    }
-    return null;
-  }
-
-  /**
-   * Corrupt a block on all datanodes.
-   *
-   * @param The name of the block
-   * @throws IOException on error accessing the given block.
-   * @return The number of block files corrupted.
-   */  
-  public int corruptBlockOnDataNodes(String blockName) throws IOException {
+  public int corruptBlockOnDataNodes(ExtendedBlock block) throws IOException{
     int blocksCorrupted = 0;
-    for (int i=0; i < dataNodes.size(); i++) {
-      if (corruptReplica(blockName, i)) {
+    File[] blockFiles = getAllBlockFiles(block);
+    for (File f : blockFiles) {
+      if (corruptBlock(f)) {
         blocksCorrupted++;
       }
     }
     return blocksCorrupted;
   }
 
+  public String readBlockOnDataNode(int i, ExtendedBlock block)
+      throws IOException {
+    assert (i >= 0 && i < dataNodes.size()) : "Invalid datanode "+i;
+    File blockFile = getBlockFile(i, block);
+    if (blockFile != null && blockFile.exists()) {
+      return DFSTestUtil.readFile(blockFile);
+    }
+    return null;
+  }
+
   /**
    * Corrupt a block on a particular datanode.
    *
-   * @param The index of the datanode
-   * @param The name of the block
+   * @param i index of the datanode
+   * @param blk name of the block
    * @throws IOException on error accessing the given block or if
    * the contents of the block (on the same datanode) differ.
    * @return true if a replica was corrupted, false otherwise
+   * Types: delete, write bad data, truncate
    */
-  public boolean corruptReplica(String blockName, int i) throws IOException {
-    Random random = new Random();
-    assert (i >= 0 && i < dataNodes.size()) : "Invalid datanode "+i;
-    int filesCorrupted = 0;
+  public static boolean corruptReplica(int i, ExtendedBlock blk)
+      throws IOException {
+    File blockFile = getBlockFile(i, blk);
+    return corruptBlock(blockFile);
+  }
 
-    // Each datanode has multiple data dirs, check each
-    for (int dn = i*2; dn < i*2+2; dn++) {
-      File dataDir = new File(getBaseDirectory() + "data");
-      File blockFile = new File(dataDir,
-          "data" + (dn+1) + FINALIZED_DIR_NAME + blockName);
-
-      // Corrupt the replica by writing some bytes into a random offset
-      if (blockFile.exists()) { 
-        System.out.println("Corrupting " + blockFile);
-        RandomAccessFile raFile = new RandomAccessFile(blockFile, "rw");
-        FileChannel channel = raFile.getChannel();
-        String badString = "BADBAD";
-        int rand = random.nextInt((int)channel.size()/2);
-        raFile.seek(rand);
-        raFile.write(badString.getBytes());
-        raFile.close();
-        filesCorrupted++;
-      }
-    }
-    assert filesCorrupted == 0 || filesCorrupted == 1
-      : "Unexpected # block files";
-    return filesCorrupted == 1;
+  /*
+   * Corrupt a block on a particular datanode
+   */
+  public static boolean corruptBlock(File blockFile) throws IOException {
+    if (blockFile == null || !blockFile.exists()) {
+      return false;
+    }
+    // Corrupt replica by writing random bytes into replica
+    Random random = new Random();
+    RandomAccessFile raFile = new RandomAccessFile(blockFile, "rw");
+    FileChannel channel = raFile.getChannel();
+    String badString = "BADBAD";
+    int rand = random.nextInt((int)channel.size()/2);
+    raFile.seek(rand);
+    raFile.write(badString.getBytes());
+    raFile.close();
+    LOG.warn("Corrupting the block " + blockFile);
+    return true;
   }
 
   /*
@@ -966,7 +1170,7 @@ public class MiniDFSCluster {
     DataNodeProperties dnprop = dataNodes.remove(i);
     DataNode dn = dnprop.datanode;
     System.out.println("MiniDFSCluster Stopping DataNode " + 
-                       dn.dnRegistration.getName() +
+                       dn.getMachineName() +
                        " from a total of " + (dataNodes.size() + 1) + 
                        " datanodes.");
     dn.shutdown();
@@ -981,7 +1185,12 @@ public class MiniDFSCluster {
     int i;
     for (i = 0; i < dataNodes.size(); i++) {
       DataNode dn = dataNodes.get(i).datanode;
-      if (dn.dnRegistration.getName().equals(name)) {
+      // get BP registration
+      DatanodeRegistration dnR = 
+        DataNodeTestUtils.getDNRegistrationByMachineName(dn, name);
+      LOG.info("for name=" + name + " found bp=" + dnR + 
+          "; with dnMn=" + dn.getMachineName());
+      if(dnR != null) {
         break;
       }
     }
@@ -1065,7 +1274,8 @@ public class MiniDFSCluster {
    * Returns true if the NameNode is running and is out of Safe Mode
    * or if waiting for safe mode is disabled.
    */
-  public boolean isClusterUp() {
+  public boolean isNameNodeUp(int nnIndex) {
+    NameNode nameNode = nameNodes[nnIndex].nameNode;
     if (nameNode == null) {
       return false;
     }
@@ -1076,6 +1286,18 @@ public class MiniDFSCluster {
     }
     return isUp;
   }
+
+  /**
+   * Returns true if all the NameNodes are running and is out of Safe Mode.
+   */
+  public boolean isClusterUp() {
+    for (int index = 0; index < nameNodes.length; index++) {
+      if (!isNameNodeUp(index)) {
+        return false;
+      }
+    }
+    return true;
+  }
   
   /**
    * Returns true if there is at least one DataNode running.
@@ -1084,33 +1306,55 @@ public class MiniDFSCluster {
     if (dataNodes == null || dataNodes.size() == 0) {
       return false;
     }
-    return true;
+    for (DataNodeProperties dn : dataNodes) {
+      if (dn.datanode.isDatanodeUp()) {
+        return true;
+      }
+    }
+    return false;
   }
   
   /**
-   * Get a client handle to the DFS cluster.
+   * Get a client handle to the DFS cluster with a single namenode.
    */
   public FileSystem getFileSystem() throws IOException {
-    return FileSystem.get(conf);
+    checkSingleNameNode();
+    return getFileSystem(0);
   }
   
+  /**
+   * Get a client handle to the DFS cluster for the namenode at given index.
+   */
+  public FileSystem getFileSystem(int nnIndex) throws IOException {
+    return FileSystem.get(getURI(nnIndex), nameNodes[nnIndex].conf);
+  }
 
   /**
    * Get another FileSystem instance that is different from FileSystem.get(conf).
    * This simulating different threads working on different FileSystem instances.
    */
-  public FileSystem getNewFileSystemInstance() throws IOException {
-    return FileSystem.newInstance(conf);
+  public FileSystem getNewFileSystemInstance(int nnIndex) throws IOException {
+    return FileSystem.newInstance(getURI(nnIndex), nameNodes[nnIndex].conf);
+  }
+  
+  /**
+   * @return a http URL
+   */
+  public String getHttpUri(int nnIndex) throws IOException {
+    return "http://"
+        + nameNodes[nnIndex].conf
+            .get(DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY);
   }
   
   /**
    * @return a {@link HftpFileSystem} object.
    */
-  public HftpFileSystem getHftpFileSystem() throws IOException {
-    final String str = "hftp://"
-        + conf.get(DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY);
+  public HftpFileSystem getHftpFileSystem(int nnIndex) throws IOException {
+    String uri = "hftp://"
+        + nameNodes[nnIndex].conf
+            .get(DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY);
     try {
-      return (HftpFileSystem)FileSystem.get(new URI(str), conf); 
+      return (HftpFileSystem)FileSystem.get(new URI(uri), conf);
     } catch (URISyntaxException e) {
       throw new IOException(e);
     }
@@ -1120,14 +1364,14 @@ public class MiniDFSCluster {
    *  @return a {@link HftpFileSystem} object as specified user. 
    */
   public HftpFileSystem getHftpFileSystemAs(final String username,
-      final Configuration conf, final String... groups
-      ) throws IOException, InterruptedException {
+      final Configuration conf, final int nnIndex, final String... groups)
+      throws IOException, InterruptedException {
     final UserGroupInformation ugi = UserGroupInformation.createUserForTesting(
         username, groups);
     return ugi.doAs(new PrivilegedExceptionAction<HftpFileSystem>() {
       @Override
       public HftpFileSystem run() throws Exception {
-        return getHftpFileSystem();
+        return getHftpFileSystem(nnIndex);
       }
     });
   }
@@ -1135,31 +1379,29 @@ public class MiniDFSCluster {
   /**
    * Get the directories where the namenode stores its image.
    */
-  public Collection<URI> getNameDirs() {
-    return FSNamesystem.getNamespaceDirs(conf);
+  public Collection<URI> getNameDirs(int nnIndex) {
+    return FSNamesystem.getNamespaceDirs(nameNodes[nnIndex].conf);
   }
 
   /**
    * Get the directories where the namenode stores its edits.
    */
-  public Collection<URI> getNameEditsDirs() {
-    return FSNamesystem.getNamespaceEditsDirs(conf);
+  public Collection<URI> getNameEditsDirs(int nnIndex) {
+    return FSNamesystem.getNamespaceEditsDirs(nameNodes[nnIndex].conf);
   }
 
-  /**
-   * Wait until the cluster is active and running.
-   */
-  public void waitActive() throws IOException {
-    if (nameNode == null) {
+  /** Wait until the given namenode gets registration from all the datanodes */
+  public void waitActive(int nnIndex) throws IOException {
+    if (nameNodes.length == 0 || nameNodes[nnIndex] == null) {
       return;
     }
-    InetSocketAddress addr = new InetSocketAddress("localhost",
-                                                   getNameNodePort());
+    InetSocketAddress addr = nameNodes[nnIndex].nameNode.getServiceRpcAddress();
     DFSClient client = new DFSClient(addr, conf);
 
-    // make sure all datanodes have registered and sent heartbeat
-    while (shouldWait(client.datanodeReport(DatanodeReportType.LIVE))) {
+    // ensure all datanodes have registered and sent heartbeat to the namenode
+    while (shouldWait(client.datanodeReport(DatanodeReportType.LIVE), addr)) {
       try {
+        LOG.info("Waiting for cluster to become active");
         Thread.sleep(100);
       } catch (InterruptedException e) {
       }
@@ -1168,10 +1410,39 @@ public class MiniDFSCluster {
     client.close();
   }
   
-  private synchronized boolean shouldWait(DatanodeInfo[] dnInfo) {
+  /**
+   * Wait until the cluster is active and running.
+   */
+  public void waitActive() throws IOException {
+    for (int index = 0; index < nameNodes.length; index++) {
+      waitActive(index);
+    }
+  }
+  
+  private synchronized boolean shouldWait(DatanodeInfo[] dnInfo,
+      InetSocketAddress addr) {
+    // If a datanode failed to start, then do not wait
+    for (DataNodeProperties dn : dataNodes) {
+      // the datanode thread communicating with the namenode should be alive
+      if (!dn.datanode.isBPServiceAlive(addr)) {
+        LOG.warn("BPOfferService failed to start in datanode " + dn.datanode
+            + " for namenode at " + addr);
+        return false;
+      }
+    }
+    
+    // Wait for expected number of datanodes to start
     if (dnInfo.length != numDataNodes) {
       return true;
     }
+    
+    // if one of the data nodes is not fully started, continue to wait
+    for (DataNodeProperties dn : dataNodes) {
+      if (!dn.datanode.isDatanodeFullyStarted()) {
+        return true;
+      }
+    }
+    
     // make sure all datanodes have sent first heartbeat to namenode,
     // using (capacity == 0) as proxy.
     for (DatanodeInfo dn : dnInfo) {
@@ -1179,6 +1450,13 @@ public class MiniDFSCluster {
         return true;
       }
     }
+    
+    // If datanode dataset is not initialized then wait
+    for (DataNodeProperties dn : dataNodes) {
+      if (dn.datanode.data == null) {
+        return true;
+      }
+    }
     return false;
   }
 
@@ -1195,11 +1473,12 @@ public class MiniDFSCluster {
    * @param dataNodeIndex - data node whose block report is desired - the index is same as for getDataNodes()
    * @return the block report for the specified data node
    */
-  public Iterable<Block> getBlockReport(int dataNodeIndex) {
+  public Iterable<Block> getBlockReport(String bpid, int dataNodeIndex) {
     if (dataNodeIndex < 0 || dataNodeIndex > dataNodes.size()) {
       throw new IndexOutOfBoundsException();
     }
-    return dataNodes.get(dataNodeIndex).datanode.getFSDataset().getBlockReport();
+    return dataNodes.get(dataNodeIndex).datanode.getFSDataset().getBlockReport(
+        bpid);
   }
   
   
@@ -1208,11 +1487,11 @@ public class MiniDFSCluster {
    * @return block reports from all data nodes
    *    BlockListAsLongs is indexed in the same order as the list of datanodes returned by getDataNodes()
    */
-  public Iterable<Block>[] getAllBlockReports() {
+  public Iterable<Block>[] getAllBlockReports(String bpid) {
     int numDataNodes = dataNodes.size();
     Iterable<Block>[] result = new BlockListAsLongs[numDataNodes];
     for (int i = 0; i < numDataNodes; ++i) {
-     result[i] = getBlockReport(i);
+     result[i] = getBlockReport(bpid, i);
     }
     return result;
   }
@@ -1235,11 +1514,30 @@ public class MiniDFSCluster {
     if (!(dataSet instanceof SimulatedFSDataset)) {
       throw new IOException("injectBlocks is valid only for SimilatedFSDataset");
     }
+    String bpid = getNamesystem().getBlockPoolId();
     SimulatedFSDataset sdataset = (SimulatedFSDataset) dataSet;
-    sdataset.injectBlocks(blocksToInject);
-    dataNodes.get(dataNodeIndex).datanode.scheduleBlockReport(0);
+    sdataset.injectBlocks(bpid, blocksToInject);
+    dataNodes.get(dataNodeIndex).datanode.scheduleAllBlockReport(0);
   }
-  
+
+  /**
+   * Multiple-NameNode version of {@link #injectBlocks(Iterable[])}.
+   */
+  public void injectBlocks(int nameNodeIndex, int dataNodeIndex,
+      Iterable<Block> blocksToInject) throws IOException {
+    if (dataNodeIndex < 0 || dataNodeIndex > dataNodes.size()) {
+      throw new IndexOutOfBoundsException();
+    }
+    FSDatasetInterface dataSet = dataNodes.get(dataNodeIndex).datanode.getFSDataset();
+    if (!(dataSet instanceof SimulatedFSDataset)) {
+      throw new IOException("injectBlocks is valid only for SimilatedFSDataset");
+    }
+    String bpid = getNamesystem(nameNodeIndex).getBlockPoolId();
+    SimulatedFSDataset sdataset = (SimulatedFSDataset) dataSet;
+    sdataset.injectBlocks(bpid, blocksToInject);
+    dataNodes.get(dataNodeIndex).datanode.scheduleAllBlockReport(0);
+  }
+
   /**
    * This method is valid only if the data nodes have simulated data
    * @param blocksToInject - blocksToInject[] is indexed in the same order as the list 
@@ -1249,7 +1547,8 @@ public class MiniDFSCluster {
    *             if any of blocks already exist in the data nodes
    *             Note the rest of the blocks are not injected.
    */
-  public void injectBlocks(Iterable<Block>[] blocksToInject) throws IOException {
+  public void injectBlocks(Iterable<Block>[] blocksToInject)
+      throws IOException {
     if (blocksToInject.length >  dataNodes.size()) {
       throw new IndexOutOfBoundsException();
     }
@@ -1288,6 +1587,163 @@ public class MiniDFSCluster {
   public static String getBaseDirectory() {
     return System.getProperty("test.build.data", "build/test/data") + "/dfs/";
   }
+
+  /**
+   * Get a storage directory for a datanode. There are two storage directories
+   * per datanode:
+   * <ol>
+   * <li><base directory>/data/data<2*dnIndex + 1></li>
+   * <li><base directory>/data/data<2*dnIndex + 2></li>
+   * </ol>
+   * 
+   * @param dnIndex datanode index (starts from 0)
+   * @param dirIndex directory index (0 or 1). Index 0 provides access to the
+   *          first storage directory. Index 1 provides access to the second
+   *          storage directory.
+   * @return Storage directory
+   */
+  public static File getStorageDir(int dnIndex, int dirIndex) {
+    return new File(getBaseDirectory() + "data/data" + (2*dnIndex + 1 + dirIndex));
+  }
+  
+  /**
+   * Get current directory corresponding to the datanode
+   * @param storageDir
+   * @return current directory
+   */
+  public static String getDNCurrentDir(File storageDir) {
+    return storageDir + "/" + Storage.STORAGE_DIR_CURRENT + "/";
+  }
+  
+  /**
+   * Get directory corresponding to block pool directory in the datanode
+   * @param storageDir
+   * @return current directory
+   */
+  public static String getBPDir(File storageDir, String bpid) {
+    return getDNCurrentDir(storageDir) + bpid + "/";
+  }
+  /**
+   * Get directory relative to block pool directory in the datanode
+   * @param storageDir
+   * @return current directory
+   */
+  public static String getBPDir(File storageDir, String bpid, String dirName) {
+    return getBPDir(storageDir, bpid) + dirName + "/";
+  }
+  
+  /**
+   * Get finalized directory for a block pool
+   * @param storageDir storage directory
+   * @param bpid Block pool Id
+   * @return finalized directory for a block pool
+   */
+  public static File getRbwDir(File storageDir, String bpid) {
+    return new File(getBPDir(storageDir, bpid, Storage.STORAGE_DIR_CURRENT)
+        + DataStorage.STORAGE_DIR_RBW );
+  }
+  
+  /**
+   * Get finalized directory for a block pool
+   * @param storageDir storage directory
+   * @param bpid Block pool Id
+   * @return finalized directory for a block pool
+   */
+  public static File getFinalizedDir(File storageDir, String bpid) {
+    return new File(getBPDir(storageDir, bpid, Storage.STORAGE_DIR_CURRENT)
+        + DataStorage.STORAGE_DIR_FINALIZED );
+  }
+  
+  /**
+   * Get file correpsonding to a block
+   * @param storageDir storage directory
+   * @param blk block to be corrupted
+   * @return file corresponding to the block
+   */
+  public static File getBlockFile(File storageDir, ExtendedBlock blk) {
+    return new File(getFinalizedDir(storageDir, blk.getBlockPoolId()), 
+        blk.getBlockName());
+  }
+  
+  /**
+   * Get all files related to a block from all the datanodes
+   * @param block block for which corresponding files are needed
+   */
+  public File[] getAllBlockFiles(ExtendedBlock block) {
+    if (dataNodes.size() == 0) return new File[0];
+    ArrayList<File> list = new ArrayList<File>();
+    for (int i=0; i < dataNodes.size(); i++) {
+      File blockFile = getBlockFile(i, block);
+      if (blockFile != null) {
+        list.add(blockFile);
+      }
+    }
+    return list.toArray(new File[list.size()]);
+  }
+  
+  /**
+   * Get files related to a block for a given datanode
+   * @param dnIndex Index of the datanode to get block files for
+   * @param block block for which corresponding files are needed
+   */
+  public static File getBlockFile(int dnIndex, ExtendedBlock block) {
+    // Check for block file in the two storage directories of the datanode
+    for (int i = 0; i <=1 ; i++) {
+      File storageDir = MiniDFSCluster.getStorageDir(dnIndex, i);
+      File blockFile = getBlockFile(storageDir, block);
+      if (blockFile.exists()) {
+        return blockFile;
+      }
+    }
+    return null;
+  }
+  
+  /**
+   * Throw an exception if the MiniDFSCluster is not started with a single
+   * namenode
+   */
+  private void checkSingleNameNode() {
+    if (nameNodes.length != 1) {
+      throw new IllegalArgumentException("Namenode index is needed");
+    }
+  }
+
+  /**
+   * Add a namenode to a federated cluster and start it. Configuration of
+   * datanodes in the cluster is refreshed to register with the new namenode.
+   * 
+   * @return newly started namenode
+   */
+  public NameNode addNameNode(Configuration conf, int namenodePort)
+      throws IOException {
+    if(!federation)
+      throw new IOException("cannot add namenode to non-federated cluster");
+    
+    int nnIndex = nameNodes.length;
+    int numNameNodes = nameNodes.length + 1;
+    NameNodeInfo[] newlist = new NameNodeInfo[numNameNodes];
+    System.arraycopy(nameNodes, 0, newlist, 0, nameNodes.length);
+    nameNodes = newlist;
+    String nameserviceId = NAMESERVICE_ID_PREFIX + (nnIndex + 1);
+    
+    String nameserviceIds = conf.get(DFSConfigKeys.DFS_FEDERATION_NAMESERVICES);
+    nameserviceIds += "," + nameserviceId;
+    conf.set(DFSConfigKeys.DFS_FEDERATION_NAMESERVICES, nameserviceIds);
+    
+    initFederatedNamenodeAddress(conf, nameserviceId, namenodePort);
+    createFederatedNameNode(nnIndex, conf, numDataNodes, true, true, null,
+        null, nameserviceId);
+
+    // Refresh datanodes with the newly started namenode
+    for (DataNodeProperties dn : dataNodes) {
+      DataNode datanode = dn.datanode;
+      datanode.refreshNamenodes(conf);
+    }
+
+    // Wait for new namenode to get registrations from all the datanodes
+    waitActive(nnIndex);
+    return nameNodes[nnIndex].nameNode;
+  }
   
   private int getFreeSocketPort() {
     int port = 0;
@@ -1322,9 +1778,7 @@ public class MiniDFSCluster {
   
   private void addToFile(String p, String address) throws IOException {
     File f = new File(p);
-    if (!f.exists()) {
-      f.createNewFile();
-    }
+    f.createNewFile();
     PrintWriter writer = new PrintWriter(new FileWriter(f, true));
     try {
       writer.println(address);

Modified: hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestBlockMissingException.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestBlockMissingException.java?rev=1099687&r1=1099686&r2=1099687&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestBlockMissingException.java (original)
+++ hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestBlockMissingException.java Thu May  5 05:40:07 2011
@@ -19,21 +19,18 @@ package org.apache.hadoop.hdfs;
 
 import java.io.File;
 import java.io.IOException;
-import java.util.Properties;
 
 import junit.framework.TestCase;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
-import org.apache.hadoop.util.StringUtils;
-import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
-import org.apache.hadoop.hdfs.protocol.Block;
+import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
 import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
 import org.apache.hadoop.hdfs.DistributedFileSystem;
 import org.apache.hadoop.hdfs.BlockMissingException;
@@ -64,7 +61,7 @@ public class TestBlockMissingException e
       // extract block locations from File system. Wait till file is closed.
       LocatedBlocks locations = null;
       locations = fileSys.dfs.getNamenode().getBlockLocations(file1.toString(),
-                                                             0, numBlocks * blockSize);
+          0, numBlocks * blockSize);
       // remove block of file
       LOG.info("Remove first block of file");
       corruptBlock(file1, locations.get(0).getBlock());
@@ -119,46 +116,15 @@ public class TestBlockMissingException e
     assertTrue("Expected BlockMissingException ", gotException);
   }
 
-  /*
-   * The Data directories for a datanode
-   */
-  private File[] getDataNodeDirs(int i) throws IOException {
-    String base_dir = MiniDFSCluster.getBaseDirectory();
-    File data_dir = new File(base_dir, "data");
-    File dir1 = new File(data_dir, "data"+(2*i+1));
-    File dir2 = new File(data_dir, "data"+(2*i+2));
-    if (dir1.isDirectory() && dir2.isDirectory()) {
-      File[] dir = new File[2];
-      dir[0] = new File(dir1, MiniDFSCluster.FINALIZED_DIR_NAME);
-      dir[1] = new File(dir2, MiniDFSCluster.FINALIZED_DIR_NAME); 
-      return dir;
-    }
-    return new File[0];
-  }
-
   //
   // Corrupt specified block of file
   //
-  void corruptBlock(Path file, Block blockNum) throws IOException {
-    long id = blockNum.getBlockId();
-
-    // Now deliberately remove/truncate data blocks from the block.
-    //
-    for (int i = 0; i < NUM_DATANODES; i++) {
-      File[] dirs = getDataNodeDirs(i);
-      
-      for (int j = 0; j < dirs.length; j++) {
-        File[] blocks = dirs[j].listFiles();
-        assertTrue("Blocks do not exist in data-dir", (blocks != null) && (blocks.length >= 0));
-        for (int idx = 0; idx < blocks.length; idx++) {
-          if (blocks[idx].getName().startsWith("blk_" + id) &&
-              !blocks[idx].getName().endsWith(".meta")) {
-            blocks[idx].delete();
-            LOG.info("Deleted block " + blocks[idx]);
-          }
-        }
-      }
+  void corruptBlock(Path file, ExtendedBlock blk) {
+    // Now deliberately remove/truncate data blocks from the file.
+    File[] blockFiles = dfs.getAllBlockFiles(blk);
+    for (File f : blockFiles) {
+      f.delete();
+      LOG.info("Deleted block " + f);
     }
   }
-
 }

Modified: hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestClientBlockVerification.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestClientBlockVerification.java?rev=1099687&r1=1099686&r2=1099687&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestClientBlockVerification.java (original)
+++ hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestClientBlockVerification.java Thu May  5 05:40:07 2011
@@ -30,8 +30,6 @@ import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.never;
 
-import static org.junit.Assert.*;
-
 public class TestClientBlockVerification {
 
   static BlockReaderTestUtil util = null;

Modified: hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestClientProtocolForPipelineRecovery.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestClientProtocolForPipelineRecovery.java?rev=1099687&r1=1099686&r2=1099687&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestClientProtocolForPipelineRecovery.java (original)
+++ hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestClientProtocolForPipelineRecovery.java Thu May  5 05:40:07 2011
@@ -23,7 +23,7 @@ import org.apache.hadoop.conf.Configurat
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hdfs.protocol.Block;
+import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
 import org.apache.hadoop.hdfs.server.namenode.LeaseExpiredException;
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.hdfs.HdfsConfiguration;
@@ -51,7 +51,7 @@ public class TestClientProtocolForPipeli
       Path file = new Path("dataprotocol.dat");    
       DFSTestUtil.createFile(fileSys, file, 1L, (short)numDataNodes, 0L);
       // get the first blockid for the file
-      Block firstBlock = DFSTestUtil.getFirstBlock(fileSys, file);
+      ExtendedBlock firstBlock = DFSTestUtil.getFirstBlock(fileSys, file);
 
       // test getNewStampAndToken on a finalized block
       try {
@@ -64,8 +64,8 @@ public class TestClientProtocolForPipeli
       // test getNewStampAndToken on a non-existent block
       try {
         long newBlockId = firstBlock.getBlockId() + 1;
-        Block newBlock = new Block(newBlockId, 0, 
-            firstBlock.getGenerationStamp());
+        ExtendedBlock newBlock = new ExtendedBlock(firstBlock.getBlockPoolId(),
+            newBlockId, 0, firstBlock.getGenerationStamp());
         namenode.updateBlockForPipeline(newBlock, "");
         Assert.fail("Cannot get a new GS from a non-existent block");
       } catch (IOException e) {

Modified: hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestCrcCorruption.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestCrcCorruption.java?rev=1099687&r1=1099686&r2=1099687&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestCrcCorruption.java (original)
+++ hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestCrcCorruption.java Thu May  5 05:40:07 2011
@@ -31,6 +31,7 @@ import static org.junit.Assert.*;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
 import org.apache.hadoop.io.IOUtils;
 
 /**
@@ -82,8 +83,9 @@ public class TestCrcCorruption {
       // file disallows this Datanode to send data to another datanode.
       // However, a client is alowed access to this block.
       //
-      File data_dir = new File(System.getProperty("test.build.data"),
-                               "dfs/data/data1" + MiniDFSCluster.FINALIZED_DIR_NAME);
+      File storageDir = MiniDFSCluster.getStorageDir(0, 1);
+      String bpid = cluster.getNamesystem().getBlockPoolId();
+      File data_dir = MiniDFSCluster.getFinalizedDir(storageDir, bpid);
       assertTrue("data directory does not exist", data_dir.exists());
       File[] blocks = data_dir.listFiles();
       assertTrue("Blocks do not exist in data-dir", (blocks != null) && (blocks.length > 0));
@@ -135,12 +137,13 @@ public class TestCrcCorruption {
           }
         }
       }
+      
       //
       // Now deliberately corrupt all meta blocks from the second
       // directory of the first datanode
       //
-      data_dir = new File(System.getProperty("test.build.data"),
-                               "dfs/data/data2" + MiniDFSCluster.FINALIZED_DIR_NAME);
+      storageDir = MiniDFSCluster.getStorageDir(0, 1);
+      data_dir = MiniDFSCluster.getFinalizedDir(storageDir, bpid);
       assertTrue("data directory does not exist", data_dir.exists());
       blocks = data_dir.listFiles();
       assertTrue("Blocks do not exist in data-dir", (blocks != null) && (blocks.length > 0));
@@ -256,7 +259,7 @@ public class TestCrcCorruption {
       DFSTestUtil.createFile(fs, file, fileSize, replFactor, 12345L /*seed*/);
       DFSTestUtil.waitReplication(fs, file, replFactor);
 
-      String block = DFSTestUtil.getFirstBlock(fs, file).getBlockName();
+      ExtendedBlock block = DFSTestUtil.getFirstBlock(fs, file);
       int blockFilesCorrupted = cluster.corruptBlockOnDataNodes(block);
       assertEquals("All replicas not corrupted", replFactor, blockFilesCorrupted);
 

Modified: hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestDFSClientRetries.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestDFSClientRetries.java?rev=1099687&r1=1099686&r2=1099687&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestDFSClientRetries.java (original)
+++ hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestDFSClientRetries.java Thu May  5 05:40:07 2011
@@ -43,9 +43,9 @@ import org.apache.hadoop.fs.FileChecksum
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.hdfs.protocol.Block;
 import org.apache.hadoop.hdfs.protocol.DatanodeID;
 import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
+import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
 import org.apache.hadoop.hdfs.protocol.LocatedBlock;
 import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
@@ -157,7 +157,7 @@ public class TestDFSClientRetries extend
     };
     when(mockNN.addBlock(anyString(), 
                          anyString(),
-                         any(Block.class),
+                         any(ExtendedBlock.class),
                          any(DatanodeInfo[].class))).thenAnswer(answer);
 
     final DFSClient client = new DFSClient(null, mockNN, conf, null);
@@ -489,7 +489,7 @@ public class TestDFSClientRetries extend
     
     public void run() {
       try {
-        fs = cluster.getNewFileSystemInstance();
+        fs = cluster.getNewFileSystemInstance(0);
         
         int bufferSize = len;
         byte[] buf = new byte[bufferSize];

Modified: hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestDFSFinalize.java
URL: http://svn.apache.org/viewvc/hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestDFSFinalize.java?rev=1099687&r1=1099686&r2=1099687&view=diff
==============================================================================
--- hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestDFSFinalize.java (original)
+++ hadoop/hdfs/branches/HDFS-1073/src/test/hdfs/org/apache/hadoop/hdfs/TestDFSFinalize.java Thu May  5 05:40:07 2011
@@ -69,7 +69,7 @@ public class TestDFSFinalize extends Tes
       assertEquals(
                    UpgradeUtilities.checksumContents(
                                                      DATA_NODE, new File(dataNodeDirs[i],"current")),
-                   UpgradeUtilities.checksumMasterContents(DATA_NODE));
+                   UpgradeUtilities.checksumMasterDataNodeContents());
     }
     for (int i = 0; i < nameNodeDirs.length; i++) {
       assertFalse(new File(nameNodeDirs[i],"previous").isDirectory());
@@ -95,10 +95,10 @@ public class TestDFSFinalize extends Tes
       String[] dataNodeDirs = conf.getStrings(DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY);
       
       log("Finalize with existing previous dir", numDirs);
-      UpgradeUtilities.createStorageDirs(NAME_NODE, nameNodeDirs, "current");
-      UpgradeUtilities.createStorageDirs(NAME_NODE, nameNodeDirs, "previous");
-      UpgradeUtilities.createStorageDirs(DATA_NODE, dataNodeDirs, "current");
-      UpgradeUtilities.createStorageDirs(DATA_NODE, dataNodeDirs, "previous");
+      UpgradeUtilities.createNameNodeStorageDirs(nameNodeDirs, "current");
+      UpgradeUtilities.createNameNodeStorageDirs(nameNodeDirs, "previous");
+      UpgradeUtilities.createDataNodeStorageDirs(dataNodeDirs, "current");
+      UpgradeUtilities.createDataNodeStorageDirs(dataNodeDirs, "previous");
       cluster = new MiniDFSCluster.Builder(conf)
                                   .format(false)
                                   .manageDataDfsDirs(false)



Mime
View raw message