/**
 * 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 static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.BitSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileChecksum;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.MiniDFSCluster.DataNodeProperties;
import org.apache.hadoop.hdfs.client.HdfsDataInputStream;
import org.apache.hadoop.hdfs.protocol.Block;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo.AdminStates;
import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy;
import org.apache.hadoop.hdfs.protocol.HdfsConstants.DatanodeReportType;
import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.hdfs.protocol.LocatedStripedBlock;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfo;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfoStriped;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockManagerTestUtil;
import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor;
import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeStorageInfo;
import org.apache.hadoop.hdfs.server.datanode.DataNode;
import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
import org.apache.hadoop.hdfs.server.namenode.INodeFile;
import org.apache.hadoop.hdfs.server.namenode.NameNode;
import org.apache.hadoop.hdfs.server.namenode.NameNodeAdapter;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.test.GenericTestUtils;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This class tests the decommissioning of datanode with striped blocks.
 */
public class TestDecommissionWithStriped {
  private static final Logger LOG = LoggerFactory
      .getLogger(TestDecommissionWithStriped.class);

  // heartbeat interval in seconds
  private static final int HEARTBEAT_INTERVAL = 1;
  // block report in msec
  private static final int BLOCKREPORT_INTERVAL_MSEC = 1000;
  // replication interval
  private static final int NAMENODE_REPLICATION_INTERVAL = 1;

  private int replicationStreamsHardLimit =
      DFSConfigKeys.DFS_NAMENODE_REPLICATION_STREAMS_HARD_LIMIT_DEFAULT;

  private Path decommissionDir;
  private Path hostsFile;
  private Path excludeFile;
  private LocalFileSystem localFileSys;

  @Rule
  public TemporaryFolder baseDir = new TemporaryFolder();

  private Configuration conf;
  private MiniDFSCluster cluster;
  private DistributedFileSystem dfs;
  private final ErasureCodingPolicy ecPolicy =
      StripedFileTestUtil.getDefaultECPolicy();
  private int numDNs;
  private final int cellSize = ecPolicy.getCellSize();
  private final int dataBlocks = ecPolicy.getNumDataUnits();
  private final int parityBlocks = ecPolicy.getNumParityUnits();
  private final int blockSize = cellSize * 4;
  private final int blockGroupSize = blockSize * dataBlocks;
  private final Path ecDir = new Path("/" + this.getClass().getSimpleName());

  private FSNamesystem fsn;
  private BlockManager bm;
  private DFSClient client;

  protected Configuration createConfiguration() {
    return new HdfsConfiguration();
  }

