HADOOP-16161. NetworkTopology#getWeightUsingNetworkLocation return unexpected result. Contributed by He Xiaoqiao.

(cherry picked from commit 389e640f0c)

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/net/NetworkTopology.java

@@ -759,6 +759,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
@@ -801,7 +802,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) {
@@ -831,8 +833,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;

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestDatanodeManager.java

@@ -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.

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/net/TestNetworkTopology.java

@@ -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);