/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.hdds.scm.net;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.stream.Collectors;

import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;

import static org.apache.hadoop.hdds.scm.net.NetConstants.DATACENTER_SCHEMA;
import static org.apache.hadoop.hdds.scm.net.NetConstants.LEAF_SCHEMA;
import static org.apache.hadoop.hdds.scm.net.NetConstants.NODEGROUP_SCHEMA;
import static org.apache.hadoop.hdds.scm.net.NetConstants.PATH_SEPARATOR_STR;
import static org.apache.hadoop.hdds.scm.net.NetConstants.RACK_SCHEMA;
import static org.apache.hadoop.hdds.scm.net.NetConstants.REGION_SCHEMA;
import static org.apache.hadoop.hdds.scm.net.NetConstants.ROOT;
import static org.apache.hadoop.hdds.scm.net.NetConstants.ROOT_SCHEMA;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.junit.Assume.assumeTrue;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.Timeout;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** Test the network topology functions. */
@RunWith(Parameterized.class)
public class TestNetworkTopologyImpl {
  private static final Logger LOG = LoggerFactory.getLogger(
      TestNetworkTopologyImpl.class);
  private NetworkTopology cluster;
  private Node[] dataNodes;
  private Random random = new Random();

  public TestNetworkTopologyImpl(NodeSchema[] schemas, Node[] nodeArray) {
    NodeSchemaManager.getInstance().init(schemas, true);
    cluster = new NetworkTopologyImpl(NodeSchemaManager.getInstance());
    dataNodes = nodeArray.clone();
    for (int i = 0; i < dataNodes.length; i++) {
      cluster.add(dataNodes[i]);
    }
  }

  @Rule
  public Timeout testTimeout = new Timeout(3000000);

  @Parameters
  public static Collection<Object[]> setupDatanodes() {
    Object[][] topologies = new Object[][]{
        {new NodeSchema[] {ROOT_SCHEMA, LEAF_SCHEMA},
            new Node[]{
                createDatanode("1.1.1.1", "/"),
                createDatanode("2.2.2.2", "/"),
                createDatanode("3.3.3.3", "/"),
                createDatanode("4.4.4.4", "/"),
                createDatanode("5.5.5.5", "/"),
                createDatanode("6.6.6.6", "/"),
                createDatanode("7.7.7.7", "/"),
                createDatanode("8.8.8.8", "/"),
            }},
        {new NodeSchema[] {ROOT_SCHEMA, RACK_SCHEMA, LEAF_SCHEMA},
            new Node[]{
                createDatanode("1.1.1.1", "/r1"),
                createDatanode("2.2.2.2", "/r1"),
                createDatanode("3.3.3.3", "/r2"),
                createDatanode("4.4.4.4", "/r2"),
                createDatanode("5.5.5.5", "/r2"),
                createDatanode("6.6.6.6", "/r3"),
                createDatanode("7.7.7.7", "/r3"),
                createDatanode("8.8.8.8", "/r3"),
            }},
        {new NodeSchema[]
            {ROOT_SCHEMA, DATACENTER_SCHEMA, RACK_SCHEMA, LEAF_SCHEMA},
            new Node[]{
                createDatanode("1.1.1.1", "/d1/r1"),
                createDatanode("2.2.2.2", "/d1/r1"),
                createDatanode("3.3.3.3", "/d1/r2"),
                createDatanode("4.4.4.4", "/d1/r2"),
                createDatanode("5.5.5.5", "/d1/r2"),
                createDatanode("6.6.6.6", "/d2/r3"),
                createDatanode("7.7.7.7", "/d2/r3"),
                createDatanode("8.8.8.8", "/d2/r3"),
            }},
        {new NodeSchema[] {ROOT_SCHEMA, DATACENTER_SCHEMA, RACK_SCHEMA,
            NODEGROUP_SCHEMA, LEAF_SCHEMA},
            new Node[]{
                createDatanode("1.1.1.1", "/d1/r1/ng1"),
                createDatanode("2.2.2.2", "/d1/r1/ng1"),
                createDatanode("3.3.3.3", "/d1/r2/ng2"),
                createDatanode("4.4.4.4", "/d1/r2/ng2"),
                createDatanode("5.5.5.5", "/d1/r2/ng3"),
                createDatanode("6.6.6.6", "/d2/r3/ng3"),
                createDatanode("7.7.7.7", "/d2/r3/ng3"),
                createDatanode("8.8.8.8", "/d2/r3/ng3"),
                createDatanode("9.9.9.9", "/d3/r1/ng1"),
                createDatanode("10.10.10.10", "/d3/r1/ng1"),
                createDatanode("11.11.11.11", "/d3/r1/ng1"),
                createDatanode("12.12.12.12", "/d3/r2/ng2"),
                createDatanode("13.13.13.13", "/d3/r2/ng2"),
                createDatanode("14.14.14.14", "/d4/r1/ng1"),
                createDatanode("15.15.15.15", "/d4/r1/ng1"),
                createDatanode("16.16.16.16", "/d4/r1/ng1"),
                createDatanode("17.17.17.17", "/d4/r1/ng2"),
                createDatanode("18.18.18.18", "/d4/r1/ng2"),
                createDatanode("19.19.19.19", "/d4/r1/ng3"),
                createDatanode("20.20.20.20", "/d4/r1/ng3"),
            }},
        {new NodeSchema[] {ROOT_SCHEMA, REGION_SCHEMA, DATACENTER_SCHEMA,
            RACK_SCHEMA, NODEGROUP_SCHEMA, LEAF_SCHEMA},
            new Node[]{
                createDatanode("1.1.1.1", "/d1/rg1/r1/ng1"),
                createDatanode("2.2.2.2", "/d1/rg1/r1/ng1"),
                createDatanode("3.3.3.3", "/d1/rg1/r1/ng2"),
                createDatanode("4.4.4.4", "/d1/rg1/r1/ng1"),
                createDatanode("5.5.5.5", "/d1/rg1/r1/ng1"),
                createDatanode("6.6.6.6", "/d1/rg1/r1/ng2"),
                createDatanode("7.7.7.7", "/d1/rg1/r1/ng2"),
                createDatanode("8.8.8.8", "/d1/rg1/r1/ng2"),
                createDatanode("9.9.9.9", "/d1/rg1/r1/ng2"),
                createDatanode("10.10.10.10", "/d1/rg1/r1/ng2"),
                createDatanode("11.11.11.11", "/d1/rg1/r2/ng1"),
                createDatanode("12.12.12.12", "/d1/rg1/r2/ng1"),
                createDatanode("13.13.13.13", "/d1/rg1/r2/ng1"),
                createDatanode("14.14.14.14", "/d1/rg1/r2/ng1"),
                createDatanode("15.15.15.15", "/d1/rg1/r2/ng1"),
                createDatanode("16.16.16.16", "/d1/rg1/r2/ng2"),
                createDatanode("17.17.17.17", "/d1/rg1/r2/ng2"),
                createDatanode("18.18.18.18", "/d1/rg1/r2/ng2"),
                createDatanode("19.19.19.19", "/d1/rg1/r2/ng2"),
                createDatanode("20.20.20.20", "/d1/rg1/r2/ng2"),
                createDatanode("21.21.21.21", "/d2/rg1/r2/ng1"),
                createDatanode("22.22.22.22", "/d2/rg1/r2/ng1"),
                createDatanode("23.23.23.23", "/d2/rg2/r2/ng1"),
                createDatanode("24.24.24.24", "/d2/rg2/r2/ng1"),
                createDatanode("25.25.25.25", "/d2/rg2/r2/ng1"),
            }}
    };
    return Arrays.asList(topologies);
  }

