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

  final Configuration conf = new HdfsConfiguration();
  initConf(conf);
  // Sets bigger retry max attempts value so that test case will timed out if
  // block pinning errors are not handled properly during block movement.
  conf.setInt(DFSConfigKeys.DFS_MOVER_RETRY_MAX_ATTEMPTS_KEY, 10000);
  final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf)
      .numDataNodes(3)
      .build();
  try {
    cluster.waitActive();
    final DistributedFileSystem dfs = cluster.getFileSystem();
    final String file = "/testMoverWithPinnedBlocks/file";
    Path dir = new Path("/testMoverWithPinnedBlocks");
    dfs.mkdirs(dir);
    // write to DISK
    dfs.setStoragePolicy(dir, "HOT");
    final FSDataOutputStream out = dfs.create(new Path(file));
    byte[] fileData = StripedFileTestUtil
        .generateBytes(DEFAULT_BLOCK_SIZE * 3);
    out.write(fileData);
    out.close();
    // verify before movement
    LocatedBlock lb = dfs.getClient().getLocatedBlocks(file, 0).get(0);
    StorageType[] storageTypes = lb.getStorageTypes();
    for (StorageType storageType : storageTypes) {
      Assert.assertTrue(StorageType.DISK == storageType);
    }
    // Adding one SSD based data node to the cluster.
    StorageType[][] newtypes = new StorageType[][] {{StorageType.SSD}};
    startAdditionalDNs(conf, 1, newtypes, cluster);
    // Mock FsDatasetSpi#getPinning to show that the block is pinned.
    for (int i = 0; i < cluster.getDataNodes().size(); i++) {
      DataNode dn = cluster.getDataNodes().get(i);
      LOG.info("Simulate block pinning in datanode {}", dn);
      InternalDataNodeTestUtils.mockDatanodeBlkPinning(dn, true);
    }
    // move file blocks to ONE_SSD policy
    dfs.setStoragePolicy(dir, "ONE_SSD");
    int rc = ToolRunner.run(conf, new Mover.Cli(),
        new String[] {"-p", dir.toString()});
    int exitcode = ExitStatus.NO_MOVE_BLOCK.getExitCode();
    Assert.assertEquals("Movement should fail", exitcode, rc);
  } finally {
    cluster.shutdown();
  }
}