hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From a..@apache.org
Subject hadoop git commit: HDFS-11482. Add storage type demand to into DFSNetworkTopology#chooseRandom. Contributed by Chen Liang.
Date Tue, 14 Mar 2017 00:31:28 GMT
Repository: hadoop
Updated Branches:
  refs/heads/trunk 55796a094 -> 9832ae0ed


HDFS-11482. Add storage type demand to into DFSNetworkTopology#chooseRandom. Contributed by Chen Liang.


Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/9832ae0e
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/9832ae0e
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/9832ae0e

Branch: refs/heads/trunk
Commit: 9832ae0ed8853d29072c9ea7031cd2373e6b16f9
Parents: 55796a0
Author: Chen Liang <cliang@apache.org>
Authored: Mon Mar 13 17:30:10 2017 -0700
Committer: Arpit Agarwal <arp@apache.org>
Committed: Mon Mar 13 17:30:10 2017 -0700

----------------------------------------------------------------------
 .../org/apache/hadoop/net/InnerNodeImpl.java    |   8 +-
 .../net/NetworkTopologyWithNodeGroup.java       |   2 +-
 .../hadoop/hdfs/net/DFSNetworkTopology.java     | 289 ++++++++++++
 .../hadoop/hdfs/net/DFSTopologyNodeImpl.java    | 275 ++++++++++++
 .../blockmanagement/DatanodeDescriptor.java     |   9 +
 .../apache/hadoop/hdfs/DFSNetworkTopology.java  |  36 --
 .../apache/hadoop/hdfs/DFSTopologyNodeImpl.java | 253 -----------
 .../hadoop/hdfs/TestDFSNetworkTopology.java     | 260 -----------
 .../hadoop/hdfs/net/TestDFSNetworkTopology.java | 449 +++++++++++++++++++
 9 files changed, 1027 insertions(+), 554 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/9832ae0e/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/net/InnerNodeImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/net/InnerNodeImpl.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/net/InnerNodeImpl.java