  @Test
  public void testContains() {
    Node nodeNotInMap = createDatanode("8.8.8.8", "/d2/r4");
    for (int i=0; i < dataNodes.length; i++) {
      assertTrue(cluster.contains(dataNodes[i]));
    }
    assertFalse(cluster.contains(nodeNotInMap));
  }

  @Test
  public void testNumOfChildren() {
    assertEquals(dataNodes.length, cluster.getNumOfLeafNode(null));
    assertEquals(0, cluster.getNumOfLeafNode("/switch1/node1"));
  }

  @Test
  public void testGetNode() {
    assertEquals(cluster.getNode(""), cluster.getNode(null));
    assertEquals(cluster.getNode(""), cluster.getNode("/"));
    assertEquals(null, cluster.getNode("/switch1/node1"));
    assertEquals(null, cluster.getNode("/switch1"));
  }

  @Test
  public void testCreateInvalidTopology() {
    List<NodeSchema> schemas = new ArrayList<NodeSchema>();
    schemas.add(ROOT_SCHEMA);
    schemas.add(RACK_SCHEMA);
    schemas.add(LEAF_SCHEMA);
    NodeSchemaManager.getInstance().init(schemas.toArray(new NodeSchema[0]),
        true);
    NetworkTopology newCluster = new NetworkTopologyImpl(
        NodeSchemaManager.getInstance());
    Node[] invalidDataNodes = new Node[] {
        createDatanode("1.1.1.1", "/r1"),
        createDatanode("2.2.2.2", "/r2"),
        createDatanode("3.3.3.3", "/d1/r2")
    };
    newCluster.add(invalidDataNodes[0]);
    newCluster.add(invalidDataNodes[1]);
    try {
      newCluster.add(invalidDataNodes[2]);
      fail("expected InvalidTopologyException");
    } catch (NetworkTopology.InvalidTopologyException e) {
      assertTrue(e.getMessage().contains("Failed to add"));
      assertTrue(e.getMessage().contains("Its path depth is not " +
          newCluster.getMaxLevel()));
    }
  }

