@Override
  boolean addStorage(DatanodeStorageInfo storage, Block reportedBlock) {
    Preconditions.checkArgument(BlockIdManager.isStripedBlockID(
        reportedBlock.getBlockId()), "reportedBlock is not striped");
    Preconditions.checkArgument(BlockIdManager.convertToStripedID(
        reportedBlock.getBlockId()) == this.getBlockId(),
        "reported blk_%s does not belong to the group of stored blk_%s",
        reportedBlock.getBlockId(), this.getBlockId());
    int blockIndex = BlockIdManager.getBlockIndex(reportedBlock);
    int index = blockIndex;
    DatanodeStorageInfo old = getStorageInfo(index);
    if (old != null && !old.equals(storage)) { // over replicated
      // check if the storage has been stored
      int i = findStorageInfo(storage);
      if (i == -1) {
        index = findSlot();
      } else {
        return true;
      }
    }
    addStorage(storage, index, blockIndex);
    return true;
  }
@Test
  public void testAddStorage() {
    // first add NUM_DATA_BLOCKS + NUM_PARITY_BLOCKS storages, i.e., a complete
    // group of blocks/storages
    DatanodeStorageInfo[] storageInfos = DFSTestUtil.createDatanodeStorageInfos(
        totalBlocks);
    Block[] blocks = createReportedBlocks(totalBlocks);
    int i = 0;
    for (; i < storageInfos.length; i += 2) {
      info.addStorage(storageInfos[i], blocks[i]);
      Assert.assertEquals(i/2 + 1, info.numNodes());
    }
    i /= 2;
    for (int j = 1; j < storageInfos.length; j += 2) {
      Assert.assertTrue(info.addStorage(storageInfos[j], blocks[j]));
      Assert.assertEquals(i + (j+1)/2, info.numNodes());
    }

    // check
    byte[] indices = (byte[]) Whitebox.getInternalState(info, "indices");
    Assert.assertEquals(totalBlocks, info.getCapacity());
    Assert.assertEquals(totalBlocks, indices.length);
    i = 0;
    for (DatanodeStorageInfo storage : storageInfos) {
      int index = info.findStorageInfo(storage);
      Assert.assertEquals(i++, index);
      Assert.assertEquals(index, indices[index]);
    }

    // the same block is reported from the same storage twice
    i = 0;
    for (DatanodeStorageInfo storage : storageInfos) {
      Assert.assertTrue(info.addStorage(storage, blocks[i++]));
    }
    Assert.assertEquals(totalBlocks, info.getCapacity());
    Assert.assertEquals(totalBlocks, info.numNodes());
    Assert.assertEquals(totalBlocks, indices.length);
    i = 0;
    for (DatanodeStorageInfo storage : storageInfos) {
      int index = info.findStorageInfo(storage);
      Assert.assertEquals(i++, index);
      Assert.assertEquals(index, indices[index]);
    }

    // the same block is reported from another storage
    DatanodeStorageInfo[] storageInfos2 =
        DFSTestUtil.createDatanodeStorageInfos(totalBlocks * 2);
    // only add the second half of info2
    for (i = totalBlocks; i < storageInfos2.length; i++) {
      info.addStorage(storageInfos2[i], blocks[i % totalBlocks]);
      Assert.assertEquals(i + 1, info.getCapacity());
      Assert.assertEquals(i + 1, info.numNodes());
      indices = (byte[]) Whitebox.getInternalState(info, "indices");
      Assert.assertEquals(i + 1, indices.length);
    }
    for (i = totalBlocks; i < storageInfos2.length; i++) {
      int index = info.findStorageInfo(storageInfos2[i]);
      Assert.assertEquals(i++, index);
      Assert.assertEquals(index - totalBlocks, indices[index]);
    }
  }