UnderReplicatedBlocks implements Iterable<Block> { BlockIterator iterator(final int level) { try { return (BlockIterator) new HopsTransactionalRequestHandler( HDFSOperationType.UNDER_REPLICATED_BLKS_ITERATOR) { @Override public void acquireLock(TransactionLocks locks) throws IOException { } @Override public Object performTask() throws StorageException, IOException { return new BlockIterator(fillPriorityQueues(level), level); } }.handle(); } catch (IOException ex) { BlockManager.LOG .error("Error while filling the priorityQueues from db", ex); return null; } } UnderReplicatedBlocks(); @Override BlockIterator iterator(); void decrementReplicationIndex(int priority); List<List<Block>> chooseUnderReplicatedBlocks(
      final int blocksToProcess);  }
@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 = getStorageInfo(b.getLocalBlock(), bm) .iterator(); DatanodeDescriptor firstDn = storageInfos.next().getDatanodeDescriptor(); DatanodeDescriptor secondDn = storageInfos.next().getDatanodeDescriptor(); bm.getDatanodeManager().removeDatanode(firstDn,false); 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(); } }