DFSAdmin extends FsShell { int startReconfiguration(String nodeType, String address) throws IOException { if ("datanode".equals(nodeType)) { ClientDatanodeProtocol dnProxy = getDataNodeProxy(address); dnProxy.startReconfiguration(); System.out.println("Started reconfiguration task on DataNode " + address); return 0; } else { System.err.println("Node type " + nodeType + " does not support reconfiguration."); return 1; } } DFSAdmin();  DFSAdmin(Configuration conf); void report(String[] argv, int i); void setSafeMode(String[] argv, int idx); int setStoragePolicy(String[] argv); int getStoragePolicy(String[] argv); int triggerBlockReport(String[] argv); void allowSnapshot(String[] argv); void disallowSnapshot(String[] argv); int saveNamespace(); int rollEdits(); int restoreFailedStorage(String arg); int refreshNodes(); int setBalancerBandwidth(String[] argv, int idx); int fetchImage(final String[] argv, final int idx); int finalizeUpgrade(); int metaSave(String[] argv, int idx); int printTopology(); int refreshServiceAcl(); int refreshUserToGroupsMappings(); int refreshSuperUserGroupsConfiguration(); int refreshCallQueue(); int reconfig(String[] argv, int i); int genericRefresh(String[] argv, int i); @Override int run(String[] argv); static void main(String[] argv);  }
@Test(timeout = 30000) public void testGetReconfigureStatus() throws IOException, InterruptedException { ReconfigurationUtil ru = mock(ReconfigurationUtil.class); datanode.setReconfigurationUtil(ru); List<ReconfigurationUtil.PropertyChange> changes = new ArrayList<ReconfigurationUtil.PropertyChange>(); File newDir = new File(cluster.getDataDirectory(), "data_new"); newDir.mkdirs(); changes.add(new ReconfigurationUtil.PropertyChange( DFS_DATANODE_DATA_DIR_KEY, newDir.toString(), datanode.getConf().get(DFS_DATANODE_DATA_DIR_KEY))); changes.add(new ReconfigurationUtil.PropertyChange( "randomKey", "new123", "old456")); when(ru.parseChangedProperties(any(Configuration.class), any(Configuration.class))).thenReturn(changes); final int port = datanode.getIpcPort(); final String address = "localhost:" + port; assertThat(admin.startReconfiguration("datanode", address), is(0)); List<String> outputs = null; int count = 100; while (count > 0) { outputs = getReconfigureStatus("datanode", address); if (!outputs.isEmpty() && outputs.get(0).contains("finished")) { break; } count--; Thread.sleep(100); } assertTrue(count > 0); assertThat(outputs.size(), is(8)); List<StorageLocation> locations = DataNode.getStorageLocations( datanode.getConf()); assertThat(locations.size(), is(1)); assertThat(locations.get(0).getFile(), is(newDir)); assertTrue(new File(newDir, Storage.STORAGE_DIR_CURRENT).isDirectory()); int successOffset = outputs.get(1).startsWith("SUCCESS:") ? 1 : 5; int failedOffset = outputs.get(1).startsWith("FAILED:") ? 1: 4; assertThat(outputs.get(successOffset), containsString("Change property " + DFS_DATANODE_DATA_DIR_KEY)); assertThat(outputs.get(successOffset + 1), is(allOf(containsString("From:"), containsString("data1"), containsString("data2")))); assertThat(outputs.get(successOffset + 2), is(not(anyOf(containsString("data1"), containsString("data2"))))); assertThat(outputs.get(successOffset + 2), is(allOf(containsString("To"), containsString("data_new")))); assertThat(outputs.get(failedOffset), containsString("Change property randomKey")); assertThat(outputs.get(failedOffset + 1), containsString("From: \"old456\"")); assertThat(outputs.get(failedOffset + 2), containsString("To: \"new123\"")); }