@Override
  protected DatanodeDescriptor chooseDataNode(final String scope,
      final Collection<Node> excludedNode, StorageType type) {
    // only the code that uses DFSNetworkTopology should trigger this code path.
    Preconditions.checkArgument(clusterMap instanceof DFSNetworkTopology);
    DFSNetworkTopology dfsClusterMap = (DFSNetworkTopology)clusterMap;
    DatanodeDescriptor a = (DatanodeDescriptor) dfsClusterMap
        .chooseRandomWithStorageTypeTwoTrial(scope, excludedNode, type);
    DatanodeDescriptor b = (DatanodeDescriptor) dfsClusterMap
        .chooseRandomWithStorageTypeTwoTrial(scope, excludedNode, type);
    return select(a, b, false);
  }
@Test
  public void testChooseDataNode() {
    Collection<Node> allNodes = new ArrayList<>(dataNodes.length);
    Collections.addAll(allNodes, dataNodes);
    if (placementPolicy instanceof AvailableSpaceBlockPlacementPolicy) {
      // exclude all datanodes when chooseDataNode, no NPE should be thrown
      ((AvailableSpaceBlockPlacementPolicy) placementPolicy)
          .chooseDataNode("~", allNodes);
    }
  }