private void processOverReplicatedBlocksAsync ( ) { // blocks should not be scheduled for deletion during safemode if ( isInSafeMode ( ) ) { return ; } if ( delayOverreplicationMonitorTime > now ( ) ) { LOG . info ( "Overreplication monitor delayed for " + ( ( delayOverreplicationMonitorTime - now ( ) ) / 1000 ) + " seconds" ) ; return ; } nameNode . clearOutstandingNodes ( ) ; final int nodes = heartbeats . size ( ) ; List < Block > blocksToProcess = new ArrayList < Block > ( Math . min ( overReplicatedBlocks . size ( ) , ReplicationConfigKeys . overreplicationWorkMultiplier * nodes ) ) ; for ( int i = 0 ; i < ReplicationConfigKeys . overreplicationWorkMultiplier ; i ++ ) { writeLock ( ) ; try { NameNode . getNameNodeMetrics ( ) . numOverReplicatedBlocks . set ( overReplicatedBlocks . size ( ) ) ; overReplicatedBlocks . pollNToList ( nodes , blocksToProcess ) ; if ( overReplicatedBlocks . isEmpty ( ) ) { break ; } } finally { writeUnlock ( ) ; } } for ( Block block : blocksToProcess ) { if ( NameNode . stateChangeLog . isDebugEnabled ( ) ) { NameNode . stateChangeLog . debug ( "BLOCK* NameSystem.processOverReplicatedBlocksAsync: " + block ) ; } if ( block instanceof OverReplicatedBlock ) { OverReplicatedBlock opb = ( OverReplicatedBlock ) block ; processOverReplicatedBlock ( block , ( short ) - 1 , opb . addedNode , opb . delNodeHint ) ; } else { processOverReplicatedBlock ( block , ( short ) - 1 , null , null ) ; } } }
This is called from the ReplicationMonitor to process over replicated blocks .