@Test(timeout=60000) public void testNumberOfBlocksToBeReplicated() throws Exception { Configuration conf = new HdfsConfiguration(); conf.setLong(DFSConfigKeys.DFS_NAMENODE_MIN_BLOCK_SIZE_KEY, 0); conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, 1); conf.setInt(DFSConfigKeys.DFS_BYTES_PER_CHECKSUM_KEY, 1); conf.setInt(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, 100); conf.setInt( DFSConfigKeys.DFS_NAMENODE_REPLICATION_WORK_MULTIPLIER_PER_ITERATION, 5); int NUM_OF_BLOCKS = 10; final short REP_FACTOR = 2; final String FILE_NAME = "/testFile"; final Path FILE_PATH = new Path(FILE_NAME); MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).numDataNodes( REP_FACTOR).build(); try { final FileSystem fs = cluster.getFileSystem(); DFSTestUtil.createFile(fs, FILE_PATH, NUM_OF_BLOCKS, REP_FACTOR, 1L); DFSTestUtil.waitReplication(fs, FILE_PATH, REP_FACTOR); cluster.startDataNodes(conf, 1, true, null, null, null, null); final BlockManager bm = cluster.getNamesystem().getBlockManager(); ExtendedBlock b = DFSTestUtil.getFirstBlock(fs, FILE_PATH); Iterator<DatanodeStorageInfo> storageInfos = bm.blocksMap.getStorages(b.getLocalBlock()) .iterator(); DatanodeDescriptor firstDn = storageInfos.next().getDatanodeDescriptor(); DatanodeDescriptor secondDn = storageInfos.next().getDatanodeDescriptor(); bm.getDatanodeManager().removeDatanode(firstDn); assertEquals(NUM_OF_BLOCKS, bm.getUnderReplicatedNotMissingBlocks()); bm.computeDatanodeWork(); assertTrue("The number of blocks to be replicated should be less than " + "or equal to " + bm.replicationStreamsHardLimit, secondDn.getNumberOfBlocksToBeReplicated() <= bm.replicationStreamsHardLimit); } finally { cluster.shutdown(); } }
synchronized BlockIterator iterator(int level) { return new BlockIterator(level); }
UnderReplicatedBlocks implements Iterable<Block> { synchronized BlockIterator iterator(int level) { return new BlockIterator(level); } }
UnderReplicatedBlocks implements Iterable<Block> { synchronized BlockIterator iterator(int level) { return new BlockIterator(level); } UnderReplicatedBlocks(); }
UnderReplicatedBlocks implements Iterable<Block> { synchronized BlockIterator iterator(int level) { return new BlockIterator(level); } UnderReplicatedBlocks(); synchronized List<List<Block>> chooseUnderReplicatedBlocks(
      int blocksToProcess); @Override synchronized BlockIterator iterator(); void decrementReplicationIndex(int priority); }
UnderReplicatedBlocks implements Iterable<Block> { synchronized BlockIterator iterator(int level) { return new BlockIterator(level); } UnderReplicatedBlocks(); synchronized List<List<Block>> chooseUnderReplicatedBlocks(
      int blocksToProcess); @Override synchronized BlockIterator iterator(); void decrementReplicationIndex(int priority);  }