  @Before
  public void setup() throws IOException {
    conf = createConfiguration();
    // Set up the hosts/exclude files.
    localFileSys = FileSystem.getLocal(conf);
    localFileSys.setWorkingDirectory(new Path(baseDir.getRoot().getPath()));
    Path workingDir = localFileSys.getWorkingDirectory();
    decommissionDir = new Path(workingDir, "work-dir/decommission");
    hostsFile = new Path(decommissionDir, "hosts");
    excludeFile = new Path(decommissionDir, "exclude");
    writeConfigFile(hostsFile, null);
    writeConfigFile(excludeFile, null);

    // Setup conf
    conf.set(DFSConfigKeys.DFS_HOSTS, hostsFile.toUri().getPath());
    conf.set(DFSConfigKeys.DFS_HOSTS_EXCLUDE, excludeFile.toUri().getPath());
    conf.setInt(DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY,
        2000);
    conf.setInt(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, HEARTBEAT_INTERVAL);
    conf.setInt(DFSConfigKeys.DFS_NAMENODE_REDUNDANCY_INTERVAL_SECONDS_KEY, 1);
    conf.setInt(DFSConfigKeys.DFS_BLOCKREPORT_INTERVAL_MSEC_KEY,
        BLOCKREPORT_INTERVAL_MSEC);
    conf.setInt(DFSConfigKeys.DFS_NAMENODE_RECONSTRUCTION_PENDING_TIMEOUT_SEC_KEY,
        4);
    conf.setInt(DFSConfigKeys.DFS_NAMENODE_REDUNDANCY_INTERVAL_SECONDS_KEY,
        NAMENODE_REPLICATION_INTERVAL);
    conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, blockSize);
    conf.setInt(
        DFSConfigKeys.DFS_DN_EC_RECONSTRUCTION_STRIPED_READ_BUFFER_SIZE_KEY,
        cellSize - 1);
    conf.setInt(DFSConfigKeys.DFS_NAMENODE_REDUNDANCY_INTERVAL_SECONDS_KEY, 1);
    conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_REDUNDANCY_CONSIDERLOAD_KEY,
        false);

    numDNs = dataBlocks + parityBlocks + 5;
    cluster = new MiniDFSCluster.Builder(conf).numDataNodes(numDNs).build();
    cluster.waitActive();
    dfs = cluster.getFileSystem(0);
    fsn = cluster.getNamesystem();
    bm = fsn.getBlockManager();
    client = getDfsClient(cluster.getNameNode(0), conf);

    dfs.enableErasureCodingPolicy(
        StripedFileTestUtil.getDefaultECPolicy().getName());
    dfs.mkdirs(ecDir);
    dfs.setErasureCodingPolicy(ecDir,
        StripedFileTestUtil.getDefaultECPolicy().getName());
  }

  @After
  public void teardown() throws IOException {
    cleanupFile(localFileSys, decommissionDir);
    if (cluster != null) {
      cluster.shutdown();
      cluster = null;
    }
  }

  @Test(timeout = 120000)
  public void testFileFullBlockGroup() throws Exception {
    LOG.info("Starting test testFileFullBlockGroup");
    testDecommission(blockSize * dataBlocks, 9, 1, "testFileFullBlockGroup");
  }

  @Test(timeout = 120000)
  public void testFileMultipleBlockGroups() throws Exception {
    LOG.info("Starting test testFileMultipleBlockGroups");
    int writeBytes = 2 * blockSize * dataBlocks;
    testDecommission(writeBytes, 9, 1, "testFileMultipleBlockGroups");
  }

  @Test(timeout = 120000)
  public void testFileSmallerThanOneCell() throws Exception {
    LOG.info("Starting test testFileSmallerThanOneCell");
    testDecommission(cellSize - 1, 4, 1, "testFileSmallerThanOneCell");
  }

  @Test(timeout = 120000)
  public void testFileSmallerThanOneStripe() throws Exception {
    LOG.info("Starting test testFileSmallerThanOneStripe");
    testDecommission(cellSize * 2, 5, 1, "testFileSmallerThanOneStripe");
  }

  @Test(timeout = 120000)
  public void testDecommissionTwoNodes() throws Exception {
    LOG.info("Starting test testDecommissionTwoNodes");
    testDecommission(blockSize * dataBlocks, 9, 2, "testDecommissionTwoNodes");
  }

  @Test(timeout = 120000)
  public void testDecommissionWithURBlockForSameBlockGroup() throws Exception {
    LOG.info("Starting test testDecommissionWithURBlocksForSameBlockGroup");

    final Path ecFile = new Path(ecDir, "testDecommissionWithCorruptBlocks");
    int writeBytes = cellSize * dataBlocks * 2;
    writeStripedFile(dfs, ecFile, writeBytes);
    Assert.assertEquals(0, bm.numOfUnderReplicatedBlocks());

    final List<DatanodeInfo> decommisionNodes = new ArrayList<DatanodeInfo>();
    LocatedBlock lb = dfs.getClient().getLocatedBlocks(ecFile.toString(), 0)
        .get(0);
    DatanodeInfo[] dnLocs = lb.getLocations();
    assertEquals(dataBlocks + parityBlocks, dnLocs.length);
    int decommNodeIndex = dataBlocks - 1;
    int stopNodeIndex = 1;

    // add the nodes which will be decommissioning
    decommisionNodes.add(dnLocs[decommNodeIndex]);

    // stop excess dns to avoid immediate reconstruction.
    DatanodeInfo[] info = client.datanodeReport(DatanodeReportType.LIVE);
    List<DataNodeProperties> stoppedDns = new ArrayList<>();
    for (DatanodeInfo liveDn : info) {
      boolean usedNode = false;
      for (DatanodeInfo datanodeInfo : dnLocs) {
        if (liveDn.getXferAddr().equals(datanodeInfo.getXferAddr())) {
          usedNode = true;
          break;
        }
      }
      if (!usedNode) {
        DataNode dn = cluster.getDataNode(liveDn.getIpcPort());
        stoppedDns.add(cluster.stopDataNode(liveDn.getXferAddr()));
        cluster.setDataNodeDead(dn.getDatanodeId());
        LOG.info("stop datanode " + dn.getDatanodeId().getHostName());
      }
    }
    DataNode dn = cluster.getDataNode(dnLocs[stopNodeIndex].getIpcPort());
    cluster.stopDataNode(dnLocs[stopNodeIndex].getXferAddr());
    cluster.setDataNodeDead(dn.getDatanodeId());
    numDNs = numDNs - 1;

    // Decommission node in a new thread. Verify that node is decommissioned.
    final CountDownLatch decomStarted = new CountDownLatch(0);
    Thread decomTh = new Thread() {
      public void run() {
        try {
          decomStarted.countDown();
          decommissionNode(0, decommisionNodes, AdminStates.DECOMMISSIONED);
        } catch (Exception e) {
          LOG.error("Exception while decommissioning", e);
          Assert.fail("Shouldn't throw exception!");
        }
      };
    };
    int deadDecommissioned = fsn.getNumDecomDeadDataNodes();
    int liveDecommissioned = fsn.getNumDecomLiveDataNodes();
    decomTh.start();
    decomStarted.await(5, TimeUnit.SECONDS);
    Thread.sleep(3000); // grace period to trigger decommissioning call
    // start datanode so that decommissioning live node will be finished
    for (DataNodeProperties dnp : stoppedDns) {
      cluster.restartDataNode(dnp);
      LOG.info("Restarts stopped datanode:{} to trigger block reconstruction",
          dnp.datanode);
    }
    cluster.waitActive();

    LOG.info("Waiting to finish decommissioning node:{}", decommisionNodes);
    decomTh.join(20000); // waiting 20secs to finish decommission
    LOG.info("Finished decommissioning node:{}", decommisionNodes);

    assertEquals(deadDecommissioned, fsn.getNumDecomDeadDataNodes());
    assertEquals(liveDecommissioned + decommisionNodes.size(),
        fsn.getNumDecomLiveDataNodes());

    // Ensure decommissioned datanode is not automatically shutdown
    assertEquals("All datanodes must be alive", numDNs,
        client.datanodeReport(DatanodeReportType.LIVE).length);

    assertNull(checkFile(dfs, ecFile, 9, decommisionNodes, numDNs));
    StripedFileTestUtil.checkData(dfs, ecFile, writeBytes, decommisionNodes,
        null, blockGroupSize);
    cleanupFile(dfs, ecFile);
  }

  /**
   * DN decommission shouldn't reconstruction busy DN block.
   * @throws Exception
   */
  @Test(timeout = 120000)
  public void testDecommissionWithBusyNode() throws Exception {
    byte busyDNIndex = 1;
    byte decommisionDNIndex = 0;
    //1. create EC file
    final Path ecFile = new Path(ecDir, "testDecommissionWithBusyNode");
    int writeBytes = cellSize * dataBlocks;
    writeStripedFile(dfs, ecFile, writeBytes);
    Assert.assertEquals(0, bm.numOfUnderReplicatedBlocks());
    FileChecksum fileChecksum1 = dfs.getFileChecksum(ecFile, writeBytes);

    //2. make once DN busy
    final INodeFile fileNode = cluster.getNamesystem().getFSDirectory()
        .getINode4Write(ecFile.toString()).asFile();
    BlockInfo firstBlock = fileNode.getBlocks()[0];
    DatanodeStorageInfo[] dnStorageInfos = bm.getStorages(firstBlock);
    DatanodeDescriptor busyNode =
        dnStorageInfos[busyDNIndex].getDatanodeDescriptor();
    for (int j = 0; j < replicationStreamsHardLimit; j++) {
      busyNode.incrementPendingReplicationWithoutTargets();
    }

    //3. decomission one node
    List<DatanodeInfo> decommisionNodes = new ArrayList<>();
    decommisionNodes.add(
        dnStorageInfos[decommisionDNIndex].getDatanodeDescriptor());
    decommissionNode(0, decommisionNodes, AdminStates.DECOMMISSIONED);
    assertEquals(decommisionNodes.size(), fsn.getNumDecomLiveDataNodes());

    //4. wait for decommission block to replicate
    Thread.sleep(3000);
    DatanodeStorageInfo[] newDnStorageInfos = bm.getStorages(firstBlock);
    Assert.assertEquals("Busy DN shouldn't be reconstructed",
        dnStorageInfos[busyDNIndex].getStorageID(),
        newDnStorageInfos[busyDNIndex].getStorageID());

    //5. check decommission DN block index, it should be reconstructed again
    LocatedBlocks lbs = cluster.getNameNodeRpc().getBlockLocations(
        ecFile.toString(), 0, writeBytes);
    LocatedStripedBlock bg = (LocatedStripedBlock) (lbs.get(0));
    int decommissionBlockIndexCount = 0;
    for (byte index : bg.getBlockIndices()) {
      if (index == decommisionDNIndex) {
        decommissionBlockIndexCount++;
      }
    }

    Assert.assertEquals("Decommission DN block should be reconstructed", 2,
        decommissionBlockIndexCount);

    FileChecksum fileChecksum2 = dfs.getFileChecksum(ecFile, writeBytes);
    Assert.assertTrue("Checksum mismatches!",
        fileChecksum1.equals(fileChecksum2));
  }

  /**
   * Decommission may generate the parity block's content with all 0
   * in some case.
   * @throws Exception
   */
  @Test(timeout = 120000)
  public void testDecommission2NodeWithBusyNode() throws Exception {
    byte busyDNIndex = 6;
    byte decommissionDNIndex = 6;
    byte decommissionDNIndex2 = 8;
    //1. create EC file
    final Path ecFile = new Path(ecDir, "testDecommission2NodeWithBusyNode");
    int writeBytes = cellSize * dataBlocks;
    writeStripedFile(dfs, ecFile, writeBytes);

    Assert.assertEquals(0, bm.numOfUnderReplicatedBlocks());
    FileChecksum fileChecksum1 = dfs.getFileChecksum(ecFile, writeBytes);

    //2. make once DN busy
    final INodeFile fileNode = cluster.getNamesystem().getFSDirectory()
        .getINode4Write(ecFile.toString()).asFile();
    BlockInfo firstBlock = fileNode.getBlocks()[0];
    DatanodeStorageInfo[] dnStorageInfos = bm.getStorages(firstBlock);
    DatanodeDescriptor busyNode = dnStorageInfos[busyDNIndex]
        .getDatanodeDescriptor();
    for (int j = 0; j < replicationStreamsHardLimit; j++) {
      busyNode.incrementPendingReplicationWithoutTargets();
    }

    //3. decommissioning one node
    List<DatanodeInfo> decommissionNodes = new ArrayList<>();
    decommissionNodes.add(dnStorageInfos[decommissionDNIndex]
        .getDatanodeDescriptor());
    decommissionNodes.add(dnStorageInfos[decommissionDNIndex2]
        .getDatanodeDescriptor());
    decommissionNode(0, decommissionNodes, AdminStates.DECOMMISSION_INPROGRESS);

    //4. wait for decommissioning and not busy block to replicate(9-2+1=8)
    GenericTestUtils.waitFor(
        () -> bm.countNodes(firstBlock).liveReplicas() >= 8,
        100, 60000);

    //5. release busy DN, make the decommissioning and busy block can replicate
    busyNode.decrementPendingReplicationWithoutTargets();

    //6. decommissioned one node,make the decommission finished
    decommissionNode(0, decommissionNodes, AdminStates.DECOMMISSIONED);

    //7. Busy DN shouldn't be reconstructed
    DatanodeStorageInfo[] newDnStorageInfos = bm.getStorages(firstBlock);
    Assert.assertEquals("Busy DN shouldn't be reconstructed",
        dnStorageInfos[busyDNIndex].getStorageID(),
        newDnStorageInfos[busyDNIndex].getStorageID());

    //8. check the checksum of a file
    FileChecksum fileChecksum2 = dfs.getFileChecksum(ecFile, writeBytes);
    Assert.assertEquals("Checksum mismatches!", fileChecksum1, fileChecksum2);

    //9. check the data is correct
    StripedFileTestUtil.checkData(dfs, ecFile, writeBytes, decommissionNodes,
        null, blockGroupSize);
  }

  /**
   * Tests to verify that the file checksum should be able to compute after the
   * decommission operation.
   *
   * Below is the block indices list after the decommission. ' represents
   * decommissioned node index.
   *
   * 0, 2, 3, 4, 5, 6, 7, 8, 1, 1'
   *
   * Here, this list contains duplicated blocks and does not maintaining any
   * order.
   */
  @Test(timeout = 120000)
  public void testFileChecksumAfterDecommission() throws Exception {
    LOG.info("Starting test testFileChecksumAfterDecommission");

    final Path ecFile = new Path(ecDir, "testFileChecksumAfterDecommission");
    int writeBytes = cellSize * dataBlocks;
    writeStripedFile(dfs, ecFile, writeBytes);
    Assert.assertEquals(0, bm.numOfUnderReplicatedBlocks());
    FileChecksum fileChecksum1 = dfs.getFileChecksum(ecFile, writeBytes);

    final List<DatanodeInfo> decommisionNodes = new ArrayList<DatanodeInfo>();
    LocatedBlock lb = dfs.getClient().getLocatedBlocks(ecFile.toString(), 0)
        .get(0);
    DatanodeInfo[] dnLocs = lb.getLocations();
    assertEquals(dataBlocks + parityBlocks, dnLocs.length);
    int decommNodeIndex = 1;

    // add the node which will be decommissioning
    decommisionNodes.add(dnLocs[decommNodeIndex]);
    decommissionNode(0, decommisionNodes, AdminStates.DECOMMISSIONED);
    assertEquals(decommisionNodes.size(), fsn.getNumDecomLiveDataNodes());
    assertNull(checkFile(dfs, ecFile, 9, decommisionNodes, numDNs));
    StripedFileTestUtil.checkData(dfs, ecFile, writeBytes, decommisionNodes,
        null, blockGroupSize);

    // verify checksum
    FileChecksum fileChecksum2 = dfs.getFileChecksum(ecFile, writeBytes);
    LOG.info("fileChecksum1:" + fileChecksum1);
    LOG.info("fileChecksum2:" + fileChecksum2);

    Assert.assertTrue("Checksum mismatches!",
        fileChecksum1.equals(fileChecksum2));
  }

  private void testDecommission(int writeBytes, int storageCount,
      int decomNodeCount, String filename) throws IOException, Exception {
    Path ecFile = new Path(ecDir, filename);
    writeStripedFile(dfs, ecFile, writeBytes);
    List<DatanodeInfo> decommisionNodes = getDecommissionDatanode(dfs, ecFile,
        writeBytes, decomNodeCount);

    int deadDecommissioned = fsn.getNumDecomDeadDataNodes();
    int liveDecommissioned = fsn.getNumDecomLiveDataNodes();
    List<LocatedBlock> lbs = ((HdfsDataInputStream) dfs.open(ecFile))
        .getAllBlocks();

    // prepare expected block index and token list.
    List<HashMap<DatanodeInfo, Byte>> locToIndexList = new ArrayList<>();
    List<HashMap<DatanodeInfo, Token<BlockTokenIdentifier>>> locToTokenList =
        new ArrayList<>();
    prepareBlockIndexAndTokenList(lbs, locToIndexList, locToTokenList);

    // Decommission node. Verify that node is decommissioned.
    decommissionNode(0, decommisionNodes, AdminStates.DECOMMISSIONED);

    assertEquals(deadDecommissioned, fsn.getNumDecomDeadDataNodes());
    assertEquals(liveDecommissioned + decommisionNodes.size(),
        fsn.getNumDecomLiveDataNodes());

    // Ensure decommissioned datanode is not automatically shutdown
    DFSClient client = getDfsClient(cluster.getNameNode(0), conf);
    assertEquals("All datanodes must be alive", numDNs,
        client.datanodeReport(DatanodeReportType.LIVE).length);

    assertNull(checkFile(dfs, ecFile, storageCount, decommisionNodes, numDNs));
    StripedFileTestUtil.checkData(dfs, ecFile, writeBytes, decommisionNodes,
        null, blockGroupSize);

    assertBlockIndexAndTokenPosition(lbs, locToIndexList, locToTokenList);

    cleanupFile(dfs, ecFile);
  }

  private void prepareBlockIndexAndTokenList(List<LocatedBlock> lbs,
      List<HashMap<DatanodeInfo, Byte>> locToIndexList,
      List<HashMap<DatanodeInfo, Token<BlockTokenIdentifier>>> locToTokenList) {
    for (LocatedBlock lb : lbs) {
      HashMap<DatanodeInfo, Byte> locToIndex = new HashMap<DatanodeInfo, Byte>();
      locToIndexList.add(locToIndex);

      HashMap<DatanodeInfo, Token<BlockTokenIdentifier>> locToToken =
          new HashMap<DatanodeInfo, Token<BlockTokenIdentifier>>();
      locToTokenList.add(locToToken);

      DatanodeInfo[] di = lb.getLocations();
      LocatedStripedBlock stripedBlk = (LocatedStripedBlock) lb;
      for (int i = 0; i < di.length; i++) {
        locToIndex.put(di[i], stripedBlk.getBlockIndices()[i]);
        locToToken.put(di[i], stripedBlk.getBlockTokens()[i]);
      }
    }
  }

  /**
   * Verify block index and token values. Must update block indices and block
   * tokens after sorting.
   */
  private void assertBlockIndexAndTokenPosition(List<LocatedBlock> lbs,
      List<HashMap<DatanodeInfo, Byte>> locToIndexList,
      List<HashMap<DatanodeInfo, Token<BlockTokenIdentifier>>> locToTokenList) {
    for (int i = 0; i < lbs.size(); i++) {
      LocatedBlock lb = lbs.get(i);
      LocatedStripedBlock stripedBlk = (LocatedStripedBlock) lb;
      HashMap<DatanodeInfo, Byte> locToIndex = locToIndexList.get(i);
      HashMap<DatanodeInfo, Token<BlockTokenIdentifier>> locToToken =
          locToTokenList.get(i);
      DatanodeInfo[] di = lb.getLocations();
      for (int j = 0; j < di.length; j++) {
        Assert.assertEquals("Block index value mismatches after sorting",
            (byte) locToIndex.get(di[j]), stripedBlk.getBlockIndices()[j]);
        Assert.assertEquals("Block token value mismatches after sorting",
            locToToken.get(di[j]), stripedBlk.getBlockTokens()[j]);
      }
    }
  }

  private List<DatanodeInfo> getDecommissionDatanode(DistributedFileSystem dfs,
      Path ecFile, int writeBytes, int decomNodeCount) throws IOException {
    ArrayList<DatanodeInfo> decommissionedNodes = new ArrayList<>();
    DatanodeInfo[] info = client.datanodeReport(DatanodeReportType.LIVE);
    BlockLocation[] fileBlockLocations = dfs.getFileBlockLocations(ecFile, 0,
        writeBytes);
    for (String dnName : fileBlockLocations[0].getNames()) {
      for (DatanodeInfo dn : info) {
        if (dnName.equals(dn.getXferAddr())) {
          decommissionedNodes.add(dn);
        }
        if (decommissionedNodes.size() >= decomNodeCount) {
          return decommissionedNodes;
        }
      }
    }
    return decommissionedNodes;
  }

  /* Get DFSClient to the namenode */
  private static DFSClient getDfsClient(NameNode nn, Configuration conf)
      throws IOException {
    return new DFSClient(nn.getNameNodeAddress(), conf);
  }

  private byte[] writeStripedFile(DistributedFileSystem fs, Path ecFile,
      int writeBytes) throws Exception {
    byte[] bytes = StripedFileTestUtil.generateBytes(writeBytes);
    DFSTestUtil.writeFile(fs, ecFile, new String(bytes));
    StripedFileTestUtil.waitBlockGroupsReported(fs, ecFile.toString());

    StripedFileTestUtil.checkData(fs, ecFile, writeBytes,
        new ArrayList<DatanodeInfo>(), null, blockGroupSize);
    return bytes;
  }

  private void writeConfigFile(Path name, List<String> nodes)
      throws IOException {
    // delete if it already exists
    if (localFileSys.exists(name)) {
      localFileSys.delete(name, true);
    }

    try (FSDataOutputStream stm = localFileSys.create(name)) {
      if (nodes != null) {
        for (String node: nodes) {
          stm.writeBytes(node);
          stm.writeBytes("\n");
        }
      }
    }
  }

  private void cleanupFile(FileSystem fileSys, Path name) throws IOException {
    assertTrue(fileSys.exists(name));
    fileSys.delete(name, true);
    assertTrue(!fileSys.exists(name));
  }

  /*
   * decommission the DN at index dnIndex or one random node if dnIndex is set
   * to -1 and wait for the node to reach the given {@code waitForState}.
   */
  private void decommissionNode(int nnIndex,
      List<DatanodeInfo> decommissionedNodes, AdminStates waitForState)
          throws IOException {
    DFSClient client = getDfsClient(cluster.getNameNode(nnIndex), conf);
    DatanodeInfo[] info = client.datanodeReport(DatanodeReportType.LIVE);

    // write nodename into the exclude file.
    ArrayList<String> excludeNodes = new ArrayList<String>();
    for (DatanodeInfo dn : decommissionedNodes) {
      boolean nodeExists = false;
      for (DatanodeInfo dninfo : info) {
        if (dninfo.getDatanodeUuid().equals(dn.getDatanodeUuid())) {
          nodeExists = true;
          break;
        }
      }
      assertTrue("Datanode: " + dn + " is not LIVE", nodeExists);
      excludeNodes.add(dn.getName());
      LOG.info("Decommissioning node: " + dn.getName());
    }
    writeConfigFile(excludeFile, excludeNodes);
    refreshNodes(cluster.getNamesystem(nnIndex), conf);
    for (DatanodeInfo dn : decommissionedNodes) {
      DatanodeInfo ret = NameNodeAdapter
          .getDatanode(cluster.getNamesystem(nnIndex), dn);
      waitNodeState(ret, waitForState);
    }
  }

  private static void refreshNodes(final FSNamesystem ns,
      final Configuration conf) throws IOException {
    ns.getBlockManager().getDatanodeManager().refreshNodes(conf);
  }

  /*
   * Wait till node is fully decommissioned.
   */
  private void waitNodeState(DatanodeInfo node, AdminStates state) {
    boolean done = state == node.getAdminState();
    while (!done) {
      LOG.info("Waiting for node " + node + " to change state to " + state
          + " current state: " + node.getAdminState());
      try {
        Thread.sleep(HEARTBEAT_INTERVAL * 500);
      } catch (InterruptedException e) {
        // nothing
      }
      done = state == node.getAdminState();
    }
    LOG.info("node " + node + " reached the state " + state);
  }

  /**
   * Verify that the number of replicas are as expected for each block in the
   * given file. For blocks with a decommissioned node, verify that their
   * replication is 1 more than what is specified. For blocks without
   * decommissioned nodes, verify their replication is equal to what is
   * specified.
   *
   * @param decommissionedNodes
   *          - if null, there is no decommissioned node for this file.
   * @return - null if no failure found, else an error message string.
   */
  private static String checkFile(FileSystem fileSys, Path name, int repl,
      List<DatanodeInfo> decommissionedNodes, int numDatanodes)
          throws IOException {
    boolean isNodeDown = decommissionedNodes.size() > 0;
    // need a raw stream
    assertTrue("Not HDFS:" + fileSys.getUri(),
        fileSys instanceof DistributedFileSystem);
    HdfsDataInputStream dis = (HdfsDataInputStream) fileSys.open(name);
    Collection<LocatedBlock> dinfo = dis.getAllBlocks();
    for (LocatedBlock blk : dinfo) { // for each block
      int hasdown = 0;
      DatanodeInfo[] nodes = blk.getLocations();
      for (int j = 0; j < nodes.length; j++) { // for each replica
        LOG.info("Block Locations size={}, locs={}, j=", nodes.length,
            nodes[j].toString(), j);
        boolean found = false;
        for (DatanodeInfo datanodeInfo : decommissionedNodes) {
          // check against decommissioned list
          if (isNodeDown
              && nodes[j].getXferAddr().equals(datanodeInfo.getXferAddr())) {
            found = true;
            hasdown++;
            // Downnode must actually be decommissioned
            if (!nodes[j].isDecommissioned()) {
              return "For block " + blk.getBlock() + " replica on " + nodes[j]
                  + " is given as downnode, " + "but is not decommissioned";
            }
            // Decommissioned node (if any) should only be last node in list.
            if (j < repl) {
              return "For block " + blk.getBlock() + " decommissioned node "
                  + nodes[j] + " was not last node in list: " + (j + 1) + " of "
                  + nodes.length;
            }
            // should only be last node in list.
            LOG.info("Block " + blk.getBlock() + " replica on " + nodes[j]
                + " is decommissioned.");
          }
        }
        // Non-downnodes must not be decommissioned
        if (!found && nodes[j].isDecommissioned()) {
          return "For block " + blk.getBlock() + " replica on " + nodes[j]
              + " is unexpectedly decommissioned";
        }
      }

      LOG.info("Block " + blk.getBlock() + " has " + hasdown
          + " decommissioned replica.");
      if (Math.min(numDatanodes, repl + hasdown) != nodes.length) {
        return "Wrong number of replicas for block " + blk.getBlock() + ": "
            + nodes.length + ", expected "
            + Math.min(numDatanodes, repl + hasdown);
      }
    }
    return null;
  }

  /**
   * Simulate that There are 2 nodes(dn0,dn1) in decommission. Firstly dn0
   * replicates in success, dn1 replicates in failure. Decommissions go on.
   */
  @Test (timeout = 120000)
  public void testDecommissionWithFailedReplicating() throws Exception {

    // Write ec file.
    Path ecFile = new Path(ecDir, "firstReplicationFailedFile");
    int writeBytes = cellSize * 6;
    writeStripedFile(dfs, ecFile, writeBytes);

    // Get 2 nodes of ec block and set them in decommission.
    // The 2 nodes are not in pendingNodes of DatanodeAdminManager.
    List<LocatedBlock> lbs = ((HdfsDataInputStream) dfs.open(ecFile))
        .getAllBlocks();
    LocatedStripedBlock blk = (LocatedStripedBlock) lbs.get(0);
    DatanodeInfo[] dnList = blk.getLocations();
    DatanodeDescriptor dn0 = bm.getDatanodeManager()
        .getDatanode(dnList[0].getDatanodeUuid());
    dn0.startDecommission();
    DatanodeDescriptor dn1 = bm.getDatanodeManager()
        .getDatanode(dnList[1].getDatanodeUuid());
    dn1.startDecommission();

    assertEquals(0, bm.getDatanodeManager().getDatanodeAdminManager()
        .getNumPendingNodes());

    // Replicate dn0 block to another dn
    // Simulate that dn0 replicates in success, dn1 replicates in failure.
    final byte blockIndex = blk.getBlockIndices()[0];
    final Block targetBlk = new Block(blk.getBlock().getBlockId() + blockIndex,
        cellSize, blk.getBlock().getGenerationStamp());
    DatanodeInfo extraDn = getDatanodeOutOfTheBlock(blk);
    DatanodeDescriptor target = bm.getDatanodeManager()
        .getDatanode(extraDn.getDatanodeUuid());
    dn0.addECBlockToBeReplicated(targetBlk,
        new DatanodeStorageInfo[] {target.getStorageInfos()[0]});

    // dn0 replicates in success
    GenericTestUtils.waitFor(
        () -> dn0.getNumberOfReplicateBlocks() == 0,
        100, 60000);
    GenericTestUtils.waitFor(
        () -> {
          Iterator<DatanodeStorageInfo> it =
              bm.getStoredBlock(targetBlk).getStorageInfos();
          while(it.hasNext()) {
            if (it.next().getDatanodeDescriptor().equals(target)) {
              return true;
            }
          }
          return false;
        },
        100, 60000);

    // There are 8 live replicas
    BlockInfoStriped blockInfo =
        (BlockInfoStriped)bm.getStoredBlock(
            new Block(blk.getBlock().getBlockId()));
    assertEquals(8, bm.countNodes(blockInfo).liveReplicas());

    // Add the 2 nodes to pendingNodes of DatanodeAdminManager
    bm.getDatanodeManager().getDatanodeAdminManager()
        .getPendingNodes().add(dn0);
    bm.getDatanodeManager().getDatanodeAdminManager()
        .getPendingNodes().add(dn1);

    waitNodeState(dn0, AdminStates.DECOMMISSIONED);
    waitNodeState(dn1, AdminStates.DECOMMISSIONED);

    // There are 9 live replicas
    assertEquals(9, bm.countNodes(blockInfo).liveReplicas());

    // After dn0 & dn1 decommissioned, all internal Blocks(0~8) are there
    Iterator<DatanodeStorageInfo> it = blockInfo.getStorageInfos();
    BitSet indexBitSet = new BitSet(9);
    while(it.hasNext()) {
      DatanodeStorageInfo storageInfo = it.next();
      if(storageInfo.getDatanodeDescriptor().equals(dn0)
          || storageInfo.getDatanodeDescriptor().equals(dn1)) {
        // Skip decommissioned nodes
        continue;
      }
      byte index = blockInfo.getStorageBlockIndex(storageInfo);
      indexBitSet.set(index);
    }
    for (int i = 0; i < 9; ++i) {
      assertEquals(true, indexBitSet.get(i));
    }
  }

  /**
   * Get a Datanode which does not contain the block.
   */
  private DatanodeInfo getDatanodeOutOfTheBlock(LocatedStripedBlock blk)
      throws Exception {
    DatanodeInfo[] allDnInfos = client.datanodeReport(DatanodeReportType.LIVE);
    DatanodeInfo[] blkDnInos= blk.getLocations();
    for (DatanodeInfo dnInfo : allDnInfos) {
      boolean in = false;
      for (DatanodeInfo blkDnInfo : blkDnInos) {
        if (blkDnInfo.equals(dnInfo)) {
          in = true;
        }
      }
      if(!in) {
        return dnInfo;
      }
    }
    return null;
  }

  @Test (timeout = 120000)
  public void testDecommissionWithMissingBlock() throws Exception {
    // Write ec file.
    Path ecFile = new Path(ecDir, "missingOneInternalBLockFile");
    int writeBytes = cellSize * 6;
    writeStripedFile(dfs, ecFile, writeBytes);

    final List<DatanodeInfo> decommisionNodes = new ArrayList<DatanodeInfo>();
    LocatedBlock lb = dfs.getClient().getLocatedBlocks(ecFile.toString(), 0)
        .get(0);
    LocatedStripedBlock lsb = (LocatedStripedBlock)lb;
    DatanodeInfo[] dnLocs = lsb.getLocations();
    BlockInfoStriped blockInfo =
        (BlockInfoStriped)bm.getStoredBlock(
            new Block(lsb.getBlock().getBlockId()));

    assertEquals(dataBlocks + parityBlocks, dnLocs.length);
    int decommNodeIndex = 1;
    int numDecommission= 4;
    int stopNodeIndex = 0;

    // Add the 4 nodes, and set the 4 nodes decommissioning.
    // So that they are decommissioning at the same time
    for (int i = decommNodeIndex; i < numDecommission + decommNodeIndex; ++i) {
      decommisionNodes.add(dnLocs[i]);
      DatanodeDescriptor dn = bm.getDatanodeManager()
          .getDatanode(dnLocs[i].getDatanodeUuid());
      dn.startDecommission();
    }
    GenericTestUtils.waitFor(
        () -> bm.countNodes(blockInfo).decommissioning() == numDecommission,
        100, 10000);

    // Namenode does not handle decommissioning nodes now
    assertEquals(0, bm.getDatanodeManager().getDatanodeAdminManager()
        .getNumPendingNodes());

    // Replicate dn1 block to another dn
    // So that one of the 4 replicas has been replicated.
    final byte blockIndex = lsb.getBlockIndices()[decommNodeIndex];
    final Block targetBlk = new Block(lsb.getBlock().getBlockId() + blockIndex,
        cellSize, lsb.getBlock().getGenerationStamp());
    DatanodeInfo extraDn = getDatanodeOutOfTheBlock(lsb);
    DatanodeDescriptor target = bm.getDatanodeManager()
        .getDatanode(extraDn.getDatanodeUuid());
    DatanodeDescriptor dnStartIndexDecommission = bm.getDatanodeManager()
        .getDatanode(dnLocs[decommNodeIndex].getDatanodeUuid());
    dnStartIndexDecommission.addECBlockToBeReplicated(targetBlk,
        new DatanodeStorageInfo[] {target.getStorageInfos()[0]});

    // Wait for replication success.
    GenericTestUtils.waitFor(
        () -> {
          Iterator<DatanodeStorageInfo> it =
              bm.getStoredBlock(targetBlk).getStorageInfos();
          while(it.hasNext()) {
            if (it.next().getDatanodeDescriptor().equals(target)) {
              return true;
            }
          }
          return false;
        },
        100, 60000);

    // Reopen ecFile, get the new locations.
    lb = dfs.getClient().getLocatedBlocks(ecFile.toString(), 0)
        .get(0);
    lsb = (LocatedStripedBlock)lb;
    DatanodeInfo[] newDnLocs = lsb.getLocations();

    // Now the block has 10 internal blocks.
    assertEquals(10, newDnLocs.length);

    // Stop the dn0(stopNodeIndex) datanode
    // So that the internal block from this dn misses
    DataNode dn = cluster.getDataNode(dnLocs[stopNodeIndex].getIpcPort());
    cluster.stopDataNode(dnLocs[stopNodeIndex].getXferAddr());
    cluster.setDataNodeDead(dn.getDatanodeId());

    // So far, there are 4 decommissioning nodes, 1 replica has been
    // replicated, and 1 replica misses. There are 8 total internal
    // blocks, 5 live and 3 decommissioning internal blocks.
    assertEquals(5, bm.countNodes(blockInfo).liveReplicas());
    assertEquals(3, bm.countNodes(blockInfo).decommissioning());

    // Handle decommission nodes in a new thread.
    // Verify that nodes are decommissioned.
    final CountDownLatch decomStarted = new CountDownLatch(0);
    new Thread(
        () -> {
          try {
            decomStarted.countDown();
            decommissionNode(0, decommisionNodes, AdminStates.DECOMMISSIONED);
          } catch (Exception e) {
            LOG.error("Exception while decommissioning", e);
            Assert.fail("Shouldn't throw exception!");
          }
        }).start();
    decomStarted.await(5, TimeUnit.SECONDS);

    // Wake up to reconstruct the block.
    BlockManagerTestUtil.wakeupPendingReconstructionTimerThread(bm);

    // Wait for decommissioning
    GenericTestUtils.waitFor(
        // Whether there are 8 live replicas after decommission.
        () -> bm.countNodes(blockInfo).liveReplicas() == 9,
        100, 60000);

    StripedFileTestUtil.checkData(dfs, ecFile, writeBytes, decommisionNodes,
        null, blockGroupSize);
    cleanupFile(dfs, ecFile);
  }

  @Test (timeout = 120000)
  public void testCountNodes() throws Exception{
    // Write ec file.
    Path ecFile = new Path(ecDir, "testCountNodes");
    int writeBytes = cellSize * 6;
    writeStripedFile(dfs, ecFile, writeBytes);

    List<LocatedBlock> lbs = ((HdfsDataInputStream) dfs.open(ecFile))
        .getAllBlocks();
    LocatedStripedBlock blk = (LocatedStripedBlock) lbs.get(0);
    DatanodeInfo[] dnList = blk.getLocations();
    DatanodeDescriptor dn0 = bm.getDatanodeManager()
        .getDatanode(dnList[0].getDatanodeUuid());
    dn0.startDecommission();

    // Replicate dn0 block to another dn
    final byte blockIndex = blk.getBlockIndices()[0];
    final Block targetBlk = new Block(blk.getBlock().getBlockId() + blockIndex,
        cellSize, blk.getBlock().getGenerationStamp());
    DatanodeInfo extraDn = getDatanodeOutOfTheBlock(blk);
    DatanodeDescriptor target = bm.getDatanodeManager()
        .getDatanode(extraDn.getDatanodeUuid());
    dn0.addECBlockToBeReplicated(targetBlk,
        new DatanodeStorageInfo[] {target.getStorageInfos()[0]});

    // dn0 replicates in success
    GenericTestUtils.waitFor(
        () -> dn0.getNumberOfReplicateBlocks() == 0,
        100, 60000);
    GenericTestUtils.waitFor(
        () -> {
          Iterator<DatanodeStorageInfo> it =
              bm.getStoredBlock(targetBlk).getStorageInfos();
          while(it.hasNext()) {
            if (it.next().getDatanodeDescriptor().equals(target)) {
              return true;
            }
          }
          return false;
        },
        100, 60000);

    // There are 9 live replicas, 0 decommissioning replicas.
    BlockInfoStriped blockInfo =
        (BlockInfoStriped)bm.getStoredBlock(
            new Block(blk.getBlock().getBlockId()));
    Iterator<BlockInfoStriped.StorageAndBlockIndex> it =
        blockInfo.getStorageAndIndexInfos().iterator();
    DatanodeStorageInfo decommissioningStorage = null;
    DatanodeStorageInfo liveStorage = null;
    while(it.hasNext()) {
      BlockInfoStriped.StorageAndBlockIndex si = it.next();
      if(si.getStorage().getDatanodeDescriptor().equals(dn0)) {
        decommissioningStorage = si.getStorage();
      }
      if(si.getStorage().getDatanodeDescriptor().equals(target)) {
        liveStorage = si.getStorage();
      }
    }
    assertNotNull(decommissioningStorage);
    assertNotNull(liveStorage);

    // Adjust internal block locations
    // [b0(decommissioning), b1, b2, b3, b4, b5, b6, b7, b8, b0(live)] changed
    // to [b0(live), b1, b2, b3, b4, b5, b6, b7, b8, b0(decommissioning)]
    BlockManagerTestUtil.removeStorage(blockInfo, decommissioningStorage);
    BlockManagerTestUtil.addStorage(blockInfo, liveStorage, targetBlk);
    BlockManagerTestUtil.addStorage(blockInfo, decommissioningStorage,
        targetBlk);
    assertEquals(0, bm.countNodes(blockInfo).decommissioning());
    assertEquals(9, bm.countNodes(blockInfo).liveReplicas());
    cleanupFile(dfs, ecFile);
  }

  /**
   * Test recovery for an ec block, its storage array contains these internal
   * blocks which are {b0, b1, b2, b3, null, b5, b6, b7, b8, b0, b1, b2,
   * b3}, array[0]{b0} in decommissioning, array[1-3]{b1, b2, b3} are
   * in decommissioned. array[4] is null, array[5-12]{b[5-8],b[0-3]} are
   * in live.
   */
  @Test (timeout = 120000)
  public void testRecoveryWithDecommission() throws Exception {
    final Path ecFile = new Path(ecDir, "testRecoveryWithDecommission");
    int writeBytes = cellSize * dataBlocks;
    byte[] originBytesArray = writeStripedFile(dfs, ecFile, writeBytes);
    List<LocatedBlock> lbs = ((HdfsDataInputStream) dfs.open(ecFile))
        .getAllBlocks();
    LocatedStripedBlock blk = (LocatedStripedBlock) lbs.get(0);
    DatanodeInfo[] dnList = blk.getLocations();
    BlockInfoStriped blockInfo =
        (BlockInfoStriped)bm.getStoredBlock(
            new Block(blk.getBlock().getBlockId()));

    // Decommission datanode dn0 contains block b0
    // Aim to add storageinfo of replicated block b0 to storages[9] of ec block
    List<DatanodeInfo> decommissionedNodes = new ArrayList<>();
    decommissionedNodes.add(dnList[0]);
    decommissionNode(0, decommissionedNodes, AdminStates.DECOMMISSIONED);

    // Now storages of ec block are (b0{decommissioned}, b[1-8]{live},
    // b0{live})
    assertEquals(9, bm.countNodes(blockInfo).liveReplicas());
    assertEquals(1, bm.countNodes(blockInfo).decommissioned());

    int decommissionNodesNum = 4;

    // Decommission nodes contain blocks of b[0-3]
    // dn0 has been decommissioned
    for (int i = 1; i < decommissionNodesNum; i++) {
      decommissionedNodes.add(dnList[i]);
    }
    decommissionNode(0, decommissionedNodes, AdminStates.DECOMMISSIONED);

    // Now storages of ec block are (b[0-3]{decommissioned}, b[4-8]{live},
    // b0{live}, b[1-3]{live})
    // There are 9 live and 4 decommissioned internal blocks
    assertEquals(9, bm.countNodes(blockInfo).liveReplicas());
    assertEquals(4, bm.countNodes(blockInfo).decommissioned());

    // There are no reconstruction tasks
    assertEquals(0, bm.getDatanodeManager().getDatanodeAdminManager()
        .getNumPendingNodes());
    assertEquals(0, bm.getUnderReplicatedNotMissingBlocks());

    // Set dn0 in decommissioning
    // So that the block on dn0 can be used for reconstruction task
    DatanodeDescriptor dn0 = bm.getDatanodeManager()
        .getDatanode(dnList[0].getDatanodeUuid());
    dn0.startDecommission();

    // Stop the datanode contains b4
    DataNode dn = cluster.getDataNode(
        dnList[decommissionNodesNum].getIpcPort());
    cluster.stopDataNode(dnList[decommissionNodesNum].getXferAddr());
    cluster.setDataNodeDead(dn.getDatanodeId());

    // Now storages of ec block are (b[0]{decommissioning},
    // b[1-3]{decommissioned}, null, b[5-8]{live}, b0{live}, b[1-3]{live})
    // There are 8 live and 1 decommissioning internal blocks
    // Wait for reconstruction EC block.
    GenericTestUtils.waitFor(
        () -> bm.countNodes(blockInfo).liveReplicas() == 9,
        100, 10000);

    byte[] readBytesArray = new byte[writeBytes];
    StripedFileTestUtil.verifyPread(dfs, ecFile, writeBytes,
        originBytesArray, readBytesArray, ecPolicy);
    cleanupFile(dfs, ecFile);
  }
}
