Mover { private ExitStatus run() { try { init(); boolean hasRemaining = new Processor().processNamespace(); return hasRemaining ? ExitStatus.IN_PROGRESS : ExitStatus.SUCCESS; } catch (IllegalArgumentException e) { System.out.println(e + ". Exiting ..."); return ExitStatus.ILLEGAL_ARGUMENTS; } catch (IOException e) { System.out.println(e + ". Exiting ..."); return ExitStatus.IO_EXCEPTION; } finally { dispatcher.shutdownNow(); } } Mover(NameNodeConnector nnc, Configuration conf, AtomicInteger retryCount); static void main(String[] args);  }
@Test public void testScheduleBlockWithinSameNode() throws Exception { final Configuration conf = new HdfsConfiguration(); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(3) .storageTypes( new StorageType[] { StorageType.DISK, StorageType.ARCHIVE }) .build(); try { cluster.waitActive(); final DistributedFileSystem dfs = cluster.getFileSystem(); final String file = "/testScheduleWithinSameNode/file"; Path dir = new Path("/testScheduleWithinSameNode"); dfs.mkdirs(dir); dfs.setStoragePolicy(dir, "HOT"); { final FSDataOutputStream out = dfs.create(new Path(file)); out.writeChars("testScheduleWithinSameNode"); out.close(); } LocatedBlock lb = dfs.getClient().getLocatedBlocks(file, 0).get(0); StorageType[] storageTypes = lb.getStorageTypes(); for (StorageType storageType : storageTypes) { Assert.assertTrue(StorageType.DISK == storageType); } dfs.setStoragePolicy(dir, "COLD"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] { "-p", dir.toString() }); Assert.assertEquals("Movement to ARCHIVE should be successfull", 0, rc); Thread.sleep(3000); lb = dfs.getClient().getLocatedBlocks(file, 0).get(0); storageTypes = lb.getStorageTypes(); for (StorageType storageType : storageTypes) { Assert.assertTrue(StorageType.ARCHIVE == storageType); } } finally { cluster.shutdown(); } }
@Test(timeout = 300000) public void testTwoReplicaSameStorageTypeShouldNotSelect() throws Exception { final Configuration conf = new HdfsConfiguration(); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(3) .storageTypes( new StorageType[][] { { StorageType.DISK, StorageType.ARCHIVE }, { StorageType.DISK, StorageType.DISK }, { StorageType.DISK, StorageType.ARCHIVE } }).build(); try { cluster.waitActive(); final DistributedFileSystem dfs = cluster.getFileSystem(); final String file = "/testForTwoReplicaSameStorageTypeShouldNotSelect"; final FSDataOutputStream out = dfs.create(new Path(file), (short) 2); out.writeChars("testForTwoReplicaSameStorageTypeShouldNotSelect"); out.close(); LocatedBlock lb = dfs.getClient().getLocatedBlocks(file, 0).get(0); StorageType[] storageTypes = lb.getStorageTypes(); for (StorageType storageType : storageTypes) { Assert.assertTrue(StorageType.DISK == storageType); } dfs.setStoragePolicy(new Path(file), "COLD"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] { "-p", file.toString() }); Assert.assertEquals("Movement to ARCHIVE should be successfull", 0, rc); Thread.sleep(3000); lb = dfs.getClient().getLocatedBlocks(file, 0).get(0); storageTypes = lb.getStorageTypes(); int archiveCount = 0; for (StorageType storageType : storageTypes) { if (StorageType.ARCHIVE == storageType) { archiveCount++; } } Assert.assertEquals(archiveCount, 2); } finally { cluster.shutdown(); } }
@Test public void testMoverFailedRetry() throws Exception { final Configuration conf = new HdfsConfiguration(); conf.set(DFSConfigKeys.DFS_MOVER_RETRY_MAX_ATTEMPTS_KEY, "2"); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(3) .storageTypes( new StorageType[][] {{StorageType.DISK, StorageType.ARCHIVE}, {StorageType.DISK, StorageType.ARCHIVE}, {StorageType.DISK, StorageType.ARCHIVE}}).build(); try { cluster.waitActive(); final DistributedFileSystem dfs = cluster.getFileSystem(); final String file = "/testMoverFailedRetry"; final FSDataOutputStream out = dfs.create(new Path(file), (short) 2); out.writeChars("testMoverFailedRetry"); out.close(); LocatedBlock lb = dfs.getClient().getLocatedBlocks(file, 0).get(0); cluster.corruptBlockOnDataNodesByDeletingBlockFile(lb.getBlock()); dfs.setStoragePolicy(new Path(file), "COLD"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] {"-p", file.toString()}); Assert.assertEquals("Movement should fail after some retry", ExitStatus.IO_EXCEPTION.getExitCode(), rc); } finally { cluster.shutdown(); } }