@VisibleForTesting
  DatanodeDescriptor[] chooseSourceDatanodes(BlockInfo block,
      List<DatanodeDescriptor> containingNodes,
      List<DatanodeStorageInfo> nodesContainingLiveReplicas,
      NumberReplicas numReplicas, List<Byte> liveBlockIndices,
      List<Byte> liveBusyBlockIndices, List<Byte> excludeReconstructed, int priority) {
    containingNodes.clear();
    nodesContainingLiveReplicas.clear();
    List<DatanodeDescriptor> srcNodes = new ArrayList<>();
    liveBlockIndices.clear();
    final boolean isStriped = block.isStriped();
    DatanodeDescriptor decommissionedSrc = null;

    BitSet liveBitSet = null;
    BitSet decommissioningBitSet = null;
    if (isStriped) {
      int blockNum = ((BlockInfoStriped) block).getTotalBlockNum();
      liveBitSet = new BitSet(blockNum);
      decommissioningBitSet = new BitSet(blockNum);
    }

    for (DatanodeStorageInfo storage : blocksMap.getStorages(block)) {
      final DatanodeDescriptor node = getDatanodeDescriptorFromStorage(storage);
      final StoredReplicaState state = checkReplicaOnStorage(numReplicas, block,
          storage, corruptReplicas.getNodes(block), false);
      if (state == StoredReplicaState.LIVE) {
        if (storage.getStorageType() == StorageType.PROVIDED) {
          storage = new DatanodeStorageInfo(node, storage.getStorageID(),
              storage.getStorageType(), storage.getState());
        }
        nodesContainingLiveReplicas.add(storage);
      }
      containingNodes.add(node);

      // do not select the replica if it is corrupt or excess
      if (state == StoredReplicaState.CORRUPT ||
          state == StoredReplicaState.EXCESS) {
        continue;
      }

      // Never use maintenance node not suitable for read
      // or unknown state replicas.
      if (state == null
          || state == StoredReplicaState.MAINTENANCE_NOT_FOR_READ) {
        continue;
      }

      // Save the live decommissioned replica in case we need it. Such replicas
      // are normally not used for replication, but if nothing else is
      // available, one can be selected as a source.
      if (state == StoredReplicaState.DECOMMISSIONED) {
        if (decommissionedSrc == null ||
            ThreadLocalRandom.current().nextBoolean()) {
          decommissionedSrc = node;
        }
        continue;
      }

      // for EC here need to make sure the numReplicas replicates state correct
      // because in the scheduleReconstruction it need the numReplicas to check
      // whether need to reconstruct the ec internal block
      byte blockIndex = -1;
      if (isStriped) {
        blockIndex = ((BlockInfoStriped) block)
            .getStorageBlockIndex(storage);
        countLiveAndDecommissioningReplicas(numReplicas, state,
            liveBitSet, decommissioningBitSet, blockIndex);
      }

      if (priority != LowRedundancyBlocks.QUEUE_HIGHEST_PRIORITY
          && (!node.isDecommissionInProgress() && !node.isEnteringMaintenance())
          && node.getNumberOfBlocksToBeReplicated() +
          node.getNumberOfBlocksToBeErasureCoded() >= maxReplicationStreams) {
        if (isStriped && (state == StoredReplicaState.LIVE
            || state == StoredReplicaState.DECOMMISSIONING)) {
          liveBusyBlockIndices.add(blockIndex);
          //HDFS-16566 ExcludeReconstructed won't be reconstructed.
          excludeReconstructed.add(blockIndex);
        }
        continue; // already reached replication limit
      }

      if (node.getNumberOfBlocksToBeReplicated() +
          node.getNumberOfBlocksToBeErasureCoded() >= replicationStreamsHardLimit) {
        if (isStriped && (state == StoredReplicaState.LIVE
            || state == StoredReplicaState.DECOMMISSIONING)) {
          liveBusyBlockIndices.add(blockIndex);
          //HDFS-16566 ExcludeReconstructed won't be reconstructed.
          excludeReconstructed.add(blockIndex);
        }
        continue;
      }

      if(isStriped || srcNodes.isEmpty()) {
        srcNodes.add(node);
        if (isStriped) {
          liveBlockIndices.add(blockIndex);
        }
        continue;
      }
      // for replicated block, switch to a different node randomly
      // this to prevent from deterministically selecting the same node even
      // if the node failed to replicate the block on previous iterations
      if (ThreadLocalRandom.current().nextBoolean()) {
        srcNodes.set(0, node);
      }
    }

    // Pick a live decommissioned replica, if nothing else is available.
    if (!isStriped && nodesContainingLiveReplicas.isEmpty() &&
        srcNodes.isEmpty() && decommissionedSrc != null) {
      srcNodes.add(decommissionedSrc);
    }

    return srcNodes.toArray(new DatanodeDescriptor[srcNodes.size()]);
  }
