Balancer { Result runOneIteration() { try { final List<DatanodeStorageReport> reports = dispatcher.init(); final long bytesLeftToMove = init(reports); if (bytesLeftToMove == 0) { return newResult(ExitStatus.SUCCESS, bytesLeftToMove, 0); } else { LOG.info( "Need to move "+ StringUtils.byteDesc(bytesLeftToMove) + " to make the cluster balanced." ); } if (!runDuringUpgrade && nnc.isUpgrading()) { System.err.println("Balancer exiting as upgrade is not finalized, " + "please finalize the HDFS upgrade before running the balancer."); LOG.error("Balancer exiting as upgrade is not finalized, " + "please finalize the HDFS upgrade before running the balancer."); return newResult(ExitStatus.UNFINALIZED_UPGRADE, bytesLeftToMove, -1); } final long bytesBeingMoved = chooseStorageGroups(); if (bytesBeingMoved == 0) { System.out.println("No block can be moved. Exiting..."); return newResult(ExitStatus.NO_MOVE_BLOCK, bytesLeftToMove, bytesBeingMoved); } else { LOG.info( "Will move " + StringUtils.byteDesc(bytesBeingMoved) + " in this iteration"); } if (!dispatcher.dispatchAndCheckContinue()) { return newResult(ExitStatus.NO_MOVE_PROGRESS, bytesLeftToMove, bytesBeingMoved); } return newResult(ExitStatus.IN_PROGRESS, bytesLeftToMove, bytesBeingMoved); } catch (IllegalArgumentException e) { System.out.println(e + ". Exiting ..."); return newResult(ExitStatus.ILLEGAL_ARGUMENTS); } catch (IOException e) { System.out.println(e + ". Exiting ..."); return newResult(ExitStatus.IO_EXCEPTION); } catch (InterruptedException e) { System.out.println(e + ". Exiting ..."); return newResult(ExitStatus.INTERRUPTED); } finally { dispatcher.shutdownNow(); } } Balancer(NameNodeConnector theblockpool, BalancerParameters p,
      Configuration conf); static void main(String[] args);  }
@Test(timeout = 100000) public void testMaxIterationTime() throws Exception { final Configuration conf = new HdfsConfiguration(); initConf(conf); int blockSize = 10*1024*1024; conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, blockSize); conf.setInt(DFSConfigKeys.DFS_BYTES_PER_CHECKSUM_KEY, blockSize); conf.setInt(DFSConfigKeys.DFS_BALANCER_MOVERTHREADS_KEY, 1); conf.setLong(DFSConfigKeys.DFS_DATANODE_BALANCE_BANDWIDTHPERSEC_KEY, 64 * 1024); conf.setLong(DFSConfigKeys.DFS_CLIENT_SOCKET_TIMEOUT_KEY, 2000L); conf.setLong(DFSConfigKeys.DFS_BALANCER_MAX_ITERATION_TIME_KEY, 2000L); final long capacity = 10L * blockSize; final long[] dnCapacities = new long[] {capacity, capacity}; final short rep = 1; final long seed = 0xFAFAFA; cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(0) .build(); try { cluster.getConfiguration(0).setInt(DFSConfigKeys.DFS_REPLICATION_KEY, 1); conf.setInt(DFSConfigKeys.DFS_REPLICATION_KEY, 1); cluster.startDataNodes(conf, 1, true, null, null, dnCapacities); cluster.waitClusterUp(); cluster.waitActive(); final Path path = new Path("/testMaxIterationTime.dat"); DistributedFileSystem fs = cluster.getFileSystem(); DFSTestUtil.createFile(fs, path, 4L * blockSize, rep, seed); cluster.startDataNodes(conf, 1, true, null, null, dnCapacities); cluster.triggerHeartbeats(); List<NameNodeConnector> connectors = Collections.emptyList(); try { BalancerParameters bParams = BalancerParameters.DEFAULT; connectors = NameNodeConnector.newNameNodeConnectors( DFSUtil.getInternalNsRpcUris(conf), Balancer.class.getSimpleName(), Balancer.BALANCER_ID_PATH, conf, bParams.getMaxIdleIteration()); for (NameNodeConnector nnc : connectors) { LOG.info("NNC to work on: " + nnc); Balancer b = new Balancer(nnc, bParams, conf); long startTime = Time.monotonicNow(); Result r = b.runOneIteration(); long runtime = Time.monotonicNow() - startTime; assertEquals("We expect ExitStatus.IN_PROGRESS to be reported.", ExitStatus.IN_PROGRESS, r.exitStatus); assertTrue("Unexpected iteration runtime: " + runtime + "ms > 3.5s", runtime < 3500); } } finally { for (NameNodeConnector nnc : connectors) { IOUtils.cleanupWithLogger(null, nnc); } } } finally { cluster.shutdown(true, true); } }