src/contrib/raid/src/test/org/apache/hadoop/hdfs/server/namenode/TestBlockPlacementPolicyRaid.java - hadoop-mapreduce - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.hadoop.hdfs.server.namenode;

 import java.io.IOException;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;

 import junit.framework.Assert;

 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.DFSTestUtil;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
 import org.apache.hadoop.hdfs.protocol.LocatedBlock;
 import org.apache.hadoop.hdfs.server.namenode.BlockPlacementPolicyRaid.CachedFullPathNames;
 import org.apache.hadoop.hdfs.server.namenode.BlockPlacementPolicyRaid.CachedLocatedBlocks;
 import org.apache.hadoop.hdfs.server.namenode.BlockPlacementPolicyRaid.FileType;
 import org.apache.hadoop.raid.RaidNode;
 import org.junit.Test;

 public class TestBlockPlacementPolicyRaid {
   private Configuration conf = null;
   private MiniDFSCluster cluster = null;
   private FSNamesystem namesystem = null;
   private BlockPlacementPolicyRaid policy = null;
   private FileSystem fs = null;
   String[] rack1 = {"/rack1"};
   String[] rack2 = {"/rack2"};
   String[] host1 = {"host1.rack1.com"};
   String[] host2 = {"host2.rack2.com"};
   String xorPrefix = null;
   String raidTempPrefix = null;
   String raidrsTempPrefix = null;
   String raidrsHarTempPrefix = null;

   final static Log LOG =
       LogFactory.getLog(TestBlockPlacementPolicyRaid.class);

   protected void setupCluster() throws IOException {
     conf = new Configuration();
     conf.setLong(DFSConfigKeys.DFS_BLOCKREPORT_INTERVAL_MSEC_KEY, 1000L);
     conf.set("dfs.replication.pending.timeout.sec", "2");
     conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, 1L);
     conf.set("dfs.block.replicator.classname",
              "org.apache.hadoop.hdfs.server.namenode.BlockPlacementPolicyRaid");
     conf.set(RaidNode.STRIPE_LENGTH_KEY, "2");
     conf.set(RaidNode.RS_PARITY_LENGTH_KEY, "3");
     conf.setInt(DFSConfigKeys.DFS_BYTES_PER_CHECKSUM_KEY, 1);
     // start the cluster with one datanode first
     cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).
         format(true).racks(rack1).hosts(host1).build();
     cluster.waitActive();
     namesystem = cluster.getNameNode().getNamesystem();
     Assert.assertTrue("BlockPlacementPolicy type is not correct.",
       namesystem.blockManager.replicator instanceof BlockPlacementPolicyRaid);
     policy = (BlockPlacementPolicyRaid) namesystem.blockManager.replicator;
     fs = cluster.getFileSystem();
     xorPrefix = RaidNode.xorDestinationPath(conf).toUri().getPath();
     raidTempPrefix = RaidNode.xorTempPrefix(conf);
     raidrsTempPrefix = RaidNode.rsTempPrefix(conf);
     raidrsHarTempPrefix = RaidNode.rsHarTempPrefix(conf);
   }

   /**
    * Test that the parity files will be placed at the good locations when we
    * create them.
    */
   @Test
   public void testChooseTargetForRaidFile() throws IOException {
     setupCluster();
     try {
       String src = "/dir/file";
       String parity = raidrsTempPrefix + src;
       DFSTestUtil.createFile(fs, new Path(src), 4, (short)1, 0L);
       DFSTestUtil.waitReplication(fs, new Path(src), (short)1);
       refreshPolicy();
       setBlockPlacementPolicy(namesystem, policy);
       // start 3 more datanodes
       String[] racks = {"/rack2", "/rack2", "/rack2",
                         "/rack2", "/rack2", "/rack2"};
       String[] hosts =
         {"host2.rack2.com", "host3.rack2.com", "host4.rack2.com",
          "host5.rack2.com", "host6.rack2.com", "host7.rack2.com"};
       cluster.startDataNodes(conf, 6, true, null, racks, hosts, null);
       int numBlocks = 6;
       DFSTestUtil.createFile(fs, new Path(parity), numBlocks, (short)2, 0L);
       DFSTestUtil.waitReplication(fs, new Path(parity), (short)2);
       FileStatus srcStat = fs.getFileStatus(new Path(src));
       BlockLocation[] srcLoc =
         fs.getFileBlockLocations(srcStat, 0, srcStat.getLen());
       FileStatus parityStat = fs.getFileStatus(new Path(parity));
       BlockLocation[] parityLoc =
           fs.getFileBlockLocations(parityStat, 0, parityStat.getLen());
       int parityLen = RaidNode.rsParityLength(conf);
       for (int i = 0; i < numBlocks / parityLen; i++) {
         Set<String> locations = new HashSet<String>();
         for (int j = 0; j < srcLoc.length; j++) {
           String [] names = srcLoc[j].getNames();
           for (int k = 0; k < names.length; k++) {
             LOG.info("Source block location: " + names[k]);
             locations.add(names[k]);
           }
         }
         for (int j = 0 ; j < parityLen; j++) {
           String[] names = parityLoc[j + i * parityLen].getNames();
           for (int k = 0; k < names.length; k++) {
             LOG.info("Parity block location: " + names[k]);
             Assert.assertTrue(locations.add(names[k]));
           }
         }
       }
     } finally {
       if (cluster != null) {
         cluster.shutdown();
       }
     }
   }

   /**
    * Test that the har parity files will be placed at the good locations when we
    * create them.
    */
   @Test
   public void testChooseTargetForHarRaidFile() throws IOException {
     setupCluster();
     try {
       String[] racks = {"/rack2", "/rack2", "/rack2",
                         "/rack2", "/rack2", "/rack2"};
       String[] hosts =
         {"host2.rack2.com", "host3.rack2.com", "host4.rack2.com",
          "host5.rack2.com", "host6.rack2.com", "host7.rack2.com"};
       cluster.startDataNodes(conf, 6, true, null, racks, hosts, null);
       String harParity = raidrsHarTempPrefix + "/dir/file";
       int numBlocks = 11;
       DFSTestUtil.createFile(fs, new Path(harParity), numBlocks, (short)1, 0L);
       DFSTestUtil.waitReplication(fs, new Path(harParity), (short)1);
       FileStatus stat = fs.getFileStatus(new Path(harParity));
       BlockLocation[] loc = fs.getFileBlockLocations(stat, 0, stat.getLen());
       int rsParityLength = RaidNode.rsParityLength(conf);
       for (int i = 0; i < numBlocks - rsParityLength; i++) {
         Set<String> locations = new HashSet<String>();
         for (int j = 0; j < rsParityLength; j++) {
           for (int k = 0; k < loc[i + j].getNames().length; k++) {
             // verify that every adjacent 4 blocks are on differnt nodes
             String name = loc[i + j].getNames()[k];
             LOG.info("Har Raid block location: " + name);
             Assert.assertTrue(locations.add(name));
           }
         }
       }
     } finally {
       if (cluster != null) {
         cluster.shutdown();
       }
     }
   }

   /**
    * Test BlockPlacementPolicyRaid.CachedLocatedBlocks
    * Verify that the results obtained from cache is the same as
    * the results obtained directly
    */
   @Test
   public void testCachedBlocks() throws IOException {
     setupCluster();
     try {
       String file1 = "/dir/file1";
       String file2 = "/dir/file2";
       DFSTestUtil.createFile(fs, new Path(file1), 3, (short)1, 0L);
       DFSTestUtil.createFile(fs, new Path(file2), 4, (short)1, 0L);
       // test blocks cache
       CachedLocatedBlocks cachedBlocks = new CachedLocatedBlocks(namesystem);
       verifyCachedBlocksResult(cachedBlocks, namesystem, file1);
       verifyCachedBlocksResult(cachedBlocks, namesystem, file1);
       verifyCachedBlocksResult(cachedBlocks, namesystem, file2);
       verifyCachedBlocksResult(cachedBlocks, namesystem, file2);
       try {
         Thread.sleep(1200L);
       } catch (InterruptedException e) {
       }
       verifyCachedBlocksResult(cachedBlocks, namesystem, file2);
       verifyCachedBlocksResult(cachedBlocks, namesystem, file1);
     } finally {
       if (cluster != null) {
         cluster.shutdown();
       }
     }
   }

   /**
    * Test BlockPlacementPolicyRaid.CachedFullPathNames
    * Verify that the results obtained from cache is the same as
    * the results obtained directly
    */
   @Test
   public void testCachedPathNames() throws IOException {
     setupCluster();
     try {
       String file1 = "/dir/file1";
       String file2 = "/dir/file2";
       DFSTestUtil.createFile(fs, new Path(file1), 3, (short)1, 0L);
       DFSTestUtil.createFile(fs, new Path(file2), 4, (short)1, 0L);
       // test full path cache
       CachedFullPathNames cachedFullPathNames =
           new CachedFullPathNames(namesystem);
       FSInodeInfo inode1 = null;
       FSInodeInfo inode2 = null;
       namesystem.dir.readLock();
       try {
         inode1 = namesystem.dir.rootDir.getNode(file1, true);
         inode2 = namesystem.dir.rootDir.getNode(file2, true);
       } finally {
         namesystem.dir.readUnlock();
       }
       verifyCachedFullPathNameResult(cachedFullPathNames, inode1);
       verifyCachedFullPathNameResult(cachedFullPathNames, inode1);
       verifyCachedFullPathNameResult(cachedFullPathNames, inode2);
       verifyCachedFullPathNameResult(cachedFullPathNames, inode2);
       try {
         Thread.sleep(1200L);
       } catch (InterruptedException e) {
       }
       verifyCachedFullPathNameResult(cachedFullPathNames, inode2);
       verifyCachedFullPathNameResult(cachedFullPathNames, inode1);
     } finally {
       if (cluster != null) {
         cluster.shutdown();
       }
     }
   }
   /**
    * Test the result of getCompanionBlocks() on the unraided files
    */
   @Test
   public void testGetCompanionBLocks() throws IOException {
     setupCluster();
     try {
       String file1 = "/dir/file1";
       String file2 = "/raid/dir/file2";
       String file3 = "/raidrs/dir/file3";
       // Set the policy to default policy to place the block in the default way
       setBlockPlacementPolicy(namesystem, new BlockPlacementPolicyDefault(
           conf, namesystem, namesystem.clusterMap));
       DFSTestUtil.createFile(fs, new Path(file1), 3, (short)1, 0L);
       DFSTestUtil.createFile(fs, new Path(file2), 4, (short)1, 0L);
       DFSTestUtil.createFile(fs, new Path(file3), 8, (short)1, 0L);
       Collection<LocatedBlock> companionBlocks;

       companionBlocks = getCompanionBlocks(
           namesystem, policy, getBlocks(namesystem, file1).get(0).getBlock());
       Assert.assertTrue(companionBlocks == null || companionBlocks.size() == 0);

       companionBlocks = getCompanionBlocks(
           namesystem, policy, getBlocks(namesystem, file1).get(2).getBlock());
       Assert.assertTrue(companionBlocks == null || companionBlocks.size() == 0);

       companionBlocks = getCompanionBlocks(
           namesystem, policy, getBlocks(namesystem, file2).get(0).getBlock());
       Assert.assertEquals(1, companionBlocks.size());

       companionBlocks = getCompanionBlocks(
           namesystem, policy, getBlocks(namesystem, file2).get(3).getBlock());
       Assert.assertEquals(1, companionBlocks.size());

       int rsParityLength = RaidNode.rsParityLength(conf);
       companionBlocks = getCompanionBlocks(
           namesystem, policy, getBlocks(namesystem, file3).get(0).getBlock());
       Assert.assertEquals(rsParityLength, companionBlocks.size());

       companionBlocks = getCompanionBlocks(
           namesystem, policy, getBlocks(namesystem, file3).get(4).getBlock());
       Assert.assertEquals(rsParityLength, companionBlocks.size());

       companionBlocks = getCompanionBlocks(
           namesystem, policy, getBlocks(namesystem, file3).get(6).getBlock());
       Assert.assertEquals(2, companionBlocks.size());
     } finally {
       if (cluster != null) {
         cluster.shutdown();
       }
     }
   }

   static void setBlockPlacementPolicy(
       FSNamesystem namesystem, BlockPlacementPolicy policy) {
     namesystem.writeLock();
     try {
       namesystem.blockManager.replicator = policy;
     } finally {
       namesystem.writeUnlock();
     }
   }

   /**
    * Test BlockPlacementPolicyRaid actually deletes the correct replica.
    * Start 2 datanodes and create 1 source file and its parity file.
    * 1) Start host1, create the parity file with replication 1
    * 2) Start host2, create the source file with replication 2
    * 3) Set repliation of source file to 1
    * Verify that the policy should delete the block with more companion blocks.
    */
   @Test
   public void testDeleteReplica() throws IOException {
     setupCluster();
     try {
       // Set the policy to default policy to place the block in the default way
       setBlockPlacementPolicy(namesystem, new BlockPlacementPolicyDefault(
           conf, namesystem, namesystem.clusterMap));
       DatanodeDescriptor datanode1 =
         namesystem.datanodeMap.values().iterator().next();
       String source = "/dir/file";
       String parity = xorPrefix + source;

       final Path parityPath = new Path(parity);
       DFSTestUtil.createFile(fs, parityPath, 3, (short)1, 0L);
       DFSTestUtil.waitReplication(fs, parityPath, (short)1);

       // start one more datanode
       cluster.startDataNodes(conf, 1, true, null, rack2, host2, null);
       DatanodeDescriptor datanode2 = null;
       for (DatanodeDescriptor d : namesystem.datanodeMap.values()) {
         if (!d.getName().equals(datanode1.getName())) {
           datanode2 = d;
         }
       }
       Assert.assertTrue(datanode2 != null);
       cluster.waitActive();
       final Path sourcePath = new Path(source);
       DFSTestUtil.createFile(fs, sourcePath, 5, (short)2, 0L);
       DFSTestUtil.waitReplication(fs, sourcePath, (short)2);

       refreshPolicy();
       Assert.assertEquals(parity,
                           policy.getParityFile(source));
       Assert.assertEquals(source,
                           policy.getSourceFile(parity, xorPrefix));

       List<LocatedBlock> sourceBlocks = getBlocks(namesystem, source);
       List<LocatedBlock> parityBlocks = getBlocks(namesystem, parity);
       Assert.assertEquals(5, sourceBlocks.size());
       Assert.assertEquals(3, parityBlocks.size());

       // verify the result of getCompanionBlocks()
       Collection<LocatedBlock> companionBlocks;
       companionBlocks = getCompanionBlocks(
           namesystem, policy, sourceBlocks.get(0).getBlock());
       verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
                             new int[]{0, 1}, new int[]{0});

       companionBlocks = getCompanionBlocks(
           namesystem, policy, sourceBlocks.get(1).getBlock());
       verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
                             new int[]{0, 1}, new int[]{0});

       companionBlocks = getCompanionBlocks(
           namesystem, policy, sourceBlocks.get(2).getBlock());
       verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
                             new int[]{2, 3}, new int[]{1});

       companionBlocks = getCompanionBlocks(
           namesystem, policy, sourceBlocks.get(3).getBlock());
       verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
                             new int[]{2, 3}, new int[]{1});

       companionBlocks = getCompanionBlocks(
           namesystem, policy, sourceBlocks.get(4).getBlock());
       verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
                             new int[]{4}, new int[]{2});

       companionBlocks = getCompanionBlocks(
           namesystem, policy, parityBlocks.get(0).getBlock());
       verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
                             new int[]{0, 1}, new int[]{0});

       companionBlocks = getCompanionBlocks(
           namesystem, policy, parityBlocks.get(1).getBlock());
       verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
                             new int[]{2, 3}, new int[]{1});

       companionBlocks = getCompanionBlocks(
           namesystem, policy, parityBlocks.get(2).getBlock());
       verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
                             new int[]{4}, new int[]{2});

       // Set the policy back to raid policy. We have to create a new object
       // here to clear the block location cache
       refreshPolicy();
       setBlockPlacementPolicy(namesystem, policy);
       // verify policy deletes the correct blocks. companion blocks should be
       // evenly distributed.
       fs.setReplication(sourcePath, (short)1);
       DFSTestUtil.waitReplication(fs, sourcePath, (short)1);
       Map<String, Integer> counters = new HashMap<String, Integer>();
       refreshPolicy();
       for (int i = 0; i < parityBlocks.size(); i++) {
         companionBlocks = getCompanionBlocks(
             namesystem, policy, parityBlocks.get(i).getBlock());

         counters = BlockPlacementPolicyRaid.countCompanionBlocks(
             companionBlocks, false);
         Assert.assertTrue(counters.get(datanode1.getName()) >= 1 &&
                           counters.get(datanode1.getName()) <= 2);
         Assert.assertTrue(counters.get(datanode1.getName()) +
                           counters.get(datanode2.getName()) ==
                           companionBlocks.size());

         counters = BlockPlacementPolicyRaid.countCompanionBlocks(
             companionBlocks, true);
         Assert.assertTrue(counters.get(datanode1.getParent().getName()) >= 1 &&
                           counters.get(datanode1.getParent().getName()) <= 2);
         Assert.assertTrue(counters.get(datanode1.getParent().getName()) +
                           counters.get(datanode2.getParent().getName()) ==
                           companionBlocks.size());
       }
     } finally {
       if (cluster != null) {
         cluster.shutdown();
       }
     }
   }

   // create a new BlockPlacementPolicyRaid to clear the cache
   private void refreshPolicy() {
       policy = new BlockPlacementPolicyRaid();
       policy.initialize(conf, namesystem, namesystem.clusterMap);
   }

   private void verifyCompanionBlocks(Collection<LocatedBlock> companionBlocks,
       List<LocatedBlock> sourceBlocks, List<LocatedBlock> parityBlocks,
       int[] sourceBlockIndexes, int[] parityBlockIndexes) {
     Set<ExtendedBlock> blockSet = new HashSet<ExtendedBlock>();
     for (LocatedBlock b : companionBlocks) {
       blockSet.add(b.getBlock());
     }
     Assert.assertEquals(sourceBlockIndexes.length + parityBlockIndexes.length,
                         blockSet.size());
     for (int index : sourceBlockIndexes) {
       Assert.assertTrue(blockSet.contains(sourceBlocks.get(index).getBlock()));
     }
     for (int index : parityBlockIndexes) {
       Assert.assertTrue(blockSet.contains(parityBlocks.get(index).getBlock()));
     }
   }

   private void verifyCachedFullPathNameResult(
       CachedFullPathNames cachedFullPathNames, FSInodeInfo inode)
   throws IOException {
     String res1 = inode.getFullPathName();
     String res2 = cachedFullPathNames.get(inode);
     LOG.info("Actual path name: " + res1);
     LOG.info("Cached path name: " + res2);
     Assert.assertEquals(cachedFullPathNames.get(inode),
                         inode.getFullPathName());
   }

   private void verifyCachedBlocksResult(CachedLocatedBlocks cachedBlocks,
       FSNamesystem namesystem, String file) throws IOException{
     long len = namesystem.getFileInfo(file, true).getLen();
     List<LocatedBlock> res1 = namesystem.getBlockLocations(
         file, 0L, len, false, false).getLocatedBlocks();
     List<LocatedBlock> res2 = cachedBlocks.get(file);
     for (int i = 0; i < res1.size(); i++) {
       LOG.info("Actual block: " + res1.get(i).getBlock());
       LOG.info("Cached block: " + res2.get(i).getBlock());
       Assert.assertEquals(res1.get(i).getBlock(), res2.get(i).getBlock());
     }
   }

   private Collection<LocatedBlock> getCompanionBlocks(
       FSNamesystem namesystem, BlockPlacementPolicyRaid policy,
       ExtendedBlock block) throws IOException {
     INodeFile inode = namesystem.blockManager.blocksMap.getINode(block
         .getLocalBlock());
     FileType type = policy.getFileType(inode.getFullPathName());
     return policy.getCompanionBlocks(inode.getFullPathName(), type,
         block.getLocalBlock());
   }

   private List<LocatedBlock> getBlocks(FSNamesystem namesystem, String file)
       throws IOException {
     long len = namesystem.getFileInfo(file, true).getLen();
     return namesystem.getBlockLocations(
                file, 0, len, false, false).getLocatedBlocks();
   }
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.hadoop.hdfs.server.namenode;

	import java.io.IOException;
	import java.util.Collection;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;

	import junit.framework.Assert;

	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.fs.BlockLocation;
	import org.apache.hadoop.fs.FileStatus;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.hdfs.DFSConfigKeys;
	import org.apache.hadoop.hdfs.DFSTestUtil;
	import org.apache.hadoop.hdfs.MiniDFSCluster;
	import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
	import org.apache.hadoop.hdfs.protocol.LocatedBlock;
	import org.apache.hadoop.hdfs.server.namenode.BlockPlacementPolicyRaid.CachedFullPathNames;
	import org.apache.hadoop.hdfs.server.namenode.BlockPlacementPolicyRaid.CachedLocatedBlocks;
	import org.apache.hadoop.hdfs.server.namenode.BlockPlacementPolicyRaid.FileType;
	import org.apache.hadoop.raid.RaidNode;
	import org.junit.Test;

	public class TestBlockPlacementPolicyRaid {
	private Configuration conf = null;
	private MiniDFSCluster cluster = null;
	private FSNamesystem namesystem = null;
	private BlockPlacementPolicyRaid policy = null;
	private FileSystem fs = null;
	String[] rack1 = {"/rack1"};
	String[] rack2 = {"/rack2"};
	String[] host1 = {"host1.rack1.com"};
	String[] host2 = {"host2.rack2.com"};
	String xorPrefix = null;
	String raidTempPrefix = null;
	String raidrsTempPrefix = null;
	String raidrsHarTempPrefix = null;

	final static Log LOG =
	LogFactory.getLog(TestBlockPlacementPolicyRaid.class);

	protected void setupCluster() throws IOException {
	conf = new Configuration();
	conf.setLong(DFSConfigKeys.DFS_BLOCKREPORT_INTERVAL_MSEC_KEY, 1000L);
	conf.set("dfs.replication.pending.timeout.sec", "2");
	conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, 1L);
	conf.set("dfs.block.replicator.classname",
	"org.apache.hadoop.hdfs.server.namenode.BlockPlacementPolicyRaid");
	conf.set(RaidNode.STRIPE_LENGTH_KEY, "2");
	conf.set(RaidNode.RS_PARITY_LENGTH_KEY, "3");
	conf.setInt(DFSConfigKeys.DFS_BYTES_PER_CHECKSUM_KEY, 1);
	// start the cluster with one datanode first
	cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).
	format(true).racks(rack1).hosts(host1).build();
	cluster.waitActive();
	namesystem = cluster.getNameNode().getNamesystem();
	Assert.assertTrue("BlockPlacementPolicy type is not correct.",
	namesystem.blockManager.replicator instanceof BlockPlacementPolicyRaid);
	policy = (BlockPlacementPolicyRaid) namesystem.blockManager.replicator;
	fs = cluster.getFileSystem();
	xorPrefix = RaidNode.xorDestinationPath(conf).toUri().getPath();
	raidTempPrefix = RaidNode.xorTempPrefix(conf);
	raidrsTempPrefix = RaidNode.rsTempPrefix(conf);
	raidrsHarTempPrefix = RaidNode.rsHarTempPrefix(conf);
	}

	/**
	* Test that the parity files will be placed at the good locations when we
	* create them.
	*/
	@Test
	public void testChooseTargetForRaidFile() throws IOException {
	setupCluster();
	try {
	String src = "/dir/file";
	String parity = raidrsTempPrefix + src;
	DFSTestUtil.createFile(fs, new Path(src), 4, (short)1, 0L);
	DFSTestUtil.waitReplication(fs, new Path(src), (short)1);
	refreshPolicy();
	setBlockPlacementPolicy(namesystem, policy);
	// start 3 more datanodes
	String[] racks = {"/rack2", "/rack2", "/rack2",
	"/rack2", "/rack2", "/rack2"};
	String[] hosts =
	{"host2.rack2.com", "host3.rack2.com", "host4.rack2.com",
	"host5.rack2.com", "host6.rack2.com", "host7.rack2.com"};
	cluster.startDataNodes(conf, 6, true, null, racks, hosts, null);
	int numBlocks = 6;
	DFSTestUtil.createFile(fs, new Path(parity), numBlocks, (short)2, 0L);
	DFSTestUtil.waitReplication(fs, new Path(parity), (short)2);
	FileStatus srcStat = fs.getFileStatus(new Path(src));
	BlockLocation[] srcLoc =
	fs.getFileBlockLocations(srcStat, 0, srcStat.getLen());
	FileStatus parityStat = fs.getFileStatus(new Path(parity));
	BlockLocation[] parityLoc =
	fs.getFileBlockLocations(parityStat, 0, parityStat.getLen());
	int parityLen = RaidNode.rsParityLength(conf);
	for (int i = 0; i < numBlocks / parityLen; i++) {
	Set<String> locations = new HashSet<String>();
	for (int j = 0; j < srcLoc.length; j++) {
	String [] names = srcLoc[j].getNames();
	for (int k = 0; k < names.length; k++) {
	LOG.info("Source block location: " + names[k]);
	locations.add(names[k]);
	}
	}
	for (int j = 0 ; j < parityLen; j++) {
	String[] names = parityLoc[j + i * parityLen].getNames();
	for (int k = 0; k < names.length; k++) {
	LOG.info("Parity block location: " + names[k]);
	Assert.assertTrue(locations.add(names[k]));
	}
	}
	}
	} finally {
	if (cluster != null) {
	cluster.shutdown();
	}
	}
	}

	/**
	* Test that the har parity files will be placed at the good locations when we
	* create them.
	*/
	@Test
	public void testChooseTargetForHarRaidFile() throws IOException {
	setupCluster();
	try {
	String[] racks = {"/rack2", "/rack2", "/rack2",
	"/rack2", "/rack2", "/rack2"};
	String[] hosts =
	{"host2.rack2.com", "host3.rack2.com", "host4.rack2.com",
	"host5.rack2.com", "host6.rack2.com", "host7.rack2.com"};
	cluster.startDataNodes(conf, 6, true, null, racks, hosts, null);
	String harParity = raidrsHarTempPrefix + "/dir/file";
	int numBlocks = 11;
	DFSTestUtil.createFile(fs, new Path(harParity), numBlocks, (short)1, 0L);
	DFSTestUtil.waitReplication(fs, new Path(harParity), (short)1);
	FileStatus stat = fs.getFileStatus(new Path(harParity));
	BlockLocation[] loc = fs.getFileBlockLocations(stat, 0, stat.getLen());
	int rsParityLength = RaidNode.rsParityLength(conf);
	for (int i = 0; i < numBlocks - rsParityLength; i++) {
	Set<String> locations = new HashSet<String>();
	for (int j = 0; j < rsParityLength; j++) {
	for (int k = 0; k < loc[i + j].getNames().length; k++) {
	// verify that every adjacent 4 blocks are on differnt nodes
	String name = loc[i + j].getNames()[k];
	LOG.info("Har Raid block location: " + name);
	Assert.assertTrue(locations.add(name));
	}
	}
	}
	} finally {
	if (cluster != null) {
	cluster.shutdown();
	}
	}
	}

	/**
	* Test BlockPlacementPolicyRaid.CachedLocatedBlocks
	* Verify that the results obtained from cache is the same as
	* the results obtained directly
	*/
	@Test
	public void testCachedBlocks() throws IOException {
	setupCluster();
	try {
	String file1 = "/dir/file1";
	String file2 = "/dir/file2";
	DFSTestUtil.createFile(fs, new Path(file1), 3, (short)1, 0L);
	DFSTestUtil.createFile(fs, new Path(file2), 4, (short)1, 0L);
	// test blocks cache
	CachedLocatedBlocks cachedBlocks = new CachedLocatedBlocks(namesystem);
	verifyCachedBlocksResult(cachedBlocks, namesystem, file1);
	verifyCachedBlocksResult(cachedBlocks, namesystem, file1);
	verifyCachedBlocksResult(cachedBlocks, namesystem, file2);
	verifyCachedBlocksResult(cachedBlocks, namesystem, file2);
	try {
	Thread.sleep(1200L);
	} catch (InterruptedException e) {
	}
	verifyCachedBlocksResult(cachedBlocks, namesystem, file2);
	verifyCachedBlocksResult(cachedBlocks, namesystem, file1);
	} finally {
	if (cluster != null) {
	cluster.shutdown();
	}
	}
	}

	/**
	* Test BlockPlacementPolicyRaid.CachedFullPathNames
	* Verify that the results obtained from cache is the same as
	* the results obtained directly
	*/
	@Test
	public void testCachedPathNames() throws IOException {
	setupCluster();
	try {
	String file1 = "/dir/file1";
	String file2 = "/dir/file2";
	DFSTestUtil.createFile(fs, new Path(file1), 3, (short)1, 0L);
	DFSTestUtil.createFile(fs, new Path(file2), 4, (short)1, 0L);
	// test full path cache
	CachedFullPathNames cachedFullPathNames =
	new CachedFullPathNames(namesystem);
	FSInodeInfo inode1 = null;
	FSInodeInfo inode2 = null;
	namesystem.dir.readLock();
	try {
	inode1 = namesystem.dir.rootDir.getNode(file1, true);
	inode2 = namesystem.dir.rootDir.getNode(file2, true);
	} finally {
	namesystem.dir.readUnlock();
	}
	verifyCachedFullPathNameResult(cachedFullPathNames, inode1);
	verifyCachedFullPathNameResult(cachedFullPathNames, inode1);
	verifyCachedFullPathNameResult(cachedFullPathNames, inode2);
	verifyCachedFullPathNameResult(cachedFullPathNames, inode2);
	try {
	Thread.sleep(1200L);
	} catch (InterruptedException e) {
	}
	verifyCachedFullPathNameResult(cachedFullPathNames, inode2);
	verifyCachedFullPathNameResult(cachedFullPathNames, inode1);
	} finally {
	if (cluster != null) {
	cluster.shutdown();
	}
	}
	}
	/**
	* Test the result of getCompanionBlocks() on the unraided files
	*/
	@Test
	public void testGetCompanionBLocks() throws IOException {
	setupCluster();
	try {
	String file1 = "/dir/file1";
	String file2 = "/raid/dir/file2";
	String file3 = "/raidrs/dir/file3";
	// Set the policy to default policy to place the block in the default way
	setBlockPlacementPolicy(namesystem, new BlockPlacementPolicyDefault(
	conf, namesystem, namesystem.clusterMap));
	DFSTestUtil.createFile(fs, new Path(file1), 3, (short)1, 0L);
	DFSTestUtil.createFile(fs, new Path(file2), 4, (short)1, 0L);
	DFSTestUtil.createFile(fs, new Path(file3), 8, (short)1, 0L);
	Collection<LocatedBlock> companionBlocks;

	companionBlocks = getCompanionBlocks(
	namesystem, policy, getBlocks(namesystem, file1).get(0).getBlock());
	Assert.assertTrue(companionBlocks == null \|\| companionBlocks.size() == 0);

	companionBlocks = getCompanionBlocks(
	namesystem, policy, getBlocks(namesystem, file1).get(2).getBlock());
	Assert.assertTrue(companionBlocks == null \|\| companionBlocks.size() == 0);

	companionBlocks = getCompanionBlocks(
	namesystem, policy, getBlocks(namesystem, file2).get(0).getBlock());
	Assert.assertEquals(1, companionBlocks.size());

	companionBlocks = getCompanionBlocks(
	namesystem, policy, getBlocks(namesystem, file2).get(3).getBlock());
	Assert.assertEquals(1, companionBlocks.size());

	int rsParityLength = RaidNode.rsParityLength(conf);
	companionBlocks = getCompanionBlocks(
	namesystem, policy, getBlocks(namesystem, file3).get(0).getBlock());
	Assert.assertEquals(rsParityLength, companionBlocks.size());

	companionBlocks = getCompanionBlocks(
	namesystem, policy, getBlocks(namesystem, file3).get(4).getBlock());
	Assert.assertEquals(rsParityLength, companionBlocks.size());

	companionBlocks = getCompanionBlocks(
	namesystem, policy, getBlocks(namesystem, file3).get(6).getBlock());
	Assert.assertEquals(2, companionBlocks.size());
	} finally {
	if (cluster != null) {
	cluster.shutdown();
	}
	}
	}

	static void setBlockPlacementPolicy(
	FSNamesystem namesystem, BlockPlacementPolicy policy) {
	namesystem.writeLock();
	try {
	namesystem.blockManager.replicator = policy;
	} finally {
	namesystem.writeUnlock();
	}
	}

	/**
	* Test BlockPlacementPolicyRaid actually deletes the correct replica.
	* Start 2 datanodes and create 1 source file and its parity file.
	* 1) Start host1, create the parity file with replication 1
	* 2) Start host2, create the source file with replication 2
	* 3) Set repliation of source file to 1
	* Verify that the policy should delete the block with more companion blocks.
	*/
	@Test
	public void testDeleteReplica() throws IOException {
	setupCluster();
	try {
	// Set the policy to default policy to place the block in the default way
	setBlockPlacementPolicy(namesystem, new BlockPlacementPolicyDefault(
	conf, namesystem, namesystem.clusterMap));
	DatanodeDescriptor datanode1 =
	namesystem.datanodeMap.values().iterator().next();
	String source = "/dir/file";
	String parity = xorPrefix + source;

	final Path parityPath = new Path(parity);
	DFSTestUtil.createFile(fs, parityPath, 3, (short)1, 0L);
	DFSTestUtil.waitReplication(fs, parityPath, (short)1);

	// start one more datanode
	cluster.startDataNodes(conf, 1, true, null, rack2, host2, null);
	DatanodeDescriptor datanode2 = null;
	for (DatanodeDescriptor d : namesystem.datanodeMap.values()) {
	if (!d.getName().equals(datanode1.getName())) {
	datanode2 = d;
	}
	}
	Assert.assertTrue(datanode2 != null);
	cluster.waitActive();
	final Path sourcePath = new Path(source);
	DFSTestUtil.createFile(fs, sourcePath, 5, (short)2, 0L);
	DFSTestUtil.waitReplication(fs, sourcePath, (short)2);

	refreshPolicy();
	Assert.assertEquals(parity,
	policy.getParityFile(source));
	Assert.assertEquals(source,
	policy.getSourceFile(parity, xorPrefix));

	List<LocatedBlock> sourceBlocks = getBlocks(namesystem, source);
	List<LocatedBlock> parityBlocks = getBlocks(namesystem, parity);
	Assert.assertEquals(5, sourceBlocks.size());
	Assert.assertEquals(3, parityBlocks.size());

	// verify the result of getCompanionBlocks()
	Collection<LocatedBlock> companionBlocks;
	companionBlocks = getCompanionBlocks(
	namesystem, policy, sourceBlocks.get(0).getBlock());
	verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
	new int[]{0, 1}, new int[]{0});

	companionBlocks = getCompanionBlocks(
	namesystem, policy, sourceBlocks.get(1).getBlock());
	verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
	new int[]{0, 1}, new int[]{0});

	companionBlocks = getCompanionBlocks(
	namesystem, policy, sourceBlocks.get(2).getBlock());
	verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
	new int[]{2, 3}, new int[]{1});

	companionBlocks = getCompanionBlocks(
	namesystem, policy, sourceBlocks.get(3).getBlock());
	verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
	new int[]{2, 3}, new int[]{1});

	companionBlocks = getCompanionBlocks(
	namesystem, policy, sourceBlocks.get(4).getBlock());
	verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
	new int[]{4}, new int[]{2});

	companionBlocks = getCompanionBlocks(
	namesystem, policy, parityBlocks.get(0).getBlock());
	verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
	new int[]{0, 1}, new int[]{0});

	companionBlocks = getCompanionBlocks(
	namesystem, policy, parityBlocks.get(1).getBlock());
	verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
	new int[]{2, 3}, new int[]{1});

	companionBlocks = getCompanionBlocks(
	namesystem, policy, parityBlocks.get(2).getBlock());
	verifyCompanionBlocks(companionBlocks, sourceBlocks, parityBlocks,
	new int[]{4}, new int[]{2});

	// Set the policy back to raid policy. We have to create a new object
	// here to clear the block location cache
	refreshPolicy();
	setBlockPlacementPolicy(namesystem, policy);
	// verify policy deletes the correct blocks. companion blocks should be
	// evenly distributed.
	fs.setReplication(sourcePath, (short)1);
	DFSTestUtil.waitReplication(fs, sourcePath, (short)1);
	Map<String, Integer> counters = new HashMap<String, Integer>();
	refreshPolicy();
	for (int i = 0; i < parityBlocks.size(); i++) {
	companionBlocks = getCompanionBlocks(
	namesystem, policy, parityBlocks.get(i).getBlock());

	counters = BlockPlacementPolicyRaid.countCompanionBlocks(
	companionBlocks, false);
	Assert.assertTrue(counters.get(datanode1.getName()) >= 1 &&
	counters.get(datanode1.getName()) <= 2);
	Assert.assertTrue(counters.get(datanode1.getName()) +
	counters.get(datanode2.getName()) ==
	companionBlocks.size());

	counters = BlockPlacementPolicyRaid.countCompanionBlocks(
	companionBlocks, true);
	Assert.assertTrue(counters.get(datanode1.getParent().getName()) >= 1 &&
	counters.get(datanode1.getParent().getName()) <= 2);
	Assert.assertTrue(counters.get(datanode1.getParent().getName()) +
	counters.get(datanode2.getParent().getName()) ==
	companionBlocks.size());
	}
	} finally {
	if (cluster != null) {
	cluster.shutdown();
	}
	}
	}

	// create a new BlockPlacementPolicyRaid to clear the cache
	private void refreshPolicy() {
	policy = new BlockPlacementPolicyRaid();
	policy.initialize(conf, namesystem, namesystem.clusterMap);
	}

	private void verifyCompanionBlocks(Collection<LocatedBlock> companionBlocks,
	List<LocatedBlock> sourceBlocks, List<LocatedBlock> parityBlocks,
	int[] sourceBlockIndexes, int[] parityBlockIndexes) {
	Set<ExtendedBlock> blockSet = new HashSet<ExtendedBlock>();
	for (LocatedBlock b : companionBlocks) {
	blockSet.add(b.getBlock());
	}
	Assert.assertEquals(sourceBlockIndexes.length + parityBlockIndexes.length,
	blockSet.size());
	for (int index : sourceBlockIndexes) {
	Assert.assertTrue(blockSet.contains(sourceBlocks.get(index).getBlock()));
	}
	for (int index : parityBlockIndexes) {
	Assert.assertTrue(blockSet.contains(parityBlocks.get(index).getBlock()));
	}
	}

	private void verifyCachedFullPathNameResult(
	CachedFullPathNames cachedFullPathNames, FSInodeInfo inode)
	throws IOException {
	String res1 = inode.getFullPathName();
	String res2 = cachedFullPathNames.get(inode);
	LOG.info("Actual path name: " + res1);
	LOG.info("Cached path name: " + res2);
	Assert.assertEquals(cachedFullPathNames.get(inode),
	inode.getFullPathName());
	}

	private void verifyCachedBlocksResult(CachedLocatedBlocks cachedBlocks,
	FSNamesystem namesystem, String file) throws IOException{
	long len = namesystem.getFileInfo(file, true).getLen();
	List<LocatedBlock> res1 = namesystem.getBlockLocations(
	file, 0L, len, false, false).getLocatedBlocks();
	List<LocatedBlock> res2 = cachedBlocks.get(file);
	for (int i = 0; i < res1.size(); i++) {
	LOG.info("Actual block: " + res1.get(i).getBlock());
	LOG.info("Cached block: " + res2.get(i).getBlock());
	Assert.assertEquals(res1.get(i).getBlock(), res2.get(i).getBlock());
	}
	}

	private Collection<LocatedBlock> getCompanionBlocks(
	FSNamesystem namesystem, BlockPlacementPolicyRaid policy,
	ExtendedBlock block) throws IOException {
	INodeFile inode = namesystem.blockManager.blocksMap.getINode(block
	.getLocalBlock());
	FileType type = policy.getFileType(inode.getFullPathName());
	return policy.getCompanionBlocks(inode.getFullPathName(), type,
	block.getLocalBlock());
	}

	private List<LocatedBlock> getBlocks(FSNamesystem namesystem, String file)
	throws IOException {
	long len = namesystem.getFileInfo(file, true).getLen();
	return namesystem.getBlockLocations(
	file, 0, len, false, false).getLocatedBlocks();
	}
	}