UnderReplicatedBlocks implements Iterable<Block> { boolean add(BlockInfoContiguous block, int curReplicas, int decomissionedReplicas, int expectedReplicas) throws StorageException, TransactionContextException { assert curReplicas >= 0 : "Negative replicas!"; int priLevel = getPriority(block, curReplicas, decomissionedReplicas, expectedReplicas); if(add(block, priLevel, expectedReplicas)) { NameNode.blockStateChangeLog.debug( "BLOCK* NameSystem.UnderReplicationBlock.add: {}" + " has only {} replicas and need {} replicas so is added to" + " neededReplications at priority level {}", block, curReplicas, expectedReplicas, priLevel); return true; } return false; } UnderReplicatedBlocks(); @Override BlockIterator iterator(); void decrementReplicationIndex(int priority); List<List<Block>> chooseUnderReplicatedBlocks(
      final int blocksToProcess);  }
@Test(timeout = 60000) public void testSetrepIncWithUnderReplicatedBlocks() throws Exception { Configuration conf = new HdfsConfiguration(); final short REPLICATION_FACTOR = 2; final String FILE_NAME = "/testFile"; final Path FILE_PATH = new Path(FILE_NAME); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).numDataNodes(REPLICATION_FACTOR + 1) .build(); try { final FileSystem fs = cluster.getFileSystem(); DFSTestUtil.createFile(fs, FILE_PATH, 1L, REPLICATION_FACTOR, 1L); DFSTestUtil.waitReplication(fs, FILE_PATH, REPLICATION_FACTOR); final BlockManager bm = cluster.getNamesystem().getBlockManager(); final ExtendedBlock b = DFSTestUtil.getFirstBlock(fs, FILE_PATH); HopsTransactionalRequestHandler handler = new HopsTransactionalRequestHandler(HDFSOperationType.TEST) { INodeIdentifier inodeIdentifier; @Override public void setUp() throws StorageException { Block blk = b.getLocalBlock(); inodeIdentifier = INodeUtil.resolveINodeFromBlock(blk); } @Override public void acquireLock(TransactionLocks locks) throws IOException { LockFactory lf = LockFactory.getInstance(); locks.add(lf.getIndividualINodeLock(INodeLockType.WRITE, inodeIdentifier)).add( lf.getIndividualBlockLock(b.getBlockId(), inodeIdentifier)) .add(lf.getBlockRelated(LockFactory.BLK.RE, LockFactory.BLK.IV)); } @Override public Object performTask() throws StorageException, IOException { DatanodeStorageInfo storage = bm.blocksMap.storageList(b.getLocalBlock()).get(0); DatanodeDescriptor dn = storage.getDatanodeDescriptor(); bm.addToInvalidates(b.getLocalBlock(), dn); bm.blocksMap.removeNode(b.getLocalBlock(), dn); return dn; } }; DatanodeDescriptor dn = (DatanodeDescriptor) handler.handle(); BlockManagerTestUtil.computeAllPendingWork(bm); DataNodeTestUtils.triggerHeartbeat(cluster.getDataNode(dn.getIpcPort())); Thread.sleep(5000); HopsTransactionalRequestHandler handler2 = new HopsTransactionalRequestHandler(HDFSOperationType.TEST) { INodeIdentifier inodeIdentifier; @Override public void setUp() throws StorageException { Block blk = b.getLocalBlock(); inodeIdentifier = INodeUtil.resolveINodeFromBlock(blk); } @Override public void acquireLock(TransactionLocks locks) throws IOException { LockFactory lf = LockFactory.getInstance(); locks.add(lf.getIndividualINodeLock(INodeLockType.WRITE, inodeIdentifier)).add( lf.getIndividualBlockLock(b.getBlockId(), inodeIdentifier)) .add(lf.getBlockRelated(LockFactory.BLK.RE, LockFactory.BLK.IV)); } @Override public Object performTask() throws StorageException, IOException { DatanodeDescriptor dn = (DatanodeDescriptor) getParams()[0]; bm.blocksMap.removeNode(b.getLocalBlock(), dn); return null; } }; handler2.setParams(dn); handler2.handle(); FsShell shell = new FsShell(conf); assertEquals(0, shell.run(new String[]{"-setrep", "-w", Integer.toString(1 + REPLICATION_FACTOR), FILE_NAME})); } finally { cluster.shutdown(); } }