@Test(timeout=100000) public void testBalancerWithPinnedBlocks() throws Exception { assumeTrue(!Path.WINDOWS); final Configuration conf = new HdfsConfiguration(); initConf(conf); conf.setBoolean(DFS_DATANODE_BLOCK_PINNING_ENABLED, true); long[] capacities = new long[] { CAPACITY, CAPACITY }; String[] racks = { RACK0, RACK1 }; int numOfDatanodes = capacities.length; cluster = new MiniDFSCluster.Builder(conf).numDataNodes(capacities.length) .hosts(new String[]{"localhost", "localhost"}) .racks(racks).simulatedCapacities(capacities).build(); try { cluster.waitActive(); client = NameNodeProxies.createProxy(conf, cluster.getFileSystem(0).getUri(), ClientProtocol.class).getProxy(); long totalCapacity = sum(capacities); long totalUsedSpace = totalCapacity * 8 / 10; InetSocketAddress[] favoredNodes = new InetSocketAddress[numOfDatanodes]; for (int i = 0; i < favoredNodes.length; i++) { favoredNodes[i] = cluster.getDataNodes().get(i).getXferAddress(); } DFSTestUtil.createFile(cluster.getFileSystem(0), filePath, false, 1024, totalUsedSpace / numOfDatanodes, DEFAULT_BLOCK_SIZE, (short) numOfDatanodes, 0, false, favoredNodes); cluster.startDataNodes(conf, 1, true, null, new String[] { RACK2 }, new long[] { CAPACITY }); totalCapacity += CAPACITY; waitForHeartBeat(totalUsedSpace, totalCapacity, client, cluster); Collection<URI> namenodes = DFSUtil.getNsServiceRpcUris(conf); int r = Balancer.run(namenodes, Balancer.Parameters.DEFAULT, conf); assertEquals(ExitStatus.NO_MOVE_PROGRESS.getExitCode(), r); } finally { cluster.shutdown(); } }
static int run(Collection<URI> namenodes, final Parameters p, Configuration conf) throws IOException, InterruptedException { final long sleeptime = conf.getLong(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_DEFAULT) * 2000 + conf.getLong(DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_KEY, DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_DEFAULT) * 1000; LOG.info("namenodes = " + namenodes); LOG.info("parameters = " + p); System.out.println("Time Stamp Iteration# Bytes Already Moved Bytes Left To Move Bytes Being Moved"); List<NameNodeConnector> connectors = Collections.emptyList(); try { connectors = NameNodeConnector.newNameNodeConnectors(namenodes, Balancer.class.getSimpleName(), BALANCER_ID_PATH, conf, p.maxIdleIteration); boolean done = false; for(int iteration = 0; !done; iteration++) { done = true; Collections.shuffle(connectors); for(NameNodeConnector nnc : connectors) { final Balancer b = new Balancer(nnc, p, conf); final Result r = b.runOneIteration(); r.print(iteration, System.out); b.resetData(conf); if (r.exitStatus == ExitStatus.IN_PROGRESS) { done = false; } else if (r.exitStatus != ExitStatus.SUCCESS) { return r.exitStatus.getExitCode(); } } if (!done) { Thread.sleep(sleeptime); } } } finally { for(NameNodeConnector nnc : connectors) { IOUtils.cleanup(LOG, nnc); } } return ExitStatus.SUCCESS.getExitCode(); }
Balancer { static int run(Collection<URI> namenodes, final Parameters p, Configuration conf) throws IOException, InterruptedException { final long sleeptime = conf.getLong(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_DEFAULT) * 2000 + conf.getLong(DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_KEY, DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_DEFAULT) * 1000; LOG.info("namenodes = " + namenodes); LOG.info("parameters = " + p); System.out.println("Time Stamp Iteration# Bytes Already Moved Bytes Left To Move Bytes Being Moved"); List<NameNodeConnector> connectors = Collections.emptyList(); try { connectors = NameNodeConnector.newNameNodeConnectors(namenodes, Balancer.class.getSimpleName(), BALANCER_ID_PATH, conf, p.maxIdleIteration); boolean done = false; for(int iteration = 0; !done; iteration++) { done = true; Collections.shuffle(connectors); for(NameNodeConnector nnc : connectors) { final Balancer b = new Balancer(nnc, p, conf); final Result r = b.runOneIteration(); r.print(iteration, System.out); b.resetData(conf); if (r.exitStatus == ExitStatus.IN_PROGRESS) { done = false; } else if (r.exitStatus != ExitStatus.SUCCESS) { return r.exitStatus.getExitCode(); } } if (!done) { Thread.sleep(sleeptime); } } } finally { for(NameNodeConnector nnc : connectors) { IOUtils.cleanup(LOG, nnc); } } return ExitStatus.SUCCESS.getExitCode(); } }
Balancer { static int run(Collection<URI> namenodes, final Parameters p, Configuration conf) throws IOException, InterruptedException { final long sleeptime = conf.getLong(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_DEFAULT) * 2000 + conf.getLong(DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_KEY, DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_DEFAULT) * 1000; LOG.info("namenodes = " + namenodes); LOG.info("parameters = " + p); System.out.println("Time Stamp Iteration# Bytes Already Moved Bytes Left To Move Bytes Being Moved"); List<NameNodeConnector> connectors = Collections.emptyList(); try { connectors = NameNodeConnector.newNameNodeConnectors(namenodes, Balancer.class.getSimpleName(), BALANCER_ID_PATH, conf, p.maxIdleIteration); boolean done = false; for(int iteration = 0; !done; iteration++) { done = true; Collections.shuffle(connectors); for(NameNodeConnector nnc : connectors) { final Balancer b = new Balancer(nnc, p, conf); final Result r = b.runOneIteration(); r.print(iteration, System.out); b.resetData(conf); if (r.exitStatus == ExitStatus.IN_PROGRESS) { done = false; } else if (r.exitStatus != ExitStatus.SUCCESS) { return r.exitStatus.getExitCode(); } } if (!done) { Thread.sleep(sleeptime); } } } finally { for(NameNodeConnector nnc : connectors) { IOUtils.cleanup(LOG, nnc); } } return ExitStatus.SUCCESS.getExitCode(); } Balancer(NameNodeConnector theblockpool, Parameters p, Configuration conf); }
Balancer { static int run(Collection<URI> namenodes, final Parameters p, Configuration conf) throws IOException, InterruptedException { final long sleeptime = conf.getLong(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_DEFAULT) * 2000 + conf.getLong(DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_KEY, DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_DEFAULT) * 1000; LOG.info("namenodes = " + namenodes); LOG.info("parameters = " + p); System.out.println("Time Stamp Iteration# Bytes Already Moved Bytes Left To Move Bytes Being Moved"); List<NameNodeConnector> connectors = Collections.emptyList(); try { connectors = NameNodeConnector.newNameNodeConnectors(namenodes, Balancer.class.getSimpleName(), BALANCER_ID_PATH, conf, p.maxIdleIteration); boolean done = false; for(int iteration = 0; !done; iteration++) { done = true; Collections.shuffle(connectors); for(NameNodeConnector nnc : connectors) { final Balancer b = new Balancer(nnc, p, conf); final Result r = b.runOneIteration(); r.print(iteration, System.out); b.resetData(conf); if (r.exitStatus == ExitStatus.IN_PROGRESS) { done = false; } else if (r.exitStatus != ExitStatus.SUCCESS) { return r.exitStatus.getExitCode(); } } if (!done) { Thread.sleep(sleeptime); } } } finally { for(NameNodeConnector nnc : connectors) { IOUtils.cleanup(LOG, nnc); } } return ExitStatus.SUCCESS.getExitCode(); } Balancer(NameNodeConnector theblockpool, Parameters p, Configuration conf); static void main(String[] args); }
Balancer { static int run(Collection<URI> namenodes, final Parameters p, Configuration conf) throws IOException, InterruptedException { final long sleeptime = conf.getLong(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_DEFAULT) * 2000 + conf.getLong(DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_KEY, DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_DEFAULT) * 1000; LOG.info("namenodes = " + namenodes); LOG.info("parameters = " + p); System.out.println("Time Stamp Iteration# Bytes Already Moved Bytes Left To Move Bytes Being Moved"); List<NameNodeConnector> connectors = Collections.emptyList(); try { connectors = NameNodeConnector.newNameNodeConnectors(namenodes, Balancer.class.getSimpleName(), BALANCER_ID_PATH, conf, p.maxIdleIteration); boolean done = false; for(int iteration = 0; !done; iteration++) { done = true; Collections.shuffle(connectors); for(NameNodeConnector nnc : connectors) { final Balancer b = new Balancer(nnc, p, conf); final Result r = b.runOneIteration(); r.print(iteration, System.out); b.resetData(conf); if (r.exitStatus == ExitStatus.IN_PROGRESS) { done = false; } else if (r.exitStatus != ExitStatus.SUCCESS) { return r.exitStatus.getExitCode(); } } if (!done) { Thread.sleep(sleeptime); } } } finally { for(NameNodeConnector nnc : connectors) { IOUtils.cleanup(LOG, nnc); } } return ExitStatus.SUCCESS.getExitCode(); } Balancer(NameNodeConnector theblockpool, Parameters p, Configuration conf); static void main(String[] args);  }