BlockManager { @VisibleForTesting DatanodeDescriptor chooseSourceDatanode(Block block, List<DatanodeDescriptor> containingNodes, List<DatanodeStorageInfo> nodesContainingLiveReplicas, NumberReplicas numReplicas, int priority) { containingNodes.clear(); nodesContainingLiveReplicas.clear(); DatanodeDescriptor srcNode = null; int live = 0; int decommissioned = 0; int corrupt = 0; int excess = 0; Collection<DatanodeDescriptor> nodesCorrupt = corruptReplicas.getNodes(block); for(DatanodeStorageInfo storage : blocksMap.getStorages(block)) { final DatanodeDescriptor node = storage.getDatanodeDescriptor(); LightWeightLinkedSet<Block> excessBlocks = excessReplicateMap.get(node.getDatanodeUuid()); int countableReplica = storage.getState() == State.NORMAL ? 1 : 0; if ((nodesCorrupt != null) && (nodesCorrupt.contains(node))) corrupt += countableReplica; else if (node.isDecommissionInProgress() || node.isDecommissioned()) decommissioned += countableReplica; else if (excessBlocks != null && excessBlocks.contains(block)) { excess += countableReplica; } else { nodesContainingLiveReplicas.add(storage); live += countableReplica; } containingNodes.add(node); if ((nodesCorrupt != null) && nodesCorrupt.contains(node)) continue; if(priority != UnderReplicatedBlocks.QUEUE_HIGHEST_PRIORITY && !node.isDecommissionInProgress() && node.getNumberOfBlocksToBeReplicated() >= maxReplicationStreams) { continue; } if (node.getNumberOfBlocksToBeReplicated() >= replicationStreamsHardLimit) { continue; } if(excessBlocks != null && excessBlocks.contains(block)) continue; if(node.isDecommissioned()) continue; if (srcNode == null) { srcNode = node; continue; } if(DFSUtil.getRandom().nextBoolean()) srcNode = node; } if(numReplicas != null) numReplicas.initialize(live, decommissioned, corrupt, excess, 0); return srcNode; } BlockManager(final Namesystem namesystem, final FSClusterStats stats,
      final Configuration conf); long getPendingReplicationBlocksCount(); long getUnderReplicatedBlocksCount(); long getCorruptReplicaBlocksCount(); long getScheduledReplicationBlocksCount(); long getPendingDeletionBlocksCount(); long getStartupDelayBlockDeletionInMs(); long getExcessBlocksCount(); long getPostponedMisreplicatedBlocksCount(); int getPendingDataNodeMessageCount(); BlockStoragePolicy getStoragePolicy(final String policyName); BlockStoragePolicy getStoragePolicy(final byte policyId); BlockStoragePolicy[] getStoragePolicies(); void setBlockPoolId(String blockPoolId); @VisibleForTesting BlockTokenSecretManager getBlockTokenSecretManager(); void activate(Configuration conf); void close(); DatanodeManager getDatanodeManager(); @VisibleForTesting BlockPlacementPolicy getBlockPlacementPolicy(); void setBlockPlacementPolicy(BlockPlacementPolicy newpolicy); void metaSave(PrintWriter out); int getMaxReplicationStreams(); boolean checkMinReplication(Block block); boolean commitOrCompleteLastBlock(BlockCollection bc,
      Block commitBlock); BlockInfo forceCompleteBlock(final BlockCollection bc,
      final BlockInfoUnderConstruction block); LocatedBlock convertLastBlockToUnderConstruction(
      BlockCollection bc); LocatedBlocks createLocatedBlocks(final BlockInfo[] blocks,
      final long fileSizeExcludeBlocksUnderConstruction,
      final boolean isFileUnderConstruction, final long offset,
      final long length, final boolean needBlockToken,
      final boolean inSnapshot, FileEncryptionInfo feInfo); ExportedBlockKeys getBlockKeys(); void setBlockToken(final LocatedBlock b,
      final BlockTokenSecretManager.AccessMode mode); DataEncryptionKey generateDataEncryptionKey(); short adjustReplication(short replication); void verifyReplication(String src,
                          short replication,
                          String clientName); boolean isSufficientlyReplicated(BlockInfo b); BlocksWithLocations getBlocks(DatanodeID datanode, long size
      ); void findAndMarkBlockAsCorrupt(final ExtendedBlock blk,
      final DatanodeInfo dn, String storageID, String reason); void setPostponeBlocksFromFuture(boolean postpone); int getUnderReplicatedNotMissingBlocks(); DatanodeStorageInfo[] chooseTarget4WebHDFS(String src,
      DatanodeDescriptor clientnode, Set<Node> excludes, long blocksize); DatanodeStorageInfo[] chooseTarget4AdditionalDatanode(String src,
      int numAdditionalNodes,
      Node clientnode,
      List<DatanodeStorageInfo> chosen,
      Set<Node> excludes,
      long blocksize,
      byte storagePolicyID); DatanodeStorageInfo[] chooseTarget4NewBlock(final String src,
      final int numOfReplicas, final Node client,
      final Set<Node> excludedNodes,
      final long blocksize,
      final List<String> favoredNodes,
      final byte storagePolicyID); boolean processReport(final DatanodeID nodeID,
      final DatanodeStorage storage,
      final BlockListAsLongs newReport, BlockReportContext context,
      boolean lastStorageInRpc); void markBlockReplicasAsCorrupt(BlockInfo block, 
      long oldGenerationStamp, long oldNumBytes, 
      DatanodeStorageInfo[] newStorages); void processQueuedMessagesForBlock(Block b); void processAllPendingDNMessages(); void processMisReplicatedBlocks(); double getReplicationQueuesInitProgress(); void setReplication(final short oldRepl, final short newRepl,
      final String src, final Block... blocks); void removeStoredBlock(Block block, DatanodeDescriptor node); void processIncrementalBlockReport(final DatanodeID nodeID,
      final StorageReceivedDeletedBlocks srdb); NumberReplicas countNodes(Block b); int getActiveBlockCount(); DatanodeStorageInfo[] getStorages(BlockInfo block); int getTotalBlocks(); void removeBlock(Block block); BlockInfo getStoredBlock(Block block); void checkReplication(BlockCollection bc); long getMissingBlocksCount(); BlockInfo addBlockCollection(BlockInfo block, BlockCollection bc); BlockCollection getBlockCollection(Block b); Iterable<DatanodeStorageInfo> getStorages(final Block block); int numCorruptReplicas(Block block); void removeBlockFromMap(Block block); int getCapacity(); long[] getCorruptReplicaBlockIds(int numExpectedBlocks,
                                   Long startingBlockId); Iterator<Block> getCorruptReplicaBlockIterator(); Collection<DatanodeDescriptor> getCorruptReplicas(Block block); int numOfUnderReplicatedBlocks(); void clearQueues(); void shutdown(); void clear(); static final Log blockLog; final Map<String, LightWeightLinkedSet<Block>> excessReplicateMap; final UnderReplicatedBlocks neededReplications; final short maxReplication; final short minReplication; final int defaultReplication; }
