HADOOP-16161. NetworkTopology#getWeightUsingNetworkLocation return unexpected result. Contributed by He Xiaoqiao.
This commit is contained in:
parent
206e633946
commit
389e640f0c
|
@ -758,6 +758,7 @@ public class NetworkTopology {
|
|||
* @param node Replica of data
|
||||
* @return weight
|
||||
*/
|
||||
@VisibleForTesting
|
||||
protected int getWeight(Node reader, Node node) {
|
||||
// 0 is local, 2 is same rack, and each level on each node increases the
|
||||
//weight by 1
|
||||
|
@ -800,7 +801,8 @@ public class NetworkTopology {
|
|||
* @param node Replica of data
|
||||
* @return weight
|
||||
*/
|
||||
private static int getWeightUsingNetworkLocation(Node reader, Node node) {
|
||||
@VisibleForTesting
|
||||
protected static int getWeightUsingNetworkLocation(Node reader, Node node) {
|
||||
//Start off by initializing to Integer.MAX_VALUE
|
||||
int weight = Integer.MAX_VALUE;
|
||||
if(reader != null && node != null) {
|
||||
|
@ -830,8 +832,10 @@ public class NetworkTopology {
|
|||
}
|
||||
currentLevel++;
|
||||
}
|
||||
// +2 to correct the weight between reader and node rather than
|
||||
// between parent of reader and parent of node.
|
||||
weight = (readerPathToken.length - currentLevel) +
|
||||
(nodePathToken.length - currentLevel);
|
||||
(nodePathToken.length - currentLevel) + 2;
|
||||
}
|
||||
}
|
||||
return weight;
|
||||
|
|
|
@ -453,6 +453,71 @@ public class TestDatanodeManager {
|
|||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testGetBlockLocations()
|
||||
throws URISyntaxException, IOException {
|
||||
// create the DatanodeManager which will be tested
|
||||
Configuration conf = new Configuration();
|
||||
FSNamesystem fsn = Mockito.mock(FSNamesystem.class);
|
||||
Mockito.when(fsn.hasWriteLock()).thenReturn(true);
|
||||
URL shellScript = getClass().getResource(
|
||||
"/" + Shell.appendScriptExtension("topology-script"));
|
||||
Path resourcePath = Paths.get(shellScript.toURI());
|
||||
FileUtil.setExecutable(resourcePath.toFile(), true);
|
||||
conf.set(DFSConfigKeys.NET_TOPOLOGY_SCRIPT_FILE_NAME_KEY,
|
||||
resourcePath.toString());
|
||||
DatanodeManager dm = mockDatanodeManager(fsn, conf);
|
||||
|
||||
int totalDNs = 5;
|
||||
// register 5 datanodes and 2 node per rack
|
||||
DatanodeInfo[] locs = new DatanodeInfo[totalDNs];
|
||||
String[] storageIDs = new String[totalDNs];
|
||||
for (int i = 0; i < totalDNs; i++) {
|
||||
// register new datanode
|
||||
String uuid = "UUID-" + i;
|
||||
String ip = "IP-" + i / 2 + "-" + i;
|
||||
DatanodeRegistration dr = Mockito.mock(DatanodeRegistration.class);
|
||||
Mockito.when(dr.getDatanodeUuid()).thenReturn(uuid);
|
||||
Mockito.when(dr.getIpAddr()).thenReturn(ip);
|
||||
dm.registerDatanode(dr);
|
||||
|
||||
// get location and storage information
|
||||
locs[i] = dm.getDatanode(uuid);
|
||||
storageIDs[i] = "storageID-" + i;
|
||||
}
|
||||
|
||||
// set first 2 locations as decommissioned
|
||||
locs[0].setDecommissioned();
|
||||
locs[1].setDecommissioned();
|
||||
|
||||
// create LocatedBlock with above locations
|
||||
ExtendedBlock b = new ExtendedBlock("somePoolID", 1234);
|
||||
LocatedBlock block = new LocatedBlock(b, locs);
|
||||
List<LocatedBlock> blocks = new ArrayList<>();
|
||||
blocks.add(block);
|
||||
|
||||
// test client in cluster
|
||||
final String targetIpInCluster = locs[4].getIpAddr();
|
||||
dm.sortLocatedBlocks(targetIpInCluster, blocks);
|
||||
DatanodeInfo[] sortedLocs = block.getLocations();
|
||||
assertEquals(totalDNs, sortedLocs.length);
|
||||
// Ensure the local node is first.
|
||||
assertEquals(targetIpInCluster, sortedLocs[0].getIpAddr());
|
||||
// Ensure the two decommissioned DNs were moved to the end.
|
||||
assertEquals(DatanodeInfo.AdminStates.DECOMMISSIONED,
|
||||
sortedLocs[sortedLocs.length -1].getAdminState());
|
||||
assertEquals(DatanodeInfo.AdminStates.DECOMMISSIONED,
|
||||
sortedLocs[sortedLocs.length - 2].getAdminState());
|
||||
|
||||
// test client not in cluster but same rack with locs[4]
|
||||
final String targetIpNotInCluster = locs[4].getIpAddr() + "-client";
|
||||
dm.sortLocatedBlocks(targetIpNotInCluster, blocks);
|
||||
DatanodeInfo[] sortedLocs2 = block.getLocations();
|
||||
assertEquals(totalDNs, sortedLocs2.length);
|
||||
// Ensure the local rack is first.
|
||||
assertEquals(locs[4].getIpAddr(), sortedLocs2[0].getIpAddr());
|
||||
}
|
||||
|
||||
/**
|
||||
* Test whether removing a host from the includes list without adding it to
|
||||
* the excludes list will exclude it from data node reports.
|
||||
|
|
|
@ -137,7 +137,62 @@ public class TestNetworkTopology {
|
|||
assertFalse(cluster.isOnSameRack(dataNodes[4], dataNodes[5]));
|
||||
assertTrue(cluster.isOnSameRack(dataNodes[5], dataNodes[6]));
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testGetWeight() throws Exception {
|
||||
DatanodeDescriptor nodeInMap = dataNodes[0];
|
||||
assertEquals(0, cluster.getWeight(nodeInMap, dataNodes[0]));
|
||||
assertEquals(2, cluster.getWeight(nodeInMap, dataNodes[1]));
|
||||
assertEquals(4, cluster.getWeight(nodeInMap, dataNodes[2]));
|
||||
|
||||
DatanodeDescriptor nodeNotInMap =
|
||||
DFSTestUtil.getDatanodeDescriptor("21.21.21.21", "/d1/r2");
|
||||
assertEquals(4, cluster.getWeightUsingNetworkLocation(nodeNotInMap,
|
||||
dataNodes[0]));
|
||||
assertEquals(4, cluster.getWeightUsingNetworkLocation(nodeNotInMap,
|
||||
dataNodes[1]));
|
||||
assertEquals(2, cluster.getWeightUsingNetworkLocation(nodeNotInMap,
|
||||
dataNodes[2]));
|
||||
}
|
||||
|
||||
/**
|
||||
* Test getWeight/getWeightUsingNetworkLocation for complex topology.
|
||||
*/
|
||||
@Test
|
||||
public void testGetWeightForDepth() throws Exception {
|
||||
NetworkTopology topology = NetworkTopology.getInstance(new Configuration());
|
||||
DatanodeDescriptor[] dns = new DatanodeDescriptor[] {
|
||||
DFSTestUtil.getDatanodeDescriptor("1.1.1.1", "/z1/d1/p1/r1"),
|
||||
DFSTestUtil.getDatanodeDescriptor("2.2.2.2", "/z1/d1/p1/r1"),
|
||||
DFSTestUtil.getDatanodeDescriptor("3.3.3.3", "/z1/d1/p2/r2"),
|
||||
DFSTestUtil.getDatanodeDescriptor("4.4.4.4", "/z1/d2/p1/r2"),
|
||||
DFSTestUtil.getDatanodeDescriptor("5.5.5.5", "/z2/d3/p1/r1"),
|
||||
};
|
||||
for (int i = 0; i < dns.length; i++) {
|
||||
topology.add(dns[i]);
|
||||
}
|
||||
|
||||
DatanodeDescriptor nodeInMap = dns[0];
|
||||
assertEquals(0, topology.getWeight(nodeInMap, dns[0]));
|
||||
assertEquals(2, topology.getWeight(nodeInMap, dns[1]));
|
||||
assertEquals(6, topology.getWeight(nodeInMap, dns[2]));
|
||||
assertEquals(8, topology.getWeight(nodeInMap, dns[3]));
|
||||
assertEquals(10, topology.getWeight(nodeInMap, dns[4]));
|
||||
|
||||
DatanodeDescriptor nodeNotInMap =
|
||||
DFSTestUtil.getDatanodeDescriptor("6.6.6.6", "/z1/d1/p1/r2");
|
||||
assertEquals(4, topology.getWeightUsingNetworkLocation(
|
||||
nodeNotInMap, dns[0]));
|
||||
assertEquals(4, topology.getWeightUsingNetworkLocation(
|
||||
nodeNotInMap, dns[1]));
|
||||
assertEquals(6, topology.getWeightUsingNetworkLocation(
|
||||
nodeNotInMap, dns[2]));
|
||||
assertEquals(8, topology.getWeightUsingNetworkLocation(
|
||||
nodeNotInMap, dns[3]));
|
||||
assertEquals(10, topology.getWeightUsingNetworkLocation(
|
||||
nodeNotInMap, dns[4]));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testGetDistance() throws Exception {
|
||||
assertEquals(cluster.getDistance(dataNodes[0], dataNodes[0]), 0);
|
||||
|
|
Loading…
Reference in New Issue