@Test(timeout=300000)
public void testBalancerWithRamDisk() throws Exception {

  final int SEED = 0xFADED;
  final short REPL_FACT = 1;
  Configuration conf = new Configuration();
  initConfWithRamDisk(conf);
  final int defaultRamDiskCapacity = 10;
  final long ramDiskStorageLimit =
    ((long) defaultRamDiskCapacity * DEFAULT_RAM_DISK_BLOCK_SIZE) +
    (DEFAULT_RAM_DISK_BLOCK_SIZE - 1);
  final long diskStorageLimit =
    ((long) defaultRamDiskCapacity * DEFAULT_RAM_DISK_BLOCK_SIZE) +
    (DEFAULT_RAM_DISK_BLOCK_SIZE - 1);
  cluster = new MiniDFSCluster
    .Builder(conf)
    .numDataNodes(1)
    .storageCapacities(new long[] { ramDiskStorageLimit, diskStorageLimit })
    .storageTypes(new StorageType[] { RAM_DISK, DEFAULT })
    .build();
  try {
    cluster.waitActive();
    // Create few files on RAM_DISK
    final String METHOD_NAME = GenericTestUtils.getMethodName();
    final Path path1 = new Path("/" + METHOD_NAME + ".01.dat");
    final Path path2 = new Path("/" + METHOD_NAME + ".02.dat");
    DistributedFileSystem fs = cluster.getFileSystem();
    DFSClient client = fs.getClient();
    DFSTestUtil.createFile(fs, path1, true,
      DEFAULT_RAM_DISK_BLOCK_SIZE, 4 * DEFAULT_RAM_DISK_BLOCK_SIZE,
      DEFAULT_RAM_DISK_BLOCK_SIZE, REPL_FACT, SEED, true);
    DFSTestUtil.createFile(fs, path2, true,
      DEFAULT_RAM_DISK_BLOCK_SIZE, 1 * DEFAULT_RAM_DISK_BLOCK_SIZE,
      DEFAULT_RAM_DISK_BLOCK_SIZE, REPL_FACT, SEED, true);
    // Sleep for a short time to allow the lazy writer thread to do its job
    Thread.sleep(6 * 1000);
    // Add another fresh DN with the same type/capacity without files on RAM_DISK
    StorageType[][] storageTypes = new StorageType[][] {{RAM_DISK, DEFAULT}};
    long[][] storageCapacities = new long[][]{{ramDiskStorageLimit, diskStorageLimit}};
    cluster.startDataNodes(conf, REPL_FACT, storageTypes, true, null,
      null, null, storageCapacities, null, false, false, false, null);
    cluster.triggerHeartbeats();
    Collection<URI> namenodes = DFSUtil.getInternalNsRpcUris(conf);
    // Run Balancer
    Balancer.Parameters p = new Balancer.Parameters(
      Parameters.DEFAULT.policy,
      Parameters.DEFAULT.threshold,
      Parameters.DEFAULT.nodesToBeExcluded,
      Parameters.DEFAULT.nodesToBeIncluded);
    final int r = Balancer.run(namenodes, p, conf);
    // Validate no RAM_DISK block should be moved
    assertEquals(ExitStatus.NO_MOVE_PROGRESS.getExitCode(), r);
    // Verify files are still on RAM_DISK
    DFSTestUtil.verifyFileReplicasOnStorageType(fs, client, path1, RAM_DISK);
    DFSTestUtil.verifyFileReplicasOnStorageType(fs, client, path2, RAM_DISK);
  } finally {
    cluster.shutdown();
  }
}