@Test(timeout = 90000)
public void testMoverFailedRetryWithPinnedBlocks() throws Exception {

  final Configuration conf = new HdfsConfiguration();
  initConf(conf);
  conf.set(DFSConfigKeys.DFS_MOVER_RETRY_MAX_ATTEMPTS_KEY, "2");
  final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf)
      .numDataNodes(2)
      .storageTypes(
          new StorageType[][] {{StorageType.DISK, StorageType.ARCHIVE},
              {StorageType.DISK, StorageType.ARCHIVE}}).build();
  try {
    cluster.waitActive();
    final DistributedFileSystem dfs = cluster.getFileSystem();
    final String parenDir = "/parent";
    dfs.mkdirs(new Path(parenDir));
    final String file1 = "/parent/testMoverFailedRetryWithPinnedBlocks1";
    // write to DISK
    final FSDataOutputStream out = dfs.create(new Path(file1), (short) 2);
    byte[] fileData = StripedFileTestUtil
        .generateBytes(DEFAULT_BLOCK_SIZE * 2);
    out.write(fileData);
    out.close();
    // Adding pinned blocks.
    createFileWithFavoredDatanodes(conf, cluster, dfs);
    // Delete block file so, block move will fail with FileNotFoundException
    LocatedBlocks locatedBlocks = dfs.getClient().getLocatedBlocks(file1, 0);
    Assert.assertEquals("Wrong block count", 2,
        locatedBlocks.locatedBlockCount());
    LocatedBlock lb = locatedBlocks.get(0);
    cluster.corruptBlockOnDataNodesByDeletingBlockFile(lb.getBlock());
    // move to ARCHIVE
    dfs.setStoragePolicy(new Path(parenDir), "COLD");
    int rc = ToolRunner.run(conf, new Mover.Cli(),
        new String[] {"-p", parenDir.toString()});
    Assert.assertEquals("Movement should fail after some retry",
        ExitStatus.NO_MOVE_PROGRESS.getExitCode(), rc);
  } finally {
    cluster.shutdown();
  }
}