Mover { private ExitStatus run() { try { init(); return new Processor().processNamespace().getExitStatus(); } 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 testScheduleBlockWithinSameNode() throws Exception { final Configuration conf = new HdfsConfiguration(); initConf(conf); 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(); initConf(conf); 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(timeout = 300000) public void testMoveWhenStoragePolicyNotSatisfying() throws Exception { final Configuration conf = new HdfsConfiguration(); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(3) .storageTypes( new StorageType[][] { { StorageType.DISK }, { StorageType.DISK }, { StorageType.DISK } }).build(); try { cluster.waitActive(); final DistributedFileSystem dfs = cluster.getFileSystem(); final String file = "/testMoveWhenStoragePolicyNotSatisfying"; final FSDataOutputStream out = dfs.create(new Path(file)); out.writeChars("testMoveWhenStoragePolicyNotSatisfying"); out.close(); dfs.setStoragePolicy(new Path(file), "COLD"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] { "-p", file.toString() }); int exitcode = ExitStatus.NO_MOVE_BLOCK.getExitCode(); Assert.assertEquals("Exit code should be " + exitcode, exitcode, rc); } finally { cluster.shutdown(); } }
@Test public void testMoverFailedRetry() 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(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.NO_MOVE_PROGRESS.getExitCode(), rc); } finally { cluster.shutdown(); } }
@Test(timeout = 300000) public void testMoverWithStripedFile() throws Exception { final Configuration conf = new HdfsConfiguration(); initConfWithStripe(conf); int numOfDatanodes =10; int storagesPerDatanode=2; long capacity = 10 * DEFAULT_STRIPE_BLOCK_SIZE; long[][] capacities = new long[numOfDatanodes][storagesPerDatanode]; for (int i = 0; i < numOfDatanodes; i++) { for(int j=0;j<storagesPerDatanode;j++){ capacities[i][j]=capacity; } } final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(numOfDatanodes) .storagesPerDatanode(storagesPerDatanode) .storageTypes(new StorageType[][]{ {StorageType.DISK, StorageType.DISK}, {StorageType.DISK, StorageType.DISK}, {StorageType.DISK, StorageType.DISK}, {StorageType.DISK, StorageType.DISK}, {StorageType.DISK, StorageType.DISK}, {StorageType.DISK, StorageType.ARCHIVE}, {StorageType.DISK, StorageType.ARCHIVE}, {StorageType.DISK, StorageType.ARCHIVE}, {StorageType.DISK, StorageType.ARCHIVE}, {StorageType.DISK, StorageType.ARCHIVE}}) .storageCapacities(capacities) .build(); try { cluster.waitActive(); ClientProtocol client = NameNodeProxies.createProxy(conf, cluster.getFileSystem(0).getUri(), ClientProtocol.class).getProxy(); String barDir = "/bar"; client.mkdirs(barDir, new FsPermission((short) 777), true); client.setStoragePolicy(barDir, HdfsConstants.HOT_STORAGE_POLICY_NAME); client.setErasureCodingPolicy(barDir, null); final String fooFile = "/bar/foo"; long fileLen = 20 * DEFAULT_STRIPE_BLOCK_SIZE ; DFSTestUtil.createFile(cluster.getFileSystem(), new Path(fooFile), fileLen,(short) 3, 0); LocatedBlocks locatedBlocks = client.getBlockLocations(fooFile, 0, fileLen); for(LocatedBlock lb : locatedBlocks.getLocatedBlocks()){ for( StorageType type : lb.getStorageTypes()){ Assert.assertEquals(StorageType.DISK, type); } } StripedFileTestUtil.verifyLocatedStripedBlocks(locatedBlocks, dataBlocks + parityBlocks); numOfDatanodes +=5; capacities = new long[5][storagesPerDatanode]; for (int i = 0; i < 5; i++) { for(int j=0;j<storagesPerDatanode;j++){ capacities[i][j]=capacity; } } cluster.startDataNodes(conf, 5, new StorageType[][]{ {StorageType.ARCHIVE, StorageType.ARCHIVE}, {StorageType.ARCHIVE, StorageType.ARCHIVE}, {StorageType.ARCHIVE, StorageType.ARCHIVE}, {StorageType.ARCHIVE, StorageType.ARCHIVE}, {StorageType.ARCHIVE, StorageType.ARCHIVE}}, true, null, null, null,capacities, null, false, false, false, null); cluster.triggerHeartbeats(); client.setStoragePolicy(barDir, "COLD"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] { "-p", barDir }); Assert.assertEquals("Movement to ARCHIVE should be successfull", 0, rc); locatedBlocks = client.getBlockLocations(fooFile, 0, fileLen); for(LocatedBlock lb : locatedBlocks.getLocatedBlocks()){ for( StorageType type : lb.getStorageTypes()){ Assert.assertEquals(StorageType.ARCHIVE, type); } } StripedFileTestUtil.verifyLocatedStripedBlocks(locatedBlocks, dataBlocks + parityBlocks); }finally{ cluster.shutdown(); } }