class BlockManager {

  @VisibleForTesting
  DatanodeDescriptor chooseSourceDatanode(Block b,
      List<DatanodeDescriptor> containingNodes,
      List<DatanodeStorageInfo> nodesContainingLiveReplicas,
      NumberReplicas numReplicas, int priority)
      throws IOException {
    containingNodes.clear();
    nodesContainingLiveReplicas.clear();
    DatanodeDescriptor srcNode = null;
    int live = 0;
    int decommissioned = 0;
    int decommissioning = 0;
    int corrupt = 0;
    int excess = 0;
    final BlockInfoContiguous block = getBlockInfo(b);

    Collection<DatanodeDescriptor> nodesCorrupt = corruptReplicas.getNodes(block);
    for(DatanodeStorageInfo storage : block.getStorages(datanodeManager)) {
      final DatanodeDescriptor node = storage.getDatanodeDescriptor();
      int countableReplica = storage.getState() == State.NORMAL ? 1 : 0; 
      if ((nodesCorrupt != null) && (nodesCorrupt.contains(node))) {
        corrupt += countableReplica;
      } else if (node.isDecommissionInProgress()) {
        decommissioning += countableReplica;
      } else if (node.isDecommissioned()) {
        decommissioned += countableReplica;
      } else if (excessReplicateMap.contains(storage, block)) {
        excess += countableReplica;
      } else {
        nodesContainingLiveReplicas.add(storage);
        live+= countableReplica;
      }
      if(!containingNodes.contains(node)) {
        containingNodes.add(node);
      }


      // Check if this replica is corrupt
      // If so, do not select the node as src node
      if ((nodesCorrupt != null) && nodesCorrupt.contains(node)) {
        continue;
      }
      if(priority != UnderReplicatedBlocks.QUEUE_HIGHEST_PRIORITY 
          && !node.isDecommissionInProgress() 
          && node.getNumberOfBlocksToBeReplicated() >= maxReplicationStreams)
      {
        continue; // already reached replication limit
      }
      if (node.getNumberOfBlocksToBeReplicated() >= replicationStreamsHardLimit) {
        continue;
      }
      // the block must not be scheduled for removal on srcNode
      if (excessReplicateMap.contains(storage, block)) {
        continue;
      }
      // never use already decommissioned nodes
      if (node.isDecommissioned()) {
        continue;
      }
      // We got this far, current node is a reasonable choice
      if (srcNode == null) {
        srcNode = node;
        continue;
      }
      // switch to a different node randomly
      // this to prevent from deterministically selecting the same node even
      // if the node failed to replicate the block on previous iterations
      if (DFSUtil.getRandom().nextBoolean()) {
        srcNode = node;
      }
    }
    if(numReplicas != null)
      numReplicas.initialize(live, decommissioned, decommissioning, corrupt,
          excess, 0);
    return srcNode;
  }

  public  BlockManager(final Namesystem namesystem, final Configuration conf);

}