  @Test
  public void testInitWithConfigFile() {
    ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
    OzoneConfiguration conf = new OzoneConfiguration();
    try {
      String filePath = classLoader.getResource(
          "./networkTopologyTestFiles/good.xml").getPath();
      conf.set(ScmConfigKeys.OZONE_SCM_NETWORK_TOPOLOGY_SCHEMA_FILE, filePath);
      NetworkTopology newCluster = new NetworkTopologyImpl(conf);
      LOG.info("network topology max level = {}", newCluster.getMaxLevel());
    } catch (Throwable e) {
      fail("should succeed");
    }
  }

  @Test
  public void testAncestor() {
    assumeTrue(cluster.getMaxLevel() > 2);
    int maxLevel = cluster.getMaxLevel();
    assertTrue(cluster.isSameParent(dataNodes[0], dataNodes[1]));
    while(maxLevel > 1) {
      assertTrue(cluster.isSameAncestor(dataNodes[0], dataNodes[1],
          maxLevel - 1));
      maxLevel--;
    }
    assertFalse(cluster.isSameParent(dataNodes[1], dataNodes[2]));
    assertFalse(cluster.isSameParent(null, dataNodes[2]));
    assertFalse(cluster.isSameParent(dataNodes[1], null));
    assertFalse(cluster.isSameParent(null, null));

    assertFalse(cluster.isSameAncestor(dataNodes[1], dataNodes[2], 0));
    assertFalse(cluster.isSameAncestor(dataNodes[1], null, 1));
    assertFalse(cluster.isSameAncestor(null, dataNodes[2], 1));
    assertFalse(cluster.isSameAncestor(null, null, 1));

    maxLevel = cluster.getMaxLevel();
    assertTrue(cluster.isSameAncestor(
        dataNodes[random.nextInt(cluster.getNumOfLeafNode(null))],
        dataNodes[random.nextInt(cluster.getNumOfLeafNode(null))],
        maxLevel - 1));
  }

  @Test
  public void testAddRemove() {
    for(int i = 0; i < dataNodes.length; i++) {
      cluster.remove(dataNodes[i]);
    }
    for(int i = 0; i < dataNodes.length; i++) {
      assertFalse(cluster.contains(dataNodes[i]));
    }
    // no leaf nodes
    assertEquals(0, cluster.getNumOfLeafNode(null));
    // no inner nodes
    assertEquals(0, cluster.getNumOfNodes(2));
    for(int i = 0; i < dataNodes.length; i++) {
      cluster.add(dataNodes[i]);
    }
    // Inner nodes are created automatically
    assertTrue(cluster.getNumOfNodes(2) > 0);

    try {
      cluster.add(cluster.chooseRandom(null).getParent());
      fail("Inner node can not be added manually");
    } catch (Exception e) {
      assertTrue(e.getMessage().startsWith(
          "Not allowed to add an inner node"));
    }

    try {
      cluster.remove(cluster.chooseRandom(null).getParent());
      fail("Inner node can not be removed manually");
    } catch (Exception e) {
      assertTrue(e.getMessage().startsWith(
          "Not allowed to remove an inner node"));
    }
  }

  @Test
  public void testGetNodesWithLevel() {
    int maxLevel = cluster.getMaxLevel();
    try {
      assertEquals(1, cluster.getNumOfNodes(0));
      fail("level 0 is not supported");
    } catch (IllegalArgumentException e) {
      assertTrue(e.getMessage().startsWith("Invalid level"));
    }

    try {
      assertEquals(1, cluster.getNumOfNodes(0));
      fail("level 0 is not supported");
    } catch (IllegalArgumentException e) {
      assertTrue(e.getMessage().startsWith("Invalid level"));
    }

    try {
      assertEquals(1, cluster.getNumOfNodes(maxLevel + 1));
      fail("level out of scope");
    } catch (IllegalArgumentException e) {
      assertTrue(e.getMessage().startsWith("Invalid level"));
    }

    try {
      assertEquals(1, cluster.getNumOfNodes(maxLevel + 1));
      fail("level out of scope");
    } catch (IllegalArgumentException e) {
      assertTrue(e.getMessage().startsWith("Invalid level"));
    }
    // root node
    assertEquals(1, cluster.getNumOfNodes(1));
    assertEquals(1, cluster.getNumOfNodes(1));
    // leaf nodes
    assertEquals(dataNodes.length, cluster.getNumOfNodes(maxLevel));
    assertEquals(dataNodes.length, cluster.getNumOfNodes(maxLevel));
  }

