@Override
  public void sortByDistance(Node reader, Node[] nodes, int activeLen) {
    // If reader is not a datanode (not in NetworkTopology tree), we need to
    // replace this reader with a sibling leaf node in tree.
    if (reader != null && !this.contains(reader)) {
      Node nodeGroup = getNode(reader.getNetworkLocation());
      if (nodeGroup != null && nodeGroup instanceof InnerNode) {
        InnerNode parentNode = (InnerNode) nodeGroup;
        // replace reader with the first children of its parent in tree
        reader = parentNode.getLeaf(0, null);
      } else {
        return;
      }
    }
    super.sortByDistance(reader, nodes, activeLen);
  }
@Test
  public void testSortByDistance() throws Exception {
    NodeBase[] testNodes = new NodeBase[4];

    // array contains both local node, local node group & local rack node
    testNodes[0] = dataNodes[1];
    testNodes[1] = dataNodes[2];
    testNodes[2] = dataNodes[3];
    testNodes[3] = dataNodes[0];
    cluster.sortByDistance(dataNodes[0], testNodes, testNodes.length);
    assertTrue(testNodes[0] == dataNodes[0]);
    assertTrue(testNodes[1] == dataNodes[1]);
    assertTrue(testNodes[2] == dataNodes[2]);
    assertTrue(testNodes[3] == dataNodes[3]);

    // array contains local node & local node group
    testNodes[0] = dataNodes[3];
    testNodes[1] = dataNodes[4];
    testNodes[2] = dataNodes[1];
    testNodes[3] = dataNodes[0];
    cluster.sortByDistance(dataNodes[0], testNodes, testNodes.length);
    assertTrue(testNodes[0] == dataNodes[0]);
    assertTrue(testNodes[1] == dataNodes[1]);

    // array contains local node & rack node
    testNodes[0] = dataNodes[5];
    testNodes[1] = dataNodes[3];
    testNodes[2] = dataNodes[2];
    testNodes[3] = dataNodes[0];
    cluster.sortByDistance(dataNodes[0], testNodes, testNodes.length);
    assertTrue(testNodes[0] == dataNodes[0]);
    assertTrue(testNodes[1] == dataNodes[2]);

    // array contains local-nodegroup node (not a data node also) & rack node
    testNodes[0] = dataNodes[6];
    testNodes[1] = dataNodes[7];
    testNodes[2] = dataNodes[2];
    testNodes[3] = dataNodes[0];
    cluster.sortByDistance(computeNode, testNodes, testNodes.length);
    assertTrue(testNodes[0] == dataNodes[0]);
    assertTrue(testNodes[1] == dataNodes[2]);
  }