DFSAdmin extends FsShell { int getReconfigurationStatus(String nodeType, String address, PrintStream out, PrintStream err) throws IOException { if ("datanode".equals(nodeType)) { ClientDatanodeProtocol dnProxy = getDataNodeProxy(address); try { ReconfigurationTaskStatus status = dnProxy.getReconfigurationStatus(); out.print("Reconfiguring status for DataNode[" + address + "]: "); if (!status.hasTask()) { out.println("no task was found."); return 0; } out.print("started at " + new Date(status.getStartTime())); if (!status.stopped()) { out.println(" and is still running."); return 0; } out.println(" and finished at " + new Date(status.getEndTime()).toString() + "."); for (Map.Entry<PropertyChange, Optional<String>> result : status.getStatus().entrySet()) { if (!result.getValue().isPresent()) { out.print("SUCCESS: "); } else { out.print("FAILED: "); } out.printf("Change property %s%n\tFrom: \"%s\"%n\tTo: \"%s\"%n", result.getKey().prop, result.getKey().oldVal, result.getKey().newVal); if (result.getValue().isPresent()) { out.println("\tError: " + result.getValue().get() + "."); } } } catch (IOException e) { err.println("DataNode reloading configuration: " + e + "."); return 1; } } else { err.println("Node type " + nodeType + " does not support reconfiguration."); return 1; } return 0; } DFSAdmin();  DFSAdmin(Configuration conf); void report(String[] argv, int i); void setSafeMode(String[] argv, int idx); int refreshNodes(); int setBalancerBandwidth(String[] argv, int idx); int triggerBlockReport(String[] argv); 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 testGetReconfigurationStatus() throws IOException, InterruptedException { ReconfigurationUtil ru = mock(ReconfigurationUtil.class); datanode.setReconfigurationUtil(ru); List<ReconfigurationUtil.PropertyChange> changes = new ArrayList<>(); 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(4)); 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 offset = 1; assertThat(outputs.get(offset), containsString("SUCCESS: Change property " + DFS_DATANODE_DATA_DIR_KEY)); assertThat(outputs.get(offset + 1), is(allOf(containsString("From:"), containsString("data_0_0"), containsString("data_0_1")))); assertThat(outputs.get(offset + 2), is(not(anyOf(containsString("data1"), containsString("data2"))))); assertThat(outputs.get(offset + 2), is(allOf(containsString("To"), containsString("data_new")))); }