  @Test
  public void testChooseRandomSimple() {
    String path =
        dataNodes[random.nextInt(dataNodes.length)].getNetworkFullPath();
    assertEquals(path, cluster.chooseRandom(path).getNetworkFullPath());
    path = path.substring(0, path.lastIndexOf(PATH_SEPARATOR_STR));
    // test chooseRandom(String scope)
    while (!path.equals(ROOT)) {
      assertTrue(cluster.chooseRandom(path).getNetworkLocation()
          .startsWith(path));
      Node node = cluster.chooseRandom("~" + path);
      assertTrue(!node.getNetworkLocation()
          .startsWith(path));
      path = path.substring(0,
          path.lastIndexOf(PATH_SEPARATOR_STR));
    }
    assertNotNull(cluster.chooseRandom(null));
    assertNotNull(cluster.chooseRandom(""));
    assertNotNull(cluster.chooseRandom("/"));
    assertNull(cluster.chooseRandom("~"));
    assertNull(cluster.chooseRandom("~/"));

    // test chooseRandom(String scope, String excludedScope)
    path = dataNodes[random.nextInt(dataNodes.length)].getNetworkFullPath();
    List<String> pathList = new ArrayList<>();
    pathList.add(path);
    assertNull(cluster.chooseRandom(path, pathList));
    assertNotNull(cluster.chooseRandom(null, pathList));
    assertNotNull(cluster.chooseRandom("", pathList));

    // test chooseRandom(String scope, Collection<Node> excludedNodes)
    assertNull(cluster.chooseRandom("", Arrays.asList(dataNodes)));
    assertNull(cluster.chooseRandom("/", Arrays.asList(dataNodes)));
    assertNull(cluster.chooseRandom("~", Arrays.asList(dataNodes)));
    assertNull(cluster.chooseRandom("~/", Arrays.asList(dataNodes)));
    assertNull(cluster.chooseRandom(null, Arrays.asList(dataNodes)));
  }

  /**
   * Following test checks that chooseRandom works for an excluded scope.
   */
  @Test
  public void testChooseRandomExcludedScope() {
    int[] excludedNodeIndexs = {0, dataNodes.length - 1,
        random.nextInt(dataNodes.length), random.nextInt(dataNodes.length)};
    String scope;
    Map<Node, Integer> frequency;
    for (int i : excludedNodeIndexs) {
      String path = dataNodes[i].getNetworkFullPath();
      while (!path.equals(ROOT)) {
        scope = "~" + path;
        frequency = pickNodesAtRandom(100, scope, null, 0);
        for (Node key : dataNodes) {
          if (key.getNetworkFullPath().startsWith(path)) {
            assertTrue(frequency.get(key) == 0);
          }
        }
        path = path.substring(0, path.lastIndexOf(PATH_SEPARATOR_STR));
      }
    }

    // null excludedScope, every node should be chosen
    frequency = pickNodes(100, null, null, null, 0);
    for (Node key : dataNodes) {
      assertTrue(frequency.get(key) != 0);
    }

    // "" excludedScope,  no node will ever be chosen
    List<String> pathList = new ArrayList();
    pathList.add("");
    frequency = pickNodes(100, pathList, null, null, 0);
    for (Node key : dataNodes) {
      assertTrue(frequency.get(key) == 0);
    }

    // "~" scope, no node will ever be chosen
    scope = "~";
    frequency = pickNodesAtRandom(100, scope, null, 0);
    for (Node key : dataNodes) {
      assertTrue(frequency.get(key) == 0);
    }
    // out network topology excluded scope, every node should be chosen
    pathList.clear();
    pathList.add("/city1");
    frequency = pickNodes(
        cluster.getNumOfLeafNode(null), pathList, null, null, 0);
    for (Node key : dataNodes) {
      assertTrue(frequency.get(key) != 0);
    }
  }