index 81eaf7f..5a2931b 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/net/InnerNodeImpl.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/net/InnerNodeImpl.java
@@ -63,7 +63,7 @@ public class InnerNodeImpl extends NodeBase implements InnerNode {
   /** Judge if this node represents a rack
    * @return true if it has no child or its children are not InnerNodes
    */
-  boolean isRack() {
+  public boolean isRack() {
     if (children.isEmpty()) {
       return true;
     }
@@ -81,7 +81,7 @@ public class InnerNodeImpl extends NodeBase implements InnerNode {
    * @param n a node
    * @return true if this node is an ancestor of <i>n</i>
    */
-  protected boolean isAncestor(Node n) {
+  public boolean isAncestor(Node n) {
     return getPath(this).equals(NodeBase.PATH_SEPARATOR_STR) ||
       (n.getNetworkLocation()+NodeBase.PATH_SEPARATOR_STR).
       startsWith(getPath(this)+NodeBase.PATH_SEPARATOR_STR);
@@ -92,12 +92,12 @@ public class InnerNodeImpl extends NodeBase implements InnerNode {
    * @param n a node
    * @return true if this node is the parent of <i>n</i>
    */
-  protected boolean isParent(Node n) {
+  public boolean isParent(Node n) {
     return n.getNetworkLocation().equals(getPath(this));
   }
 
   /* Return a child name of this node who is an ancestor of node <i>n</i> */
-  protected String getNextAncestorName(Node n) {
+  public String getNextAncestorName(Node n) {
     if (!isAncestor(n)) {
       throw new IllegalArgumentException(
                                          this + "is not an ancestor of " + n);

http://git-wip-us.apache.org/repos/asf/hadoop/blob/9832ae0e/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/net/NetworkTopologyWithNodeGroup.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/net/NetworkTopologyWithNodeGroup.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/net/NetworkTopologyWithNodeGroup.java
index a20d5fc..bec0fe1 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/net/NetworkTopologyWithNodeGroup.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/net/NetworkTopologyWithNodeGroup.java
@@ -308,7 +308,7 @@ public class NetworkTopologyWithNodeGroup extends NetworkTopology {
     }
 
     @Override
-    boolean isRack() {
+    public boolean isRack() {
       // it is node group
       if (getChildren().isEmpty()) {
         return false;

http://git-wip-us.apache.org/repos/asf/hadoop/blob/9832ae0e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/DFSNetworkTopology.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/DFSNetworkTopology.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/DFSNetworkTopology.java
new file mode 100644
index 0000000..ee83dba
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/DFSNetworkTopology.java
@@ -0,0 +1,289 @@
+/**
+ * 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.hdfs.net;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor;
+import org.apache.hadoop.net.NetworkTopology;
+import org.apache.hadoop.net.Node;
+import org.apache.hadoop.net.NodeBase;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Random;
+
+/**
+ * The HDFS specific network topology class. The main purpose of doing this
+ * subclassing is to add storage-type-aware chooseRandom method. All the
+ * remaining parts should be the same.
+ *
+ * Currently a placeholder to test storage type info.
+ * TODO : add "chooseRandom with storageType info" function.
+ */
+public class DFSNetworkTopology extends NetworkTopology {
+
+  private static final Random RANDOM = new Random();
+
+  public static DFSNetworkTopology getInstance(Configuration conf) {
+    DFSNetworkTopology nt = new DFSNetworkTopology();
+    return (DFSNetworkTopology)nt.init(DFSTopologyNodeImpl.FACTORY);
+  }
+
+  /**
+   * Randomly choose one node from <i>scope</i>, with specified storage type.
+   *
+   * If scope starts with ~, choose one from the all nodes except for the
+   * ones in <i>scope</i>; otherwise, choose one from <i>scope</i>.
+   * If excludedNodes is given, choose a node that's not in excludedNodes.
+   *
+   * @param scope range of nodes from which a node will be chosen
+   * @param excludedNodes nodes to be excluded from
+   * @return the chosen node
+   */
+  public Node chooseRandomWithStorageType(final String scope,
+      final Collection<Node> excludedNodes, StorageType type) {
+    netlock.readLock().lock();
+    try {
+      if (scope.startsWith("~")) {
+        return chooseRandomWithStorageType(
+            NodeBase.ROOT, scope.substring(1), excludedNodes, type);
+      } else {
+        return chooseRandomWithStorageType(
+            scope, null, excludedNodes, type);
+      }
+    } finally {
+      netlock.readLock().unlock();
+    }
+  }
+
+  /**
+   * Choose a random node based on given scope, excludedScope and excludedNodes
+   * set. Although in general the topology has at most three layers, this class
+   * will not impose such assumption.
+   *
+   * At high level, the idea is like this, say:
+   *
+   * R has two children A and B, and storage type is X, say:
+   * A has X = 6 (rooted at A there are 6 datanodes with X) and B has X = 8.
+   *
+   * Then R will generate a random int between 1~14, if it's <= 6, recursively
+   * call into A, otherwise B. This will maintain a uniformed randomness of
+   * choosing datanodes.
+   *
+   * The tricky part is how to handle excludes.
+   *
+   * For excludedNodes, since this set is small: currently the main reason of
+   * being an excluded node is because it already has a replica. So randomly
+   * picking up this node again should be rare. Thus we only check that, if the
+   * chosen node is excluded, we do chooseRandom again.
+   *
+   * For excludedScope, we locate the root of the excluded scope. Subtracting
+   * all it's ancestors' storage counters accordingly, this way the excluded
+   * root is out of the picture.
+   *
+   * TODO : this function has duplicate code as NetworkTopology, need to
+   * refactor in the future.
+   *
+   * @param scope
+   * @param excludedScope
+   * @param excludedNodes
+   * @return
+   */
+  @VisibleForTesting
+  Node chooseRandomWithStorageType(final String scope,
+      String excludedScope, final Collection<Node> excludedNodes,
+      StorageType type) {
+    if (excludedScope != null) {
+      if (scope.startsWith(excludedScope)) {
+        return null;
+      }
+      if (!excludedScope.startsWith(scope)) {
+        excludedScope = null;
+      }
+    }
+    Node node = getNode(scope);
+    if (node == null) {
+      LOG.debug("Invalid scope {}, non-existing node", scope);
+      return null;
+    }
+    if (!(node instanceof DFSTopologyNodeImpl)) {
+      // a node is either DFSTopologyNodeImpl, or a DatanodeDescriptor
+      return ((DatanodeDescriptor)node).hasStorageType(type) ? node : null;
+    }
+    DFSTopologyNodeImpl root = (DFSTopologyNodeImpl)node;
+    Node excludeRoot = excludedScope == null ? null : getNode(excludedScope);
+
+    // check to see if there are nodes satisfying the condition at all
+    int availableCount = root.getSubtreeStorageCount(type);
+    if (excludeRoot != null && root.isAncestor(excludeRoot)) {
+      if (excludeRoot instanceof DFSTopologyNodeImpl) {
+        availableCount -= ((DFSTopologyNodeImpl)excludeRoot)
+            .getSubtreeStorageCount(type);
+      } else {
+        availableCount -= ((DatanodeDescriptor)excludeRoot)
+            .hasStorageType(type) ? 1 : 0;
+      }
+    }
+    if (excludedNodes != null) {
+      for (Node excludedNode : excludedNodes) {
+        // all excluded nodes should be DatanodeDescriptor
+        Preconditions.checkArgument(excludedNode instanceof DatanodeDescriptor);
+        availableCount -= ((DatanodeDescriptor) excludedNode)
+            .hasStorageType(type) ? 1 : 0;
+      }
+    }
+    if (availableCount <= 0) {
+      // should never be <0 in general, adding <0 check for safety purpose
+      return null;
+    }
+    // to this point, it is guaranteed that there is at least one node
+    // that satisfies the requirement, keep trying until we found one.
+    Node chosen;
+    do {
+      chosen = chooseRandomWithStorageTypeAndExcludeRoot(root, excludeRoot,
+          type);
+      if (excludedNodes == null || !excludedNodes.contains(chosen)) {
+        break;
+      } else {
+        LOG.debug("Node {} is excluded, continuing.", chosen);
+      }
+    } while (true);
+    LOG.debug("chooseRandom returning {}", chosen);
+    return chosen;
+  }
+
+  /**
+   * Choose a random node that has the required storage type, under the given
+   * root, with an excluded subtree root (could also just be a leaf node).
+   *
+   * Note that excludedNode is checked after a random node, so it is not being
+   * handled here.
+   *
+   * @param root the root node where we start searching for a datanode
+   * @param excludeRoot the root of the subtree what should be excluded
+   * @param type the expected storage type
+   * @return a random datanode, with the storage type, and is not in excluded
+   * scope
+   */
+  private Node chooseRandomWithStorageTypeAndExcludeRoot(
+      DFSTopologyNodeImpl root, Node excludeRoot, StorageType type) {
+    Node chosenNode;
+    if (root.isRack()) {
+      // children are datanode descriptor
+      ArrayList<Node> candidates = new ArrayList<>();
+      for (Node node : root.getChildren()) {
+        if (node.equals(excludeRoot)) {
+          continue;
+        }
+        DatanodeDescriptor dnDescriptor = (DatanodeDescriptor)node;
+        if (dnDescriptor.hasStorageType(type)) {
+          candidates.add(node);
+        }
+      }
+      if (candidates.size() == 0) {
+        return null;
+      }
+      // to this point, all nodes in candidates are valid choices, and they are
+      // all datanodes, pick a random one.
+      chosenNode = candidates.get(RANDOM.nextInt(candidates.size()));
+    } else {
+      // the children are inner nodes
+      ArrayList<DFSTopologyNodeImpl> candidates =
+          getEligibleChildren(root, excludeRoot, type);
+      if (candidates.size() == 0) {
+        return null;
+      }
+      // again, all children are also inner nodes, we can do this cast.
+      // to maintain uniformality, the search needs to be based on the counts
+      // of valid datanodes. Below is a random weighted choose.
+      int totalCounts = 0;
+      int[] countArray = new int[candidates.size()];
+      for (int i = 0; i < candidates.size(); i++) {
+        DFSTopologyNodeImpl innerNode = candidates.get(i);
+        int subTreeCount = innerNode.getSubtreeStorageCount(type);
+        totalCounts += subTreeCount;
+        countArray[i] = subTreeCount;
+      }
+      // generate a random val between [1, totalCounts]
+      int randomCounts = RANDOM.nextInt(totalCounts) + 1;
+      int idxChosen = 0;
+      // searching for the idxChosen can potentially be done with binary
+      // search, but does not seem to worth it here.
+      for (int i = 0; i < countArray.length; i++) {
+        if (randomCounts <= countArray[i]) {
+          idxChosen = i;
+          break;
+        }
+        randomCounts -= countArray[i];
+      }
+      DFSTopologyNodeImpl nextRoot = candidates.get(idxChosen);
+      chosenNode = chooseRandomWithStorageTypeAndExcludeRoot(
+          nextRoot, excludeRoot, type);
+    }
+    return chosenNode;
+  }
+
+  /**
+   * Given root, excluded root and storage type. Find all the children of the
+   * root, that has the storage type available. One check is that if the
+   * excluded root is under a children, this children must subtract the storage
+   * count of the excluded root.
+   * @param root the subtree root we check.
+   * @param excludeRoot the root of the subtree that should be excluded.
+   * @param type the storage type we look for.
+   * @return a list of possible nodes, each of them is eligible as the next
+   * level root we search.
+   */
+  private ArrayList<DFSTopologyNodeImpl> getEligibleChildren(
+      DFSTopologyNodeImpl root, Node excludeRoot, StorageType type) {
+    ArrayList<DFSTopologyNodeImpl> candidates = new ArrayList<>();
+    int excludeCount = 0;
+    if (excludeRoot != null && root.isAncestor(excludeRoot)) {
+      // the subtree to be excluded is under the given root,
+      // find out the number of nodes to be excluded.
+      if (excludeRoot instanceof DFSTopologyNodeImpl) {
+        // if excludedRoot is an inner node, get the counts of all nodes on
+        // this subtree of that storage type.
+        excludeCount = ((DFSTopologyNodeImpl) excludeRoot)
+            .getSubtreeStorageCount(type);
+      } else {
+        // if excludedRoot is a datanode, simply ignore this one node
+        if (((DatanodeDescriptor) excludeRoot).hasStorageType(type)) {
+          excludeCount = 1;
+        }
+      }
+    }
+    // have calculated the number of storage counts to be excluded.
+    // walk through all children to check eligibility.
+    for (Node node : root.getChildren()) {
+      DFSTopologyNodeImpl dfsNode = (DFSTopologyNodeImpl) node;
+      int storageCount = dfsNode.getSubtreeStorageCount(type);
+      if (excludeRoot != null && excludeCount != 0 &&
+          (dfsNode.isAncestor(excludeRoot) || dfsNode.equals(excludeRoot))) {
+        storageCount -= excludeCount;
+      }
+      if (storageCount > 0) {
+        candidates.add(dfsNode);
+      }
+    }
+    return candidates;
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/9832ae0e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/DFSTopologyNodeImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/DFSTopologyNodeImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/DFSTopologyNodeImpl.java
new file mode 100644
index 0000000..92bb303
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/DFSTopologyNodeImpl.java
@@ -0,0 +1,275 @@
+/**
+ * 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.hdfs.net;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor;
+import org.apache.hadoop.net.InnerNode;
+import org.apache.hadoop.net.InnerNodeImpl;
+import org.apache.hadoop.net.Node;
+
+import java.util.EnumMap;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * The HDFS-specific representation of a network topology inner node. The
+ * difference is this class includes the information about the storage type
+ * info of this subtree. This info will be used when selecting subtrees
+ * in block placement.
+ */
+public class DFSTopologyNodeImpl extends InnerNodeImpl {
+
+  static final InnerNodeImpl.Factory FACTORY
+      = new DFSTopologyNodeImpl.Factory();
+
+  static final class Factory extends InnerNodeImpl.Factory {
+    private Factory() {}
+
+    @Override
+    public InnerNodeImpl newInnerNode(String path) {
+      return new DFSTopologyNodeImpl(path);
+    }
+  }
+
+  /**
+   * The core data structure of this class. The information about what storage
+   * types this subtree has. Basically, a map whose key is a child
+   * id, value is a enum map including the counts of each storage type. e.g.
+   * DISK type has count 5 means there are 5 leaf datanodes with DISK type
+   * available. This value is set/updated upon datanode joining and leaving.
+   *
+   * NOTE : It might be sufficient to keep only a map from storage type
+   * to count, omitting the child node id. But this might make it hard to keep
+   * consistency when there are updates from children.
+   *
+   * For example, if currently R has two children A and B with storage X, Y, and
+   * A : X=1 Y=1
+   * B : X=2 Y=2
+   * so we store X=3 Y=3 as total on R.
+   *
+   * Now say A has a new X plugged in and becomes X=2 Y=1.
+   *
+   * If we know that "A adds one X", it is easy to update R by +1 on X. However,
+   * if we don't know "A adds one X", but instead got "A now has X=2 Y=1",
+   * (which seems to be the case in current heartbeat) we will not know how to
+   * update R. While if we store on R "A has X=1 and Y=1" then we can simply
+   * update R by completely replacing the A entry and all will be good.
+   */
+  private final HashMap
+      <String, EnumMap<StorageType, Integer>> childrenStorageInfo;
+
+  DFSTopologyNodeImpl(String path) {
+    super(path);
+    childrenStorageInfo = new HashMap<>();
+  }
+
+  DFSTopologyNodeImpl(
+      String name, String location, InnerNode parent, int level) {
+    super(name, location, parent, level);
+    childrenStorageInfo = new HashMap<>();
+  }
+
+  public int getSubtreeStorageCount(StorageType type) {
+    int res = 0;
+    for (Map.Entry<String, EnumMap<StorageType, Integer>> entry :
+        childrenStorageInfo.entrySet()) {
+      if (entry.getValue().containsKey(type)) {
+        res += entry.getValue().get(type);
+      }
+    }
+    return res;
+  }
+
+  int getNumOfChildren() {
+    return children.size();
+  }
+
+  @Override
+  public boolean add(Node n) {
+    if (!isAncestor(n)) {
+      throw new IllegalArgumentException(n.getName()
+          + ", which is located at " + n.getNetworkLocation()
+          + ", is not a descendant of " + getPath(this));
+    }
+    // In HDFS topology, the leaf node should always be DatanodeDescriptor
+    if (!(n instanceof DatanodeDescriptor)) {
+      throw new IllegalArgumentException("Unexpected node type "
+          + n.getClass().getName());
+    }
+    DatanodeDescriptor dnDescriptor = (DatanodeDescriptor) n;
+    if (isParent(n)) {
+      // this node is the parent of n; add n directly
+      n.setParent(this);
+      n.setLevel(this.level + 1);
+      Node prev = childrenMap.put(n.getName(), n);
+      if (prev != null) {
+        for(int i=0; i<children.size(); i++) {
+          if (children.get(i).getName().equals(n.getName())) {
+            children.set(i, n);
+            return false;
+          }
+        }
+      }
+      children.add(n);
+      numOfLeaves++;
+      synchronized (childrenStorageInfo) {
+        if (!childrenStorageInfo.containsKey(dnDescriptor.getName())) {
+          childrenStorageInfo.put(
+              dnDescriptor.getName(), new EnumMap<>(StorageType.class));
+        }
+        for (StorageType st : dnDescriptor.getStorageTypes()) {
+          childrenStorageInfo.get(dnDescriptor.getName()).put(st, 1);
+        }
+      }
+      return true;
+    } else {
+      // find the next ancestor node
+      String parentName = getNextAncestorName(n);
+      InnerNode parentNode = (InnerNode)childrenMap.get(parentName);
+      if (parentNode == null) {
+        // create a new InnerNode
+        parentNode = createParentNode(parentName);
+        children.add(parentNode);
+        childrenMap.put(parentNode.getName(), parentNode);
+      }
+      // add n to the subtree of the next ancestor node
+      if (parentNode.add(n)) {
+        numOfLeaves++;
+        synchronized (childrenStorageInfo) {
+          if (!childrenStorageInfo.containsKey(parentNode.getName())) {
+            childrenStorageInfo.put(
+                parentNode.getName(), new EnumMap<>(StorageType.class));
+            for (StorageType st : dnDescriptor.getStorageTypes()) {
+              childrenStorageInfo.get(parentNode.getName()).put(st, 1);
+            }
+          } else {
+            EnumMap<StorageType, Integer> currentCount =
+                childrenStorageInfo.get(parentNode.getName());
+            for (StorageType st : dnDescriptor.getStorageTypes()) {
+              if (currentCount.containsKey(st)) {
+                currentCount.put(st, currentCount.get(st) + 1);
+              } else {
+                currentCount.put(st, 1);
+              }
+            }
+          }
+        }
+        return true;
+      } else {
+        return false;
+      }
+    }
+  }
+
+  @VisibleForTesting
+  HashMap <String, EnumMap<StorageType, Integer>> getChildrenStorageInfo() {
+    return childrenStorageInfo;
+  }
+
+
+  private DFSTopologyNodeImpl createParentNode(String parentName) {
+    return new DFSTopologyNodeImpl(
+        parentName, getPath(this), this, this.getLevel() + 1);
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    return super.equals(o);
+  }
+
+  @Override
+  public int hashCode() {
+    return super.hashCode();
+  }
+
+  @Override
+  public boolean remove(Node n) {
+    if (!isAncestor(n)) {
+      throw new IllegalArgumentException(n.getName()
+          + ", which is located at " + n.getNetworkLocation()
+          + ", is not a descendant of " + getPath(this));
+    }
+    // In HDFS topology, the leaf node should always be DatanodeDescriptor
+    if (!(n instanceof DatanodeDescriptor)) {
+      throw new IllegalArgumentException("Unexpected node type "
+          + n.getClass().getName());
+    }
+    DatanodeDescriptor dnDescriptor = (DatanodeDescriptor) n;
+    if (isParent(n)) {
+      // this node is the parent of n; remove n directly
+      if (childrenMap.containsKey(n.getName())) {
+        for (int i=0; i<children.size(); i++) {
+          if (children.get(i).getName().equals(n.getName())) {
+            children.remove(i);
+            childrenMap.remove(n.getName());
+            synchronized (childrenStorageInfo) {
+              childrenStorageInfo.remove(dnDescriptor.getName());
+            }
+            numOfLeaves--;
+            n.setParent(null);
+            return true;
+          }
+        }
+      }
+      return false;
+    } else {
+      // find the next ancestor node: the parent node
+      String parentName = getNextAncestorName(n);
+      DFSTopologyNodeImpl parentNode =
+          (DFSTopologyNodeImpl)childrenMap.get(parentName);
+      if (parentNode == null) {
+        return false;
+      }
+      // remove n from the parent node
+      boolean isRemoved = parentNode.remove(n);
+      if (isRemoved) {
+        // if the parent node has no children, remove the parent node too
+        synchronized (childrenStorageInfo) {
+          EnumMap<StorageType, Integer> currentCount =
+              childrenStorageInfo.get(parentNode.getName());
+          EnumSet<StorageType> toRemove = EnumSet.noneOf(StorageType.class);
+          for (StorageType st : dnDescriptor.getStorageTypes()) {
+            int newCount = currentCount.get(st) - 1;
+            if (newCount == 0) {
+              toRemove.add(st);
+            }
+            currentCount.put(st, newCount);
+          }
+          for (StorageType st : toRemove) {
+            currentCount.remove(st);
+          }
+        }
+        if (parentNode.getNumOfChildren() == 0) {
+          for(int i=0; i < children.size(); i++) {
+            if (children.get(i).getName().equals(parentName)) {
+              children.remove(i);
+              childrenMap.remove(parentName);
+              childrenStorageInfo.remove(parentNode.getName());
+              break;
+            }
+          }
+        }
+        numOfLeaves--;
+      }
+      return isRemoved;
+    }
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hadoop/blob/9832ae0e/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java
index 7b3e4e1..53d272a 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java
@@ -941,5 +941,14 @@ public class DatanodeDescriptor extends DatanodeInfo {
   public boolean isRegistered() {
     return isAlive() && !forceRegistration;
   }
+
+  public boolean hasStorageType(StorageType type) {
+    for (DatanodeStorageInfo dnStorage : getStorageInfos()) {
+      if (dnStorage.getStorageType() == type) {
+        return true;
+      }
+    }
+    return false;
+  }
 }
 

http://git-wip-us.apache.org/repos/asf/hadoop/blob/9832ae0e/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/DFSNetworkTopology.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/DFSNetworkTopology.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/DFSNetworkTopology.java
deleted file mode 100644
index a6b8c00..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/DFSNetworkTopology.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/**
- * 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.hdfs;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.net.NetworkTopology;
-
-/**
- * The HDFS specific network topology class. The main purpose of doing this
- * subclassing is to add storage-type-aware chooseRandom method. All the
- * remaining parts should be the same.
- *
- * Currently a placeholder to test storage type info.
- * TODO : add "chooseRandom with storageType info" function.
- */
-public class DFSNetworkTopology extends NetworkTopology {
-  public static DFSNetworkTopology getInstance(Configuration conf) {
-    DFSNetworkTopology nt = new DFSNetworkTopology();
-    return (DFSNetworkTopology)nt.init(DFSTopologyNodeImpl.FACTORY);
-  }
-}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/9832ae0e/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/DFSTopologyNodeImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/DFSTopologyNodeImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/DFSTopologyNodeImpl.java
deleted file mode 100644
index e746823..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/DFSTopologyNodeImpl.java
+++ /dev/null
@@ -1,253 +0,0 @@
-/**
- * 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.hdfs;
-
-import com.google.common.annotations.VisibleForTesting;
-import org.apache.hadoop.fs.StorageType;
-import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor;
-import org.apache.hadoop.net.InnerNode;
-import org.apache.hadoop.net.InnerNodeImpl;
-import org.apache.hadoop.net.Node;
-
-import java.util.EnumMap;
-import java.util.EnumSet;
-import java.util.HashMap;
-
-/**
- * The HDFS-specific representation of a network topology inner node. The
- * difference is this class includes the information about the storage type
- * info of this subtree. This info will be used when selecting subtrees
- * in block placement.
- */
-public class DFSTopologyNodeImpl extends InnerNodeImpl {
-
-  static final InnerNodeImpl.Factory FACTORY
-      = new DFSTopologyNodeImpl.Factory();
-
-  static final class Factory extends InnerNodeImpl.Factory {
-    private Factory() {}
-
-    @Override
-    public InnerNodeImpl newInnerNode(String path) {
-      return new DFSTopologyNodeImpl(path);
-    }
-  }
-
-  /**
-   * The core data structure of this class. The information about what storage
-   * types this subtree has. Basically, a map whose key is a child
-   * id, value is a enum map including the counts of each storage type. e.g.
-   * DISK type has count 5 means there are 5 leaf datanodes with DISK type
-   * available. This value is set/updated upon datanode joining and leaving.
-   *
-   * NOTE : It might be sufficient to keep only a map from storage type
-   * to count, omitting the child node id. But this might make it hard to keep
-   * consistency when there are updates from children.
-   *
-   * For example, if currently R has two children A and B with storage X, Y, and
-   * A : X=1 Y=1
-   * B : X=2 Y=2
-   * so we store X=3 Y=3 as total on R.
-   *
-   * Now say A has a new X plugged in and becomes X=2 Y=1.
-   *
-   * If we know that "A adds one X", it is easy to update R by +1 on X. However,
-   * if we don't know "A adds one X", but instead got "A now has X=2 Y=1",
-   * (which seems to be the case in current heartbeat) we will not know how to
-   * update R. While if we store on R "A has X=1 and Y=1" then we can simply
-   * update R by completely replacing the A entry and all will be good.
-   */
-  private final HashMap
-      <String, EnumMap<StorageType, Integer>> childrenStorageInfo;
-
-  DFSTopologyNodeImpl(String path) {
-    super(path);
-    childrenStorageInfo = new HashMap<>();
-  }
-
-  DFSTopologyNodeImpl(
-      String name, String location, InnerNode parent, int level) {
-    super(name, location, parent, level);
-    childrenStorageInfo = new HashMap<>();
-  }
-
-  int getNumOfChildren() {
-    return children.size();
-  }
-
-  @Override
-  public boolean add(Node n) {
-    if (!isAncestor(n)) {
-      throw new IllegalArgumentException(n.getName()
-          + ", which is located at " + n.getNetworkLocation()
-          + ", is not a descendant of " + getPath(this));
-    }
-    // In HDFS topology, the leaf node should always be DatanodeDescriptor
-    if (!(n instanceof DatanodeDescriptor)) {
-      throw new IllegalArgumentException("Unexpected node type "
-          + n.getClass().getName());
-    }
-    DatanodeDescriptor dnDescriptor = (DatanodeDescriptor) n;
-    if (isParent(n)) {
-      // this node is the parent of n; add n directly
-      n.setParent(this);
-      n.setLevel(this.level + 1);
-      Node prev = childrenMap.put(n.getName(), n);
-      if (prev != null) {
-        for(int i=0; i<children.size(); i++) {
-          if (children.get(i).getName().equals(n.getName())) {
-            children.set(i, n);
-            return false;
-          }
-        }
-      }
-      children.add(n);
-      numOfLeaves++;
-      synchronized (childrenStorageInfo) {
-        if (!childrenStorageInfo.containsKey(dnDescriptor.getName())) {
-          childrenStorageInfo.put(
-              dnDescriptor.getName(), new EnumMap<>(StorageType.class));
-        }
-        for (StorageType st : dnDescriptor.getStorageTypes()) {
-          childrenStorageInfo.get(dnDescriptor.getName()).put(st, 1);
-        }
-      }
-      return true;
-    } else {
-      // find the next ancestor node
-      String parentName = getNextAncestorName(n);
-      InnerNode parentNode = (InnerNode)childrenMap.get(parentName);
-      if (parentNode == null) {
-        // create a new InnerNode
-        parentNode = createParentNode(parentName);
-        children.add(parentNode);
-        childrenMap.put(parentNode.getName(), parentNode);
-      }
-      // add n to the subtree of the next ancestor node
-      if (parentNode.add(n)) {
-        numOfLeaves++;
-        synchronized (childrenStorageInfo) {
-          if (!childrenStorageInfo.containsKey(parentNode.getName())) {
-            childrenStorageInfo.put(
-                parentNode.getName(), new EnumMap<>(StorageType.class));
-            for (StorageType st : dnDescriptor.getStorageTypes()) {
-              childrenStorageInfo.get(parentNode.getName()).put(st, 1);
-            }
-          } else {
-            EnumMap<StorageType, Integer> currentCount =
-                childrenStorageInfo.get(parentNode.getName());
-            for (StorageType st : dnDescriptor.getStorageTypes()) {
-              if (currentCount.containsKey(st)) {
-                currentCount.put(st, currentCount.get(st) + 1);
-              } else {
-                currentCount.put(st, 1);
-              }
-            }
-          }
-        }
-        return true;
-      } else {
-        return false;
-      }
-    }
-  }
-
-  @VisibleForTesting
-  HashMap <String, EnumMap<StorageType, Integer>> getChildrenStorageInfo() {
-    return childrenStorageInfo;
-  }
-
-
-  private DFSTopologyNodeImpl createParentNode(String parentName) {
-    return new DFSTopologyNodeImpl(
-        parentName, getPath(this), this, this.getLevel() + 1);
-  }
-
-  @Override
-  public boolean remove(Node n) {
-    if (!isAncestor(n)) {
-      throw new IllegalArgumentException(n.getName()
-          + ", which is located at " + n.getNetworkLocation()
-          + ", is not a descendant of " + getPath(this));
-    }
-    // In HDFS topology, the leaf node should always be DatanodeDescriptor
-    if (!(n instanceof DatanodeDescriptor)) {
-      throw new IllegalArgumentException("Unexpected node type "
-          + n.getClass().getName());
-    }
-    DatanodeDescriptor dnDescriptor = (DatanodeDescriptor) n;
-    if (isParent(n)) {
-      // this node is the parent of n; remove n directly
-      if (childrenMap.containsKey(n.getName())) {
-        for (int i=0; i<children.size(); i++) {
-          if (children.get(i).getName().equals(n.getName())) {
-            children.remove(i);
-            childrenMap.remove(n.getName());
-            synchronized (childrenStorageInfo) {
-              childrenStorageInfo.remove(dnDescriptor.getName());
-            }
-            numOfLeaves--;
-            n.setParent(null);
-            return true;
-          }
-        }
-      }
-      return false;
-    } else {
-      // find the next ancestor node: the parent node
-      String parentName = getNextAncestorName(n);
-      DFSTopologyNodeImpl parentNode =
-          (DFSTopologyNodeImpl)childrenMap.get(parentName);
-      if (parentNode == null) {
-        return false;
-      }
-      // remove n from the parent node
-      boolean isRemoved = parentNode.remove(n);
-      if (isRemoved) {
-        // if the parent node has no children, remove the parent node too
-        synchronized (childrenStorageInfo) {
-          EnumMap<StorageType, Integer> currentCount =
-              childrenStorageInfo.get(parentNode.getName());
-          EnumSet<StorageType> toRemove = EnumSet.noneOf(StorageType.class);
-          for (StorageType st : dnDescriptor.getStorageTypes()) {
-            int newCount = currentCount.get(st) - 1;
-            if (newCount == 0) {
-              toRemove.add(st);
-            }
-            currentCount.put(st, newCount);
-          }
-          for (StorageType st : toRemove) {
-            currentCount.remove(st);
-          }
-        }
-        if (parentNode.getNumOfChildren() == 0) {
-          for(int i=0; i < children.size(); i++) {
-            if (children.get(i).getName().equals(parentName)) {
-              children.remove(i);
-              childrenMap.remove(parentName);
-              childrenStorageInfo.remove(parentNode.getName());
-              break;
-            }
-          }
-        }
-        numOfLeaves--;
-      }
-      return isRemoved;
-    }
-  }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hadoop/blob/9832ae0e/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSNetworkTopology.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSNetworkTopology.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSNetworkTopology.java
deleted file mode 100644
index ac1edf9..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSNetworkTopology.java
+++ /dev/null
@@ -1,260 +0,0 @@
-/**
- * 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.hdfs;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.StorageType;
-import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor;
-import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeStorageInfo;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.Timeout;
-
-import java.util.EnumMap;
-import java.util.HashMap;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-
-/**
- * This class tests the correctness of storage type info stored in
- * DFSNetworkTopology.
- */
-public class TestDFSNetworkTopology {
-  private static final Log LOG =
-      LogFactory.getLog(TestDFSNetworkTopology.class);
-  private final static DFSNetworkTopology CLUSTER =
-      DFSNetworkTopology.getInstance(new Configuration());
-  private DatanodeDescriptor[] dataNodes;
-
-  @Rule
-  public Timeout testTimeout = new Timeout(30000);
-
-  @Before
-  public void setupDatanodes() {
-    final String[] racks = {
-        "/l1/d1/r1", "/l1/d1/r1", "/l1/d1/r2", "/l1/d1/r2", "/l1/d1/r2",
-
-        "/l1/d2/r3", "/l1/d2/r3", "/l1/d2/r3",
-
-        "/l2/d3/r1", "/l2/d3/r2", "/l2/d3/r3", "/l2/d3/r4", "/l2/d3/r5",
-
-        "/l2/d4/r1", "/l2/d4/r1", "/l2/d4/r1", "/l2/d4/r1", "/l2/d4/r1",
-        "/l2/d4/r1", "/l2/d4/r1"};
-    final String[] hosts = {
-        "host1", "host2", "host3", "host4", "host5",
-        "host6", "host7", "host8", "host9", "host10",
-        "host11", "host12", "host13", "host14", "host15",
-        "host16", "host17", "host18", "host19", "host20"};
-    final StorageType[] types = {
-        StorageType.ARCHIVE, StorageType.DISK, StorageType.ARCHIVE,
-        StorageType.DISK, StorageType.DISK,
-
-        StorageType.DISK, StorageType.RAM_DISK, StorageType.SSD,
-
-        StorageType.DISK, StorageType.RAM_DISK, StorageType.DISK,
-        StorageType.ARCHIVE, StorageType.ARCHIVE,
-
-        StorageType.DISK, StorageType.DISK, StorageType.RAM_DISK,
-        StorageType.RAM_DISK, StorageType.ARCHIVE, StorageType.ARCHIVE,
-        StorageType.SSD};
-    final DatanodeStorageInfo[] storages =
-        DFSTestUtil.createDatanodeStorageInfos(20, racks, hosts, types);
-    dataNodes = DFSTestUtil.toDatanodeDescriptor(storages);
-    for (int i = 0; i < dataNodes.length; i++) {
-      CLUSTER.add(dataNodes[i]);
-    }
-    dataNodes[9].setDecommissioned();
-    dataNodes[10].setDecommissioned();
-  }
-
-  /**
-   * Test getting the storage type info of subtree.
-   * @throws Exception
-   */
-  @Test
-  public void testGetStorageTypeInfo() throws Exception {
-    // checking level = 2 nodes
-    DFSTopologyNodeImpl d1 =
-        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1/d1");
-    HashMap<String, EnumMap<StorageType, Integer>> d1info =
-        d1.getChildrenStorageInfo();
-    assertEquals(2, d1info.keySet().size());
-    assertTrue(d1info.get("r1").size() == 2 && d1info.get("r2").size() == 2);
-    assertEquals(1, (int)d1info.get("r1").get(StorageType.DISK));
-    assertEquals(1, (int)d1info.get("r1").get(StorageType.ARCHIVE));
-    assertEquals(2, (int)d1info.get("r2").get(StorageType.DISK));
-    assertEquals(1, (int)d1info.get("r2").get(StorageType.ARCHIVE));
-
-    DFSTopologyNodeImpl d2 =
-        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1/d2");
-    HashMap<String, EnumMap<StorageType, Integer>> d2info =
-        d2.getChildrenStorageInfo();
-    assertEquals(1, d2info.keySet().size());
-    assertTrue(d2info.get("r3").size() == 3);
-    assertEquals(1, (int)d2info.get("r3").get(StorageType.DISK));
-    assertEquals(1, (int)d2info.get("r3").get(StorageType.RAM_DISK));
-    assertEquals(1, (int)d2info.get("r3").get(StorageType.SSD));
-
-    DFSTopologyNodeImpl d3 =
-        (DFSTopologyNodeImpl) CLUSTER.getNode("/l2/d3");
-    HashMap<String, EnumMap<StorageType, Integer>> d3info =
-        d3.getChildrenStorageInfo();
-    assertEquals(5, d3info.keySet().size());
-    assertEquals(1, (int)d3info.get("r1").get(StorageType.DISK));
-    assertEquals(1, (int)d3info.get("r2").get(StorageType.RAM_DISK));
-    assertEquals(1, (int)d3info.get("r3").get(StorageType.DISK));
-    assertEquals(1, (int)d3info.get("r4").get(StorageType.ARCHIVE));
-    assertEquals(1, (int)d3info.get("r5").get(StorageType.ARCHIVE));
-
-    DFSTopologyNodeImpl d4 =
-        (DFSTopologyNodeImpl) CLUSTER.getNode("/l2/d4");
-    HashMap<String, EnumMap<StorageType, Integer>> d4info =
-        d4.getChildrenStorageInfo();
-    assertEquals(1, d4info.keySet().size());
-    assertEquals(2, (int)d4info.get("r1").get(StorageType.DISK));
-    assertEquals(2, (int)d4info.get("r1").get(StorageType.RAM_DISK));
-    assertEquals(2, (int)d4info.get("r1").get(StorageType.ARCHIVE));
-    assertEquals(1, (int)d4info.get("r1").get(StorageType.SSD));
-
-    DFSTopologyNodeImpl l1 =
-        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1");
-    HashMap<String, EnumMap<StorageType, Integer>> l1info =
-        l1.getChildrenStorageInfo();
-    assertEquals(2, l1info.keySet().size());
-    assertTrue(l1info.get("d1").size() == 2
-        && l1info.get("d2").size() == 3);
-    assertEquals(2, (int)l1info.get("d1").get(StorageType.ARCHIVE));
-    assertEquals(3, (int)l1info.get("d1").get(StorageType.DISK));
-    assertEquals(1, (int)l1info.get("d2").get(StorageType.DISK));
-    assertEquals(1, (int)l1info.get("d2").get(StorageType.RAM_DISK));
-    assertEquals(1, (int)l1info.get("d2").get(StorageType.SSD));
-
-    // checking level = 1 nodes
-    DFSTopologyNodeImpl l2 =
-        (DFSTopologyNodeImpl) CLUSTER.getNode("/l2");
-    HashMap<String, EnumMap<StorageType, Integer>> l2info =
-        l2.getChildrenStorageInfo();
-    assertTrue(l2info.get("d3").size() == 3
-        && l2info.get("d4").size() == 4);
-    assertEquals(2, l2info.keySet().size());
-    assertEquals(2, (int)l2info.get("d3").get(StorageType.DISK));
-    assertEquals(2, (int)l2info.get("d3").get(StorageType.ARCHIVE));
-    assertEquals(1, (int)l2info.get("d3").get(StorageType.RAM_DISK));
-    assertEquals(2, (int)l2info.get("d4").get(StorageType.DISK));
-    assertEquals(2, (int)l2info.get("d4").get(StorageType.ARCHIVE));
-    assertEquals(2, (int)l2info.get("d4").get(StorageType.RAM_DISK));
-    assertEquals(1, (int)l2info.get("d4").get(StorageType.SSD));
-  }
-
-  /**
-   * Test the correctness of storage type info when nodes are added and removed.
-   * @throws Exception
-   */
-  @Test
-  public void testAddAndRemoveTopology() throws Exception {
-    String[] newRack = {"/l1/d1/r1", "/l1/d1/r3", "/l1/d3/r3", "/l1/d3/r3"};
-    String[] newHost = {"nhost1", "nhost2", "nhost3", "nhost4"};
-    String[] newips = {"30.30.30.30", "31.31.31.31", "32.32.32.32",
-        "33.33.33.33"};
-    StorageType[] newTypes = {StorageType.DISK, StorageType.SSD,
-        StorageType.SSD, StorageType.SSD};
-    DatanodeDescriptor[] newDD = new DatanodeDescriptor[4];
-
-    for (int i = 0; i<4; i++) {
-      DatanodeStorageInfo dsi = DFSTestUtil.createDatanodeStorageInfo(
-          "s" + newHost[i], newips[i], newRack[i], newHost[i],
-          newTypes[i], null);
-      newDD[i] = dsi.getDatanodeDescriptor();
-      CLUSTER.add(newDD[i]);
-    }
-
-    DFSTopologyNodeImpl d1 =
-        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1/d1");
-    HashMap<String, EnumMap<StorageType, Integer>> d1info =
-        d1.getChildrenStorageInfo();
-    assertEquals(3, d1info.keySet().size());
-    assertTrue(d1info.get("r1").size() == 2 && d1info.get("r2").size() == 2
-      && d1info.get("r3").size() == 1);
-    assertEquals(2, (int)d1info.get("r1").get(StorageType.DISK));
-    assertEquals(1, (int)d1info.get("r1").get(StorageType.ARCHIVE));
-    assertEquals(2, (int)d1info.get("r2").get(StorageType.DISK));
-    assertEquals(1, (int)d1info.get("r2").get(StorageType.ARCHIVE));
-    assertEquals(1, (int)d1info.get("r3").get(StorageType.SSD));
-
-    DFSTopologyNodeImpl d3 =
-        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1/d3");
-    HashMap<String, EnumMap<StorageType, Integer>> d3info =
-        d3.getChildrenStorageInfo();
-    assertEquals(1, d3info.keySet().size());
-    assertTrue(d3info.get("r3").size() == 1);
-    assertEquals(2, (int)d3info.get("r3").get(StorageType.SSD));
-
-    DFSTopologyNodeImpl l1 =
-        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1");
-    HashMap<String, EnumMap<StorageType, Integer>> l1info =
-        l1.getChildrenStorageInfo();
-    assertEquals(3, l1info.keySet().size());
-    assertTrue(l1info.get("d1").size() == 3 &&
-        l1info.get("d2").size() == 3 && l1info.get("d3").size() == 1);
-    assertEquals(4, (int)l1info.get("d1").get(StorageType.DISK));
-    assertEquals(2, (int)l1info.get("d1").get(StorageType.ARCHIVE));
-    assertEquals(1, (int)l1info.get("d1").get(StorageType.SSD));
-    assertEquals(1, (int)l1info.get("d2").get(StorageType.SSD));
-    assertEquals(1, (int)l1info.get("d2").get(StorageType.RAM_DISK));
-    assertEquals(1, (int)l1info.get("d2").get(StorageType.DISK));
-    assertEquals(2, (int)l1info.get("d3").get(StorageType.SSD));
-
-
-    for (int i = 0; i<4; i++) {
-      CLUSTER.remove(newDD[i]);
-    }
-
-    // /d1/r3 should've been out, /d1/r1 should've been resumed
-    DFSTopologyNodeImpl nd1 =
-        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1/d1");
-    HashMap<String, EnumMap<StorageType, Integer>> nd1info =
-        nd1.getChildrenStorageInfo();
-    assertEquals(2, nd1info.keySet().size());
-    assertTrue(nd1info.get("r1").size() == 2 && nd1info.get("r2").size() == 2);
-    assertEquals(1, (int)nd1info.get("r1").get(StorageType.DISK));
-    assertEquals(1, (int)nd1info.get("r1").get(StorageType.ARCHIVE));
-    assertEquals(2, (int)nd1info.get("r2").get(StorageType.DISK));
-    assertEquals(1, (int)nd1info.get("r2").get(StorageType.ARCHIVE));
-
-    // /l1/d3 should've been out, and /l1/d1 should've been resumed
-    DFSTopologyNodeImpl nl1 =
-        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1");
-    HashMap<String, EnumMap<StorageType, Integer>> nl1info =
-        nl1.getChildrenStorageInfo();
-    assertEquals(2, nl1info.keySet().size());
-    assertTrue(l1info.get("d1").size() == 2
-        && l1info.get("d2").size() == 3);
-    assertEquals(2, (int)nl1info.get("d1").get(StorageType.ARCHIVE));
-    assertEquals(3, (int)nl1info.get("d1").get(StorageType.DISK));
-    assertEquals(1, (int)l1info.get("d2").get(StorageType.DISK));
-    assertEquals(1, (int)l1info.get("d2").get(StorageType.RAM_DISK));
-    assertEquals(1, (int)l1info.get("d2").get(StorageType.SSD));
-
-    assertNull(CLUSTER.getNode("/l1/d3"));
-  }
-}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/9832ae0e/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/net/TestDFSNetworkTopology.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/net/TestDFSNetworkTopology.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/net/TestDFSNetworkTopology.java
new file mode 100644
index 0000000..32ecf886
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/net/TestDFSNetworkTopology.java
@@ -0,0 +1,449 @@
+/**
+ * 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.hdfs.net;
+
+import com.google.common.collect.Sets;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdfs.DFSTestUtil;
+import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor;
+import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeStorageInfo;
+import org.apache.hadoop.net.Node;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.Timeout;
+
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * This class tests the correctness of storage type info stored in
+ * DFSNetworkTopology.
+ */
+public class TestDFSNetworkTopology {
+  private static final Log LOG =
+      LogFactory.getLog(TestDFSNetworkTopology.class);
+  private final static DFSNetworkTopology CLUSTER =
+      DFSNetworkTopology.getInstance(new Configuration());
+  private DatanodeDescriptor[] dataNodes;
+
+  @Rule
+  public Timeout testTimeout = new Timeout(30000);
+
+  @Before
+  public void setupDatanodes() {
+    final String[] racks = {
+        "/l1/d1/r1", "/l1/d1/r1", "/l1/d1/r2", "/l1/d1/r2", "/l1/d1/r2",
+
+        "/l1/d2/r3", "/l1/d2/r3", "/l1/d2/r3",
+
+        "/l2/d3/r1", "/l2/d3/r2", "/l2/d3/r3", "/l2/d3/r4", "/l2/d3/r5",
+
+        "/l2/d4/r1", "/l2/d4/r1", "/l2/d4/r1", "/l2/d4/r1", "/l2/d4/r1",
+        "/l2/d4/r1", "/l2/d4/r1"};
+    final String[] hosts = {
+        "host1", "host2", "host3", "host4", "host5",
+        "host6", "host7", "host8",
+        "host9", "host10", "host11", "host12", "host13",
+        "host14", "host15", "host16", "host17", "host18", "host19", "host20"};
+    final StorageType[] types = {
+        StorageType.ARCHIVE, StorageType.DISK, StorageType.ARCHIVE,
+        StorageType.DISK, StorageType.DISK,
+
+        StorageType.DISK, StorageType.RAM_DISK, StorageType.SSD,
+
+        StorageType.DISK, StorageType.RAM_DISK, StorageType.DISK,
+        StorageType.ARCHIVE, StorageType.ARCHIVE,
+
+        StorageType.DISK, StorageType.DISK, StorageType.RAM_DISK,
+        StorageType.RAM_DISK, StorageType.ARCHIVE, StorageType.ARCHIVE,
+        StorageType.SSD};
+    final DatanodeStorageInfo[] storages =
+        DFSTestUtil.createDatanodeStorageInfos(20, racks, hosts, types);
+    dataNodes = DFSTestUtil.toDatanodeDescriptor(storages);
+    for (int i = 0; i < dataNodes.length; i++) {
+      CLUSTER.add(dataNodes[i]);
+    }
+    dataNodes[9].setDecommissioned();
+    dataNodes[10].setDecommissioned();
+  }
+
+  /**
+   * Test getting the storage type info of subtree.
+   * @throws Exception
+   */
+  @Test
+  public void testGetStorageTypeInfo() throws Exception {
+    // checking level = 2 nodes
+    DFSTopologyNodeImpl d1 =
+        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1/d1");
+    HashMap<String, EnumMap<StorageType, Integer>> d1info =
+        d1.getChildrenStorageInfo();
+    assertEquals(2, d1info.keySet().size());
+    assertTrue(d1info.get("r1").size() == 2 && d1info.get("r2").size() == 2);
+    assertEquals(1, (int)d1info.get("r1").get(StorageType.DISK));
+    assertEquals(1, (int)d1info.get("r1").get(StorageType.ARCHIVE));
+    assertEquals(2, (int)d1info.get("r2").get(StorageType.DISK));
+    assertEquals(1, (int)d1info.get("r2").get(StorageType.ARCHIVE));
+
+    DFSTopologyNodeImpl d2 =
+        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1/d2");
+    HashMap<String, EnumMap<StorageType, Integer>> d2info =
+        d2.getChildrenStorageInfo();
+    assertEquals(1, d2info.keySet().size());
+    assertTrue(d2info.get("r3").size() == 3);
+    assertEquals(1, (int)d2info.get("r3").get(StorageType.DISK));
+    assertEquals(1, (int)d2info.get("r3").get(StorageType.RAM_DISK));
+    assertEquals(1, (int)d2info.get("r3").get(StorageType.SSD));
+
+    DFSTopologyNodeImpl d3 =
+        (DFSTopologyNodeImpl) CLUSTER.getNode("/l2/d3");
+    HashMap<String, EnumMap<StorageType, Integer>> d3info =
+        d3.getChildrenStorageInfo();
+    assertEquals(5, d3info.keySet().size());
+    assertEquals(1, (int)d3info.get("r1").get(StorageType.DISK));
+    assertEquals(1, (int)d3info.get("r2").get(StorageType.RAM_DISK));
+    assertEquals(1, (int)d3info.get("r3").get(StorageType.DISK));
+    assertEquals(1, (int)d3info.get("r4").get(StorageType.ARCHIVE));
+    assertEquals(1, (int)d3info.get("r5").get(StorageType.ARCHIVE));
+
+    DFSTopologyNodeImpl d4 =
+        (DFSTopologyNodeImpl) CLUSTER.getNode("/l2/d4");
+    HashMap<String, EnumMap<StorageType, Integer>> d4info =
+        d4.getChildrenStorageInfo();
+    assertEquals(1, d4info.keySet().size());
+    assertEquals(2, (int)d4info.get("r1").get(StorageType.DISK));
+    assertEquals(2, (int)d4info.get("r1").get(StorageType.RAM_DISK));
+    assertEquals(2, (int)d4info.get("r1").get(StorageType.ARCHIVE));
+    assertEquals(1, (int)d4info.get("r1").get(StorageType.SSD));
+
+    DFSTopologyNodeImpl l1 =
+        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1");
+    HashMap<String, EnumMap<StorageType, Integer>> l1info =
+        l1.getChildrenStorageInfo();
+    assertEquals(2, l1info.keySet().size());
+    assertTrue(l1info.get("d1").size() == 2
+        && l1info.get("d2").size() == 3);
+    assertEquals(2, (int)l1info.get("d1").get(StorageType.ARCHIVE));
+    assertEquals(3, (int)l1info.get("d1").get(StorageType.DISK));
+    assertEquals(1, (int)l1info.get("d2").get(StorageType.DISK));
+    assertEquals(1, (int)l1info.get("d2").get(StorageType.RAM_DISK));
+    assertEquals(1, (int)l1info.get("d2").get(StorageType.SSD));
+
+    // checking level = 1 nodes
+    DFSTopologyNodeImpl l2 =
+        (DFSTopologyNodeImpl) CLUSTER.getNode("/l2");
+    HashMap<String, EnumMap<StorageType, Integer>> l2info =
+        l2.getChildrenStorageInfo();
+    assertTrue(l2info.get("d3").size() == 3
+        && l2info.get("d4").size() == 4);
+    assertEquals(2, l2info.keySet().size());
+    assertEquals(2, (int)l2info.get("d3").get(StorageType.DISK));
+    assertEquals(2, (int)l2info.get("d3").get(StorageType.ARCHIVE));
+    assertEquals(1, (int)l2info.get("d3").get(StorageType.RAM_DISK));
+    assertEquals(2, (int)l2info.get("d4").get(StorageType.DISK));
+    assertEquals(2, (int)l2info.get("d4").get(StorageType.ARCHIVE));
+    assertEquals(2, (int)l2info.get("d4").get(StorageType.RAM_DISK));
+    assertEquals(1, (int)l2info.get("d4").get(StorageType.SSD));
+  }
+
+  /**
+   * Test the correctness of storage type info when nodes are added and removed.
+   * @throws Exception
+   */
+  @Test
+  public void testAddAndRemoveTopology() throws Exception {
+    String[] newRack = {"/l1/d1/r1", "/l1/d1/r3", "/l1/d3/r3", "/l1/d3/r3"};
+    String[] newHost = {"nhost1", "nhost2", "nhost3", "nhost4"};
+    String[] newips = {"30.30.30.30", "31.31.31.31", "32.32.32.32",
+        "33.33.33.33"};
+    StorageType[] newTypes = {StorageType.DISK, StorageType.SSD,
+        StorageType.SSD, StorageType.SSD};
+    DatanodeDescriptor[] newDD = new DatanodeDescriptor[4];
+
+    for (int i = 0; i<4; i++) {
+      DatanodeStorageInfo dsi = DFSTestUtil.createDatanodeStorageInfo(
+          "s" + newHost[i], newips[i], newRack[i], newHost[i],
+          newTypes[i], null);
+      newDD[i] = dsi.getDatanodeDescriptor();
+      CLUSTER.add(newDD[i]);
+    }
+
+    DFSTopologyNodeImpl d1 =
+        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1/d1");
+    HashMap<String, EnumMap<StorageType, Integer>> d1info =
+        d1.getChildrenStorageInfo();
+    assertEquals(3, d1info.keySet().size());
+    assertTrue(d1info.get("r1").size() == 2 && d1info.get("r2").size() == 2
+      && d1info.get("r3").size() == 1);
+    assertEquals(2, (int)d1info.get("r1").get(StorageType.DISK));
+    assertEquals(1, (int)d1info.get("r1").get(StorageType.ARCHIVE));
+    assertEquals(2, (int)d1info.get("r2").get(StorageType.DISK));
+    assertEquals(1, (int)d1info.get("r2").get(StorageType.ARCHIVE));
+    assertEquals(1, (int)d1info.get("r3").get(StorageType.SSD));
+
+    DFSTopologyNodeImpl d3 =
+        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1/d3");
+    HashMap<String, EnumMap<StorageType, Integer>> d3info =
+        d3.getChildrenStorageInfo();
+    assertEquals(1, d3info.keySet().size());
+    assertTrue(d3info.get("r3").size() == 1);
+    assertEquals(2, (int)d3info.get("r3").get(StorageType.SSD));
+
+    DFSTopologyNodeImpl l1 =
+        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1");
+    HashMap<String, EnumMap<StorageType, Integer>> l1info =
+        l1.getChildrenStorageInfo();
+    assertEquals(3, l1info.keySet().size());
+    assertTrue(l1info.get("d1").size() == 3 &&
+        l1info.get("d2").size() == 3 && l1info.get("d3").size() == 1);
+    assertEquals(4, (int)l1info.get("d1").get(StorageType.DISK));
+    assertEquals(2, (int)l1info.get("d1").get(StorageType.ARCHIVE));
+    assertEquals(1, (int)l1info.get("d1").get(StorageType.SSD));
+    assertEquals(1, (int)l1info.get("d2").get(StorageType.SSD));
+    assertEquals(1, (int)l1info.get("d2").get(StorageType.RAM_DISK));
+    assertEquals(1, (int)l1info.get("d2").get(StorageType.DISK));
+    assertEquals(2, (int)l1info.get("d3").get(StorageType.SSD));
+
+
+    for (int i = 0; i<4; i++) {
+      CLUSTER.remove(newDD[i]);
+    }
+
+    // /d1/r3 should've been out, /d1/r1 should've been resumed
+    DFSTopologyNodeImpl nd1 =
+        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1/d1");
+    HashMap<String, EnumMap<StorageType, Integer>> nd1info =
+        nd1.getChildrenStorageInfo();
+    assertEquals(2, nd1info.keySet().size());
+    assertTrue(nd1info.get("r1").size() == 2 && nd1info.get("r2").size() == 2);
+    assertEquals(1, (int)nd1info.get("r1").get(StorageType.DISK));
+    assertEquals(1, (int)nd1info.get("r1").get(StorageType.ARCHIVE));
+    assertEquals(2, (int)nd1info.get("r2").get(StorageType.DISK));
+    assertEquals(1, (int)nd1info.get("r2").get(StorageType.ARCHIVE));
+
+    // /l1/d3 should've been out, and /l1/d1 should've been resumed
+    DFSTopologyNodeImpl nl1 =
+        (DFSTopologyNodeImpl) CLUSTER.getNode("/l1");
+    HashMap<String, EnumMap<StorageType, Integer>> nl1info =
+        nl1.getChildrenStorageInfo();
+    assertEquals(2, nl1info.keySet().size());
+    assertTrue(l1info.get("d1").size() == 2
+        && l1info.get("d2").size() == 3);
+    assertEquals(2, (int)nl1info.get("d1").get(StorageType.ARCHIVE));
+    assertEquals(3, (int)nl1info.get("d1").get(StorageType.DISK));
+    assertEquals(1, (int)l1info.get("d2").get(StorageType.DISK));
+    assertEquals(1, (int)l1info.get("d2").get(StorageType.RAM_DISK));
+    assertEquals(1, (int)l1info.get("d2").get(StorageType.SSD));
+
+    assertNull(CLUSTER.getNode("/l1/d3"));
+  }
+
+  @Test
+  public void testChooseRandomWithStorageType() throws Exception {
+    Node n;
+    DatanodeDescriptor dd;
+    // test the choose random can return desired storage type nodes without
+    // exclude
+    Set<String> diskUnderL1 =
+        Sets.newHashSet("host2", "host4", "host5", "host6");
+    Set<String> archiveUnderL1 = Sets.newHashSet("host1", "host3");
+    Set<String> ramdiskUnderL1 = Sets.newHashSet("host7");
+    Set<String> ssdUnderL1 = Sets.newHashSet("host8");
+    for (int i = 0; i < 10; i++) {
+      n = CLUSTER.chooseRandomWithStorageType("/l1", null, null,
+          StorageType.DISK);
+      assertTrue(n instanceof DatanodeDescriptor);
+      dd = (DatanodeDescriptor) n;
+      assertTrue(diskUnderL1.contains(dd.getHostName()));
+
+      n = CLUSTER.chooseRandomWithStorageType("/l1", null, null,
+          StorageType.RAM_DISK);
+      assertTrue(n instanceof DatanodeDescriptor);
+      dd = (DatanodeDescriptor) n;
+      assertTrue(ramdiskUnderL1.contains(dd.getHostName()));
+
+      n = CLUSTER.chooseRandomWithStorageType("/l1", null, null,
+          StorageType.ARCHIVE);
+      assertTrue(n instanceof DatanodeDescriptor);
+      dd = (DatanodeDescriptor) n;
+      assertTrue(archiveUnderL1.contains(dd.getHostName()));
+
+      n = CLUSTER.chooseRandomWithStorageType("/l1", null, null,
+          StorageType.SSD);
+      assertTrue(n instanceof DatanodeDescriptor);
+      dd = (DatanodeDescriptor) n;
+      assertTrue(ssdUnderL1.contains(dd.getHostName()));
+    }
+  }
+
+  @Test
+  public void testChooseRandomWithStorageTypeWithExcluded() throws Exception {
+    Node n;
+    DatanodeDescriptor dd;
+    // below test choose random with exclude, for /l2/d3, every rack has exactly
+    // one host
+    // /l2/d3 has five racks r[1~5] but only r4 and r5 have ARCHIVE
+    // host12 is the one under "/l2/d3/r4", host13 is the one under "/l2/d3/r5"
+    n = CLUSTER.chooseRandomWithStorageType("/l2/d3/r4", null, null,
+        StorageType.ARCHIVE);
+    HashSet<Node> excluded = new HashSet<>();
+    // exclude the host on r4 (since there is only one host, no randomness here)
+    excluded.add(n);
+
+    for (int i = 0; i<10; i++) {
+      n = CLUSTER.chooseRandomWithStorageType("/l2/d3", null, null,
+          StorageType.ARCHIVE);
+      assertTrue(n instanceof DatanodeDescriptor);
+      dd = (DatanodeDescriptor) n;
+      assertTrue(dd.getHostName().equals("host12") ||
+          dd.getHostName().equals("host13"));
+    }
+
+    // test exclude nodes
+    for (int i = 0; i<10; i++) {
+      n = CLUSTER.chooseRandomWithStorageType("/l2/d3", null, excluded,
+          StorageType.ARCHIVE);
+      assertTrue(n instanceof DatanodeDescriptor);
+      dd = (DatanodeDescriptor) n;
+      assertTrue(dd.getHostName().equals("host13"));
+    }
+
+    // test exclude scope
+    for (int i = 0; i<10; i++) {
+      n = CLUSTER.chooseRandomWithStorageType("/l2/d3", "/l2/d3/r4", null,
+          StorageType.ARCHIVE);
+      assertTrue(n instanceof DatanodeDescriptor);
+      dd = (DatanodeDescriptor) n;
+      assertTrue(dd.getHostName().equals("host13"));
+    }
+
+    // test exclude scope + excluded node with expected null return node
+    for (int i = 0; i<10; i++) {
+      n = CLUSTER.chooseRandomWithStorageType("/l2/d3", "/l2/d3/r5", excluded,
+          StorageType.ARCHIVE);
+      assertNull(n);
+    }
+
+    // test exclude scope + excluded node with expected non-null return node
+    n = CLUSTER.chooseRandomWithStorageType("/l1/d2", null, null,
+        StorageType.DISK);
+    dd = (DatanodeDescriptor)n;
+    assertEquals("host6", dd.getHostName());
+    // exclude the host on r4 (since there is only one host, no randomness here)
+    excluded.add(n);
+    Set<String> expectedSet = Sets.newHashSet("host4", "host5");
+    for (int i = 0; i<10; i++) {
+      // under l1, there are four hosts with DISK:
+      // /l1/d1/r1/host2, /l1/d1/r2/host4, /l1/d1/r2/host5 and /l1/d2/r3/host6
+      // host6 is excludedNode, host2 is under excluded range scope /l1/d1/r1
+      // so should always return r4 or r5
+      n = CLUSTER.chooseRandomWithStorageType(
+          "/l1", "/l1/d1/r1", excluded, StorageType.DISK);
+      dd = (DatanodeDescriptor) n;
+      assertTrue(expectedSet.contains(dd.getHostName()));
+    }
+  }
+
+
+  /**
+   * This test tests the wrapper method. The wrapper method only takes one scope
+   * where if it starts with a ~, it is an excluded scope, and searching always
+   * from root. Otherwise it is a scope.
+   * @throws Exception throws exception.
+   */
+  @Test
+  public void testChooseRandomWithStorageTypeWrapper() throws Exception {
+    Node n;
+    DatanodeDescriptor dd;
+    n = CLUSTER.chooseRandomWithStorageType("/l2/d3/r4", null, null,
+        StorageType.ARCHIVE);
+    HashSet<Node> excluded = new HashSet<>();
+    // exclude the host on r4 (since there is only one host, no randomness here)
+    excluded.add(n);
+
+    // search with given scope being desired scope
+    for (int i = 0; i<10; i++) {
+      n = CLUSTER.chooseRandomWithStorageType(
+          "/l2/d3", null, StorageType.ARCHIVE);
+      assertTrue(n instanceof DatanodeDescriptor);
+      dd = (DatanodeDescriptor) n;
+      assertTrue(dd.getHostName().equals("host12") ||
+          dd.getHostName().equals("host13"));
+    }
+
+    for (int i = 0; i<10; i++) {
+      n = CLUSTER.chooseRandomWithStorageType(
+          "/l2/d3", excluded, StorageType.ARCHIVE);
+      assertTrue(n instanceof DatanodeDescriptor);
+      dd = (DatanodeDescriptor) n;
+      assertTrue(dd.getHostName().equals("host13"));
+    }
+
+    // search with given scope being exclude scope
+
+    // a total of 4 ramdisk nodes:
+    // /l1/d2/r3/host7, /l2/d3/r2/host10, /l2/d4/r1/host7 and /l2/d4/r1/host10
+    // so if we exclude /l2/d4/r1, if should be always either host7 or host10
+    for (int i = 0; i<10; i++) {
+      n = CLUSTER.chooseRandomWithStorageType(
+          "~/l2/d4", null, StorageType.RAM_DISK);
+      assertTrue(n instanceof DatanodeDescriptor);
+      dd = (DatanodeDescriptor) n;
+      assertTrue(dd.getHostName().equals("host7") ||
+          dd.getHostName().equals("host10"));
+    }
+
+    // similar to above, except that we also exclude host10 here. so it should
+    // always be host7
+    n = CLUSTER.chooseRandomWithStorageType("/l2/d3/r2", null, null,
+        StorageType.RAM_DISK);
+    // add host10 to exclude
+    excluded.add(n);
+    for (int i = 0; i<10; i++) {
+      n = CLUSTER.chooseRandomWithStorageType(
+          "~/l2/d4", excluded, StorageType.RAM_DISK);
+      assertTrue(n instanceof DatanodeDescriptor);
+      dd = (DatanodeDescriptor) n;
+      assertTrue(dd.getHostName().equals("host7"));
+    }
+  }
+
+  @Test
+  public void testNonExistingNode() throws Exception {
+    Node n;
+    n = CLUSTER.chooseRandomWithStorageType(
+        "/l100", null, null, StorageType.DISK);
+    assertNull(n);
+    n = CLUSTER.chooseRandomWithStorageType(
+        "/l100/d100", null, null, StorageType.DISK);
+    assertNull(n);
+    n = CLUSTER.chooseRandomWithStorageType(
+        "/l100/d100/r100", null, null, StorageType.DISK);
+    assertNull(n);
+  }
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org


Mime
View raw message