DatanodeManager { public void sortLocatedBlocks(final String targethost, final List<LocatedBlock> locatedblocks) { Node client = getDatanodeByHost(targethost); if (client == null) { List<String> hosts = new ArrayList<String> (1); hosts.add(targethost); String rName = dnsToSwitchMapping.resolve(hosts).get(0); if (rName != null) client = new NodeBase(rName + NodeBase.PATH_SEPARATOR_STR + targethost); } Comparator<DatanodeInfo> comparator = avoidStaleDataNodesForRead ? new DFSUtil.DecomStaleComparator(staleInterval) : DFSUtil.DECOM_COMPARATOR; for (LocatedBlock b : locatedblocks) { DatanodeInfo[] di = b.getLocations(); Arrays.sort(di, comparator); int lastActiveIndex = di.length - 1; while (lastActiveIndex > 0 && isInactive(di[lastActiveIndex])) { --lastActiveIndex; } int activeLen = lastActiveIndex + 1; networktopology.sortByDistance(client, b.getLocations(), activeLen); b.updateCachedStorageInfo(); } } DatanodeManager(final BlockManager blockManager, final Namesystem namesystem,
      final Configuration conf); NetworkTopology getNetworkTopology(); @VisibleForTesting DecommissionManager getDecomManager(); @VisibleForTesting void setHeartbeatExpireInterval(long expiryMs); @VisibleForTesting FSClusterStats getFSClusterStats(); DatanodeStatistics getDatanodeStatistics(); void sortLocatedBlocks(final String targethost,
      final List<LocatedBlock> locatedblocks); DatanodeDescriptor getDatanodeByHost(final String host); DatanodeDescriptor getDatanodeByXferAddr(String host, int xferPort); Host2NodesMap getHost2DatanodeMap(); DatanodeDescriptor getDatanode(DatanodeID nodeID
      ); DatanodeStorageInfo[] getDatanodeStorageInfos(
      DatanodeID[] datanodeID, String[] storageIDs); void removeDatanode(final DatanodeID node
      ); HashMap<String, Integer> getDatanodesSoftwareVersions(); List<String> resolveNetworkLocation(List<String> names); void registerDatanode(DatanodeRegistration nodeReg); void refreshNodes(final Configuration conf); int getNumLiveDataNodes(); int getNumDeadDataNodes(); List<DatanodeDescriptor> getDecommissioningNodes(); boolean shouldAvoidStaleDataNodesForWrite(); long getBlocksPerPostponedMisreplicatedBlocksRescan(); int getNumStaleNodes(); int getNumStaleStorages(); void fetchDatanodes(final List<DatanodeDescriptor> live, 
      final List<DatanodeDescriptor> dead, final boolean removeDecommissionNode); List<DatanodeDescriptor> getDatanodeListForReport(
      final DatanodeReportType type); DatanodeCommand[] handleHeartbeat(DatanodeRegistration nodeReg,
      StorageReport[] reports, final String blockPoolId,
      long cacheCapacity, long cacheUsed, int xceiverCount, 
      int maxTransfers, int failedVolumes,
      VolumeFailureSummary volumeFailureSummary); void setBalancerBandwidth(long bandwidth); void markAllDatanodesStale(); void clearPendingQueues(); void resetLastCachingDirectiveSentTime(); @Override String toString(); void clearPendingCachingCommands(); void setShouldSendCachingCommands(boolean shouldSendCachingCommands);  }
@Test public void testSortLocatedBlocks() throws IOException { FSNamesystem fsn = Mockito.mock(FSNamesystem.class); Mockito.when(fsn.hasWriteLock()).thenReturn(true); DatanodeManager dm = new DatanodeManager(Mockito.mock(BlockManager.class), fsn, new Configuration()); DatanodeInfo[] locs = new DatanodeInfo[5]; String[] storageIDs = new String[5]; StorageType[] storageTypes = new StorageType[]{ StorageType.ARCHIVE, StorageType.DEFAULT, StorageType.DISK, StorageType.RAM_DISK, StorageType.SSD }; for(int i = 0; i < 5; i++) { String uuid = "UUID-"+i; String ip = "IP-" + i; DatanodeRegistration dr = Mockito.mock(DatanodeRegistration.class); Mockito.when(dr.getDatanodeUuid()).thenReturn(uuid); Mockito.when(dr.getIpAddr()).thenReturn(ip); Mockito.when(dr.getXferAddr()).thenReturn(ip + ":9000"); Mockito.when(dr.getXferPort()).thenReturn(9000); Mockito.when(dr.getSoftwareVersion()).thenReturn("version1"); dm.registerDatanode(dr); locs[i] = dm.getDatanode(uuid); storageIDs[i] = "storageID-"+i; } locs[0].setDecommissioned(); locs[1].setDecommissioned(); ExtendedBlock b = new ExtendedBlock("somePoolID", 1234); LocatedBlock block = new LocatedBlock(b, locs, storageIDs, storageTypes); List<LocatedBlock> blocks = new ArrayList<>(); blocks.add(block); final String targetIp = locs[4].getIpAddr(); dm.sortLocatedBlocks(targetIp, blocks); DatanodeInfo[] sortedLocs = block.getLocations(); storageIDs = block.getStorageIDs(); storageTypes = block.getStorageTypes(); assertThat(sortedLocs.length, is(5)); assertThat(storageIDs.length, is(5)); assertThat(storageTypes.length, is(5)); for(int i = 0; i < sortedLocs.length; i++) { assertThat(((DatanodeInfoWithStorage)sortedLocs[i]).getStorageID(), is(storageIDs[i])); assertThat(((DatanodeInfoWithStorage)sortedLocs[i]).getStorageType(), is(storageTypes[i])); } assertThat(sortedLocs[0].getIpAddr(), is(targetIp)); assertThat(sortedLocs[sortedLocs.length-1].getAdminState(), is(DatanodeInfo.AdminStates.DECOMMISSIONED)); assertThat(sortedLocs[sortedLocs.length-2].getAdminState(), is(DatanodeInfo.AdminStates.DECOMMISSIONED)); }