  /**
   * Following test checks that chooseRandom works for an excluded nodes.
   */
  @Test
  public void testChooseRandomExcludedNode() {
    Node[][] excludedNodeLists = {
        {},
        {dataNodes[0]},
        {dataNodes[dataNodes.length - 1]},
        {dataNodes[random.nextInt(dataNodes.length)]},
        {dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)]
        },
        {dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
        }};
    int leafNum = cluster.getNumOfLeafNode(null);
    Map<Node, Integer> frequency;
    for(Node[] list : excludedNodeLists) {
      List<Node> excludedList = Arrays.asList(list);
      int ancestorGen = 0;
      while(ancestorGen < cluster.getMaxLevel()) {
        frequency = pickNodesAtRandom(leafNum, null, excludedList, ancestorGen);
        List<Node> ancestorList = NetUtils.getAncestorList(cluster,
            excludedList, ancestorGen);
        for (Node key : dataNodes) {
          if (excludedList.contains(key) ||
              (ancestorList.size() > 0 &&
                  ancestorList.stream()
                      .map(a -> (InnerNode) a)
                      .filter(a -> a.isAncestor(key))
                      .collect(Collectors.toList()).size() > 0)) {
            assertTrue(frequency.get(key) == 0);
          }
        }
        ancestorGen++;
      }
    }
    // all nodes excluded, no node will be picked
    List<Node> excludedList = Arrays.asList(dataNodes);
    int ancestorGen = 0;
    while(ancestorGen < cluster.getMaxLevel()) {
      frequency = pickNodesAtRandom(leafNum, null, excludedList, ancestorGen);
      for (Node key : dataNodes) {
        assertTrue(frequency.get(key) == 0);
      }
      ancestorGen++;
    }
    // out scope excluded nodes, each node will be picked
    excludedList = Arrays.asList(createDatanode("1.1.1.1.", "/city1/rack1"));
    ancestorGen = 0;
    while(ancestorGen < cluster.getMaxLevel()) {
      frequency = pickNodes(leafNum, null, excludedList, null, ancestorGen);
      for (Node key : dataNodes) {
        assertTrue(frequency.get(key) != 0);
      }
      ancestorGen++;
    }
  }

  /**
   * Following test checks that chooseRandom works for excluded nodes and scope.
   */
  @Test
  public void testChooseRandomExcludedNodeAndScope() {
    int[] excludedNodeIndexs = {0, dataNodes.length - 1,
        random.nextInt(dataNodes.length), random.nextInt(dataNodes.length)};
    Node[][] excludedNodeLists = {
        {},
        {dataNodes[0]},
        {dataNodes[dataNodes.length - 1]},
        {dataNodes[random.nextInt(dataNodes.length)]},
        {dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)]
        },
        {dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
        }};
    int leafNum = cluster.getNumOfLeafNode(null);
    Map<Node, Integer> frequency;
    String scope;
    for (int i : excludedNodeIndexs) {
      String path = dataNodes[i].getNetworkFullPath();
      while (!path.equals(ROOT)) {
        scope = "~" + path;
        int ancestorGen = 0;
        while(ancestorGen < cluster.getMaxLevel()) {
          for (Node[] list : excludedNodeLists) {
            List<Node> excludedList = Arrays.asList(list);
            frequency =
                pickNodesAtRandom(leafNum, scope, excludedList, ancestorGen);
            List<Node> ancestorList = NetUtils.getAncestorList(cluster,
                excludedList, ancestorGen);
            for (Node key : dataNodes) {
              if (excludedList.contains(key) ||
                  key.getNetworkFullPath().startsWith(path) ||
                  (ancestorList.size() > 0 &&
                      ancestorList.stream()
                          .map(a -> (InnerNode) a)
                          .filter(a -> a.isAncestor(key))
                          .collect(Collectors.toList()).size() > 0)) {
                assertTrue(frequency.get(key) == 0);
              }
            }
          }
          ancestorGen++;
        }
        path = path.substring(0, path.lastIndexOf(PATH_SEPARATOR_STR));
      }
    }
    // all nodes excluded, no node will be picked
    List<Node> excludedList = Arrays.asList(dataNodes);
    for (int i : excludedNodeIndexs) {
      String path = dataNodes[i].getNetworkFullPath();
      while (!path.equals(ROOT)) {
        scope = "~" + path;
        int ancestorGen = 0;
        while (ancestorGen < cluster.getMaxLevel()) {
          frequency =
              pickNodesAtRandom(leafNum, scope, excludedList, ancestorGen);
          for (Node key : dataNodes) {
            assertTrue(frequency.get(key) == 0);
          }
          ancestorGen++;
        }
        path = path.substring(0, path.lastIndexOf(PATH_SEPARATOR_STR));
      }
    }

    // no node excluded and no excluded scope, each node will be picked
    int ancestorGen = 0;
    while (ancestorGen < cluster.getMaxLevel()) {
      frequency = pickNodes(leafNum, null, null, null, ancestorGen);
      for (Node key : dataNodes) {
        assertTrue(frequency.get(key) != 0);
      }
      ancestorGen++;
    }
  }

  /**
   * Following test checks that chooseRandom works for excluded nodes, scope
   * and ancestor generation.
   */
  @Test
  public void testChooseRandomWithAffinityNode() {
    int[] excludedNodeIndexs = {0, dataNodes.length - 1,
        random.nextInt(dataNodes.length), random.nextInt(dataNodes.length)};
    Node[][] excludedNodeLists = {
        {},
        {dataNodes[0]},
        {dataNodes[dataNodes.length - 1]},
        {dataNodes[random.nextInt(dataNodes.length)]},
        {dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)]
        },
        {dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
        }};
    int[] affinityNodeIndexs = {0, dataNodes.length - 1,
        random.nextInt(dataNodes.length), random.nextInt(dataNodes.length)};
    Node[][] excludedScopeIndexs = {{dataNodes[0]},
        {dataNodes[dataNodes.length - 1]},
        {dataNodes[random.nextInt(dataNodes.length)]},
        {dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)]
        },
        {dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
        }};
    int leafNum = cluster.getNumOfLeafNode(null);
    Map<Node, Integer> frequency;
    List<String> pathList = new ArrayList<>();
    for (int k : affinityNodeIndexs) {
      for (Node[] excludedScopes : excludedScopeIndexs) {
        pathList.clear();
        pathList.addAll(Arrays.stream(excludedScopes)
            .map(node -> node.getNetworkFullPath())
            .collect(Collectors.toList()));
        while (!pathList.get(0).equals(ROOT)) {
          int ancestorGen = cluster.getMaxLevel() - 1;
          while (ancestorGen > 0) {
            for (Node[] list : excludedNodeLists) {
              List<Node> excludedList = Arrays.asList(list);
              frequency = pickNodes(leafNum, pathList, excludedList,
                  dataNodes[k], ancestorGen);
              Node affinityAncestor = dataNodes[k].getAncestor(ancestorGen);
              for (Node key : dataNodes) {
                if (affinityAncestor != null) {
                  if (frequency.get(key) > 0) {
                    assertTrue(affinityAncestor.isAncestor(key));
                  } else if (!affinityAncestor.isAncestor(key)) {
                    continue;
                  } else if (excludedList != null &&
                      excludedList.contains(key)) {
                    continue;
                  } else if (pathList != null &&
                      pathList.stream().anyMatch(path ->
                          key.getNetworkFullPath().startsWith(path))) {
                    continue;
                  } else {
                    fail("Node is not picked when sequentially going " +
                        "through ancestor node's leaf nodes. node:" +
                        key.getNetworkFullPath() + ", ancestor node:" +
                        affinityAncestor.getNetworkFullPath() +
                        ", excludedScope: " + pathList.toString() + ", " +
                        "excludedList:" + excludedList.toString());
                  }
                }
              }
            }
            ancestorGen--;
          }
          pathList = pathList.stream().map(path ->
              path.substring(0, path.lastIndexOf(PATH_SEPARATOR_STR)))
              .collect(Collectors.toList());
        }
      }
    }

    // all nodes excluded, no node will be picked
    String scope;
    List<Node> excludedList = Arrays.asList(dataNodes);
    for (int k : affinityNodeIndexs) {
      for (int i : excludedNodeIndexs) {
        String path = dataNodes[i].getNetworkFullPath();
        while (!path.equals(ROOT)) {
          scope = "~" + path;
          int ancestorGen = 0;
          while (ancestorGen < cluster.getMaxLevel()) {
            frequency = pickNodesAtRandom(leafNum, scope, excludedList,
                dataNodes[k], ancestorGen);
            for (Node key : dataNodes) {
              assertTrue(frequency.get(key) == 0);
            }
            ancestorGen++;
          }
          path = path.substring(0, path.lastIndexOf(PATH_SEPARATOR_STR));
        }
      }
    }
    // no node excluded and no excluded scope, each node will be picked
    int ancestorGen = cluster.getMaxLevel() - 1;
    for (int k : affinityNodeIndexs) {
      while (ancestorGen > 0) {
        frequency =
            pickNodes(leafNum, null, null, dataNodes[k], ancestorGen);
        Node affinityAncestor = dataNodes[k].getAncestor(ancestorGen);
        for (Node key : dataNodes) {
          if (frequency.get(key) > 0) {
            if (affinityAncestor != null) {
              assertTrue(affinityAncestor.isAncestor(key));
            }
          }
        }
        ancestorGen--;
      }
    }
    // check invalid ancestor generation
    try {
      cluster.chooseRandom(null, null, null, dataNodes[0],
          cluster.getMaxLevel());
      fail("ancestor generation exceeds max level, should fail");
    } catch (Exception e) {
      assertTrue(e.getMessage().startsWith("ancestorGen " +
          cluster.getMaxLevel() +
          " exceeds this network topology acceptable level"));
    }
  }

  @Test
  public void testCost() {
    // network topology with default cost
    List<NodeSchema> schemas = new ArrayList<>();
    schemas.add(ROOT_SCHEMA);
    schemas.add(RACK_SCHEMA);
    schemas.add(NODEGROUP_SCHEMA);
    schemas.add(LEAF_SCHEMA);

    NodeSchemaManager manager = NodeSchemaManager.getInstance();
    manager.init(schemas.toArray(new NodeSchema[0]), true);
    NetworkTopology newCluster =
        new NetworkTopologyImpl(manager);
    Node[] nodeList = new Node[] {
        createDatanode("1.1.1.1", "/r1/ng1"),
        createDatanode("2.2.2.2", "/r1/ng1"),
        createDatanode("3.3.3.3", "/r1/ng2"),
        createDatanode("4.4.4.4", "/r2/ng1"),
    };
    for (Node node: nodeList) {
      newCluster.add(node);
    }
    Node outScopeNode1 = createDatanode("5.5.5.5", "/r2/ng2");
    Node outScopeNode2 = createDatanode("6.6.6.6", "/r2/ng2");
    assertEquals(Integer.MAX_VALUE,
        newCluster.getDistanceCost(nodeList[0], null));
    assertEquals(Integer.MAX_VALUE,
        newCluster.getDistanceCost(null, nodeList[0]));
    assertEquals(Integer.MAX_VALUE,
        newCluster.getDistanceCost(outScopeNode1, nodeList[0]));
    assertEquals(Integer.MAX_VALUE,
        newCluster.getDistanceCost(nodeList[0], outScopeNode1));
    assertEquals(Integer.MAX_VALUE,
        newCluster.getDistanceCost(outScopeNode1, outScopeNode2));

    assertEquals(0, newCluster.getDistanceCost(null, null));
    assertEquals(0, newCluster.getDistanceCost(nodeList[0], nodeList[0]));
    assertEquals(2, newCluster.getDistanceCost(nodeList[0], nodeList[1]));
    assertEquals(4, newCluster.getDistanceCost(nodeList[0], nodeList[2]));
    assertEquals(6, newCluster.getDistanceCost(nodeList[0], nodeList[3]));

    // network topology with customized cost
    schemas.clear();
    schemas.add(new NodeSchema.Builder()
        .setType(NodeSchema.LayerType.ROOT).setCost(5).build());
    schemas.add(new NodeSchema.Builder()
        .setType(NodeSchema.LayerType.INNER_NODE).setCost(3).build());
    schemas.add(new NodeSchema.Builder()
        .setType(NodeSchema.LayerType.INNER_NODE).setCost(1).build());
    schemas.add(new NodeSchema.Builder()
        .setType(NodeSchema.LayerType.LEAF_NODE).build());
    manager = NodeSchemaManager.getInstance();
    manager.init(schemas.toArray(new NodeSchema[0]), true);
    newCluster = new NetworkTopologyImpl(manager);
    for (Node node: nodeList) {
      newCluster.add(node);
    }
    assertEquals(Integer.MAX_VALUE,
        newCluster.getDistanceCost(nodeList[0], null));
    assertEquals(Integer.MAX_VALUE,
        newCluster.getDistanceCost(null, nodeList[0]));
    assertEquals(Integer.MAX_VALUE,
        newCluster.getDistanceCost(outScopeNode1, nodeList[0]));
    assertEquals(Integer.MAX_VALUE,
        newCluster.getDistanceCost(nodeList[0], outScopeNode1));
    assertEquals(Integer.MAX_VALUE,
        newCluster.getDistanceCost(outScopeNode1, outScopeNode2));

    assertEquals(0, newCluster.getDistanceCost(null, null));
    assertEquals(0, newCluster.getDistanceCost(nodeList[0], nodeList[0]));
    assertEquals(2, newCluster.getDistanceCost(nodeList[0], nodeList[1]));
    assertEquals(8, newCluster.getDistanceCost(nodeList[0], nodeList[2]));
    assertEquals(18, newCluster.getDistanceCost(nodeList[0], nodeList[3]));
  }

  @Test
  public void testSortByDistanceCost() {
    Node[][] nodes = {
        {},
        {dataNodes[0]},
        {dataNodes[dataNodes.length - 1]},
        {dataNodes[random.nextInt(dataNodes.length)]},
        {dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)]
        },
        {dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
        },
        {dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
        },
        {dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
            dataNodes[random.nextInt(dataNodes.length)],
        }};
    Node[] readers = {null, dataNodes[0], dataNodes[dataNodes.length - 1],
        dataNodes[random.nextInt(dataNodes.length)],
        dataNodes[random.nextInt(dataNodes.length)],
        dataNodes[random.nextInt(dataNodes.length)]
    };
    for (Node reader : readers) {
      for (Node[] nodeList : nodes) {
        int length = nodeList.length;
        while (length > 0) {
          List<? extends Node> ret = cluster.sortByDistanceCost(reader,
              Arrays.asList(nodeList), length);
          for (int i = 0; i < ret.size(); i++) {
            if ((i + 1) < ret.size()) {
              int cost1 = cluster.getDistanceCost(reader, ret.get(i));
              int cost2 = cluster.getDistanceCost(reader, ret.get(i + 1));
              assertTrue("reader:" + (reader != null ?
                  reader.getNetworkFullPath() : "null") +
                  ",node1:" + ret.get(i).getNetworkFullPath() +
                  ",node2:" + ret.get(i + 1).getNetworkFullPath() +
                  ",cost1:" + cost1 + ",cost2:" + cost2,
                  cost1 == Integer.MAX_VALUE || cost1 <= cost2);
            }
          }
          length--;
        }
      }
    }

    // sort all nodes
    List<Node> nodeList = Arrays.asList(dataNodes.clone());
    for (Node reader : readers) {
      int length = nodeList.size();
      while (length >= 0) {
        List<? extends Node> sortedNodeList =
            cluster.sortByDistanceCost(reader, nodeList, length);
        for (int i = 0; i < sortedNodeList.size(); i++) {
          if ((i + 1) < sortedNodeList.size()) {
            int cost1 = cluster.getDistanceCost(reader, sortedNodeList.get(i));
            int cost2 = cluster.getDistanceCost(
                reader, sortedNodeList.get(i + 1));
            // node can be removed when called in testConcurrentAccess
            assertTrue("reader:" + (reader != null ?
                reader.getNetworkFullPath() : "null") +
                ",node1:" + sortedNodeList.get(i).getNetworkFullPath() +
                ",node2:" + sortedNodeList.get(i + 1).getNetworkFullPath() +
                ",cost1:" + cost1 + ",cost2:" + cost2,
                cost1 == Integer.MAX_VALUE || cost1 <= cost2);
          }
        }
        length--;
      }
    }
  }

  private static Node createDatanode(String name, String path) {
    return new NodeImpl(name, path, NetConstants.NODE_COST_DEFAULT);
  }

  /**
   * This picks a large number of nodes at random in order to ensure coverage.
   *
   * @param numNodes the number of nodes
   * @param excludedScope the excluded scope
   * @param excludedNodes the excluded node list
   * @param ancestorGen the chosen node cannot share the same ancestor at
   *                    this generation with excludedNodes
   * @return the frequency that nodes were chosen
   */
  private Map<Node, Integer> pickNodesAtRandom(int numNodes,
      String excludedScope, Collection<Node> excludedNodes, int ancestorGen) {
    Map<Node, Integer> frequency = new HashMap<Node, Integer>();
    for (Node dnd : dataNodes) {
      frequency.put(dnd, 0);
    }
    for (int j = 0; j < numNodes; j++) {
      Node node = cluster.chooseRandom(excludedScope, excludedNodes,
          ancestorGen);
      if (node != null) {
        frequency.put(node, frequency.get(node) + 1);
      }
    }
    LOG.info("Result:{}", frequency);
    return frequency;
  }

  /**
   * This picks a large number of nodes at random in order to ensure coverage.
   *
   * @param numNodes the number of nodes
   * @param excludedScope the excluded scope
   * @param excludedNodes the excluded node list
   * @param affinityNode the chosen node should share the same ancestor at
   *                     generation "ancestorGen" with this node
   * @param ancestorGen  the chosen node cannot share the same ancestor at
   *                     this generation with excludedNodes
   * @return the frequency that nodes were chosen
   */
  private Map<Node, Integer> pickNodesAtRandom(int numNodes,
      String excludedScope, Collection<Node> excludedNodes, Node affinityNode,
      int ancestorGen) {
    Map<Node, Integer> frequency = new HashMap<Node, Integer>();
    for (Node dnd : dataNodes) {
      frequency.put(dnd, 0);
    }

    List<String> pathList = new ArrayList<>();
    pathList.add(excludedScope.substring(1));
    for (int j = 0; j < numNodes; j++) {

      Node node = cluster.chooseRandom("", pathList, excludedNodes,
          affinityNode, ancestorGen);
      if (node != null) {
        frequency.put(node, frequency.get(node) + 1);
      }
    }
    LOG.info("Result:{}", frequency);
    return frequency;
  }

  /**
   * This picks a large amount of nodes sequentially.
   *
   * @param numNodes the number of nodes
   * @param excludedScopes the excluded scopes, should not start with "~"
   * @param excludedNodes the excluded node list
   * @param affinityNode the chosen node should share the same ancestor at
   *                     generation "ancestorGen" with this node
   * @param ancestorGen  the chosen node cannot share the same ancestor at
   *                     this generation with excludedNodes
   * @return the frequency that nodes were chosen
   */
  private Map<Node, Integer> pickNodes(int numNodes,
      List<String> excludedScopes, Collection<Node> excludedNodes,
      Node affinityNode, int ancestorGen) {
    Map<Node, Integer> frequency = new HashMap<>();
    for (Node dnd : dataNodes) {
      frequency.put(dnd, 0);
    }
    excludedNodes = excludedNodes == null ? null :
        excludedNodes.stream().distinct().collect(Collectors.toList());
    for (int j = 0; j < numNodes; j++) {
      Node node = cluster.getNode(j, null, excludedScopes, excludedNodes,
          affinityNode, ancestorGen);
      if (node != null) {
        frequency.put(node, frequency.get(node) + 1);
      }
    }

    LOG.info("Result:{}", frequency);
    return frequency;
  }
}