@Test public void testHighestPriReplSrcChosenDespiteMaxReplLimit() throws Exception { bm.maxReplicationStreams = 0; bm.replicationStreamsHardLimit = 1; long blockId = 42; Block aBlock = new Block(blockId, 0, 0); List<DatanodeDescriptor> origNodes = getNodes(0, 1); addBlockOnNodes(blockId,origNodes.subList(0,1)); List<DatanodeDescriptor> cntNodes = new LinkedList<DatanodeDescriptor>(); List<DatanodeStorageInfo> liveNodes = new LinkedList<DatanodeStorageInfo>(); assertNotNull("Chooses source node for a highest-priority replication" + " even if all available source nodes have reached their replication" + " limits below the hard limit.", bm.chooseSourceDatanode( aBlock, cntNodes, liveNodes, new NumberReplicas(), UnderReplicatedBlocks.QUEUE_HIGHEST_PRIORITY)); assertNull("Does not choose a source node for a less-than-highest-priority" + " replication since all available source nodes have reached" + " their replication limits.", bm.chooseSourceDatanode( aBlock, cntNodes, liveNodes, new NumberReplicas(), UnderReplicatedBlocks.QUEUE_VERY_UNDER_REPLICATED)); DatanodeStorageInfo targets[] = { origNodes.get(1).getStorageInfos()[0] }; origNodes.get(0).addBlockToBeReplicated(aBlock, targets); assertNull("Does not choose a source node for a highest-priority" + " replication when all available nodes exceed the hard limit.", bm.chooseSourceDatanode( aBlock, cntNodes, liveNodes, new NumberReplicas(), UnderReplicatedBlocks.QUEUE_HIGHEST_PRIORITY)); }
@Test public void testFavorDecomUntilHardLimit() throws Exception { bm.maxReplicationStreams = 0; bm.replicationStreamsHardLimit = 1; long blockId = 42; Block aBlock = new Block(blockId, 0, 0); List<DatanodeDescriptor> origNodes = getNodes(0, 1); addBlockOnNodes(blockId,origNodes.subList(0,1)); origNodes.get(0).startDecommission(); List<DatanodeDescriptor> cntNodes = new LinkedList<DatanodeDescriptor>(); List<DatanodeStorageInfo> liveNodes = new LinkedList<DatanodeStorageInfo>(); assertNotNull("Chooses decommissioning source node for a normal replication" + " if all available source nodes have reached their replication" + " limits below the hard limit.", bm.chooseSourceDatanode( aBlock, cntNodes, liveNodes, new NumberReplicas(), UnderReplicatedBlocks.QUEUE_UNDER_REPLICATED)); DatanodeStorageInfo targets[] = { origNodes.get(1).getStorageInfos()[0] }; origNodes.get(0).addBlockToBeReplicated(aBlock, targets); assertNull("Does not choose a source decommissioning node for a normal" + " replication when all available nodes exceed the hard limit.", bm.chooseSourceDatanode( aBlock, cntNodes, liveNodes, new NumberReplicas(), UnderReplicatedBlocks.QUEUE_UNDER_REPLICATED)); }