@Test
  public void testChooseSrcDNWithDupECInDecommissioningNode() throws Exception {
    long blockId = -9223372036854775776L; // real ec block id
    Block aBlock = new Block(blockId, 0, 0);
    // RS-3-2 EC policy
    ErasureCodingPolicy ecPolicy =
        SystemErasureCodingPolicies.getPolicies().get(1);
    // striped blockInfo
    BlockInfoStriped aBlockInfoStriped = new BlockInfoStriped(aBlock, ecPolicy);
    // ec storageInfo
    DatanodeStorageInfo ds1 = DFSTestUtil.createDatanodeStorageInfo(
        "storage1", "1.1.1.1", "rack1", "host1");
    DatanodeStorageInfo ds2 = DFSTestUtil.createDatanodeStorageInfo(
        "storage2", "2.2.2.2", "rack2", "host2");
    DatanodeStorageInfo ds3 = DFSTestUtil.createDatanodeStorageInfo(
        "storage3", "3.3.3.3", "rack3", "host3");
    DatanodeStorageInfo ds4 = DFSTestUtil.createDatanodeStorageInfo(
        "storage4", "4.4.4.4", "rack4", "host4");
    DatanodeStorageInfo ds5 = DFSTestUtil.createDatanodeStorageInfo(
        "storage5", "5.5.5.5", "rack5", "host5");
    DatanodeStorageInfo ds6 = DFSTestUtil.createDatanodeStorageInfo(
        "storage6", "6.6.6.6", "rack6", "host6");

    // link block with storage
    aBlockInfoStriped.addStorage(ds1, aBlock);
    aBlockInfoStriped.addStorage(ds2, new Block(blockId + 1, 0, 0));
    aBlockInfoStriped.addStorage(ds3, new Block(blockId + 2, 0, 0));
    aBlockInfoStriped.addStorage(ds4, new Block(blockId + 3, 0, 0));
    aBlockInfoStriped.addStorage(ds5, new Block(blockId + 4, 0, 0));
    // NOTE: duplicate block 0，this DN will replace the decommission ds1 DN
    aBlockInfoStriped.addStorage(ds6, aBlock);

    addEcBlockToBM(blockId, ecPolicy);
    // decommission datanode where store block 0
    ds1.getDatanodeDescriptor().startDecommission();

    List<DatanodeDescriptor> containingNodes =
        new LinkedList<DatanodeDescriptor>();
    List<DatanodeStorageInfo> nodesContainingLiveReplicas =
        new LinkedList<DatanodeStorageInfo>();
    NumberReplicas numReplicas = new NumberReplicas();
    List<Byte> liveBlockIndices = new ArrayList<>();
    List<Byte> liveBusyBlockIndices = new ArrayList<>();
    List<Byte> excludeReconstructedIndices = new ArrayList<>();

    bm.chooseSourceDatanodes(
        aBlockInfoStriped,
        containingNodes,
        nodesContainingLiveReplicas,
        numReplicas, liveBlockIndices,
        liveBusyBlockIndices,
        excludeReconstructedIndices,
        LowRedundancyBlocks.QUEUE_HIGHEST_PRIORITY);
    assertEquals("There are 5 live replicas in " +
            "[ds2, ds3, ds4, ds5, ds6] datanodes ",
        5, numReplicas.liveReplicas());
    assertEquals("The ds1 datanode is in decommissioning, " +
            "so there is no redundant replica",
        0, numReplicas.redundantInternalBlocks());
  }