Index: src/java/org/apache/hadoop/dfs/Balancer.java
===================================================================
--- src/java/org/apache/hadoop/dfs/Balancer.java	(revision 0)
+++ src/java/org/apache/hadoop/dfs/Balancer.java	(revision 0)
@@ -0,0 +1,1230 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.dfs;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.DataOutput;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.dfs.BlocksWithLocations.BlockWithLocations;
+import org.apache.hadoop.dfs.FSConstants.DatanodeReportType;
+import org.apache.hadoop.fs.FsShell;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.io.retry.RetryProxy;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetworkTopology;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+
+/** A tool to balance the cluster */
+public class Balancer implements Tool {
+  private static final Log LOG = 
+    LogFactory.getLog("org.apache.hadoop.dfs.Balancer");
+  final private static long MAX_BLOCKS_SIZE_TO_FETCH = 2*1024*1024*1024L; //2GB
+
+  private Configuration conf;
+
+  private double threshold = 10D;
+  private NamenodeProtocol namenode;
+  private ClientProtocol client;
+  
+  // all data node lists
+  private Collection<Source> overUtilizedDatanodes
+                               = new LinkedList<Source>();
+  private Collection<Source> aboveAvgUtilizedDatanodes
+                               = new LinkedList<Source>();
+  private Collection<BalancerDatanode> belowAvgUtilizedDatanodes
+                               = new LinkedList<BalancerDatanode>();
+  private Collection<BalancerDatanode> underUtilizedDatanodes
+                               = new LinkedList<BalancerDatanode>();
+  
+  private Collection<Source> sources
+                               = new HashSet<Source>();
+  private Collection<BalancerDatanode> targets
+                               = new HashSet<BalancerDatanode>();
+  
+  private Map<Block, BalancerBlock> blockList
+                 = new HashMap<Block, BalancerBlock>();
+  private Map<String, BalancerDatanode> datanodes
+                 = new HashMap<String, BalancerDatanode>();
+  
+  private NetworkTopology cluster = new NetworkTopology();
+  
+  private double avgUtilization = 0.0D;
+  
+  /* This class keeps track of a scheduled block move */
+  private class PendingBlockMove {
+    private BalancerBlock block;
+    private Source source;
+    private BalancerDatanode proxySource;
+    private BalancerDatanode target;
+    
+    /** constructor */
+    private PendingBlockMove() {
+    }
+    
+    /* Return source */
+    private Source getSource() {
+      return source;
+    }
+    
+    /* Return target */
+    private BalancerDatanode getTarget() {
+      return target;
+    }
+    
+    /* Return target */
+    private BalancerBlock getBlock() {
+      return block;
+    }
+    
+    /* choose a block & a proxy source for this pendingMove 
+     * whose souce & target have already been chosen.
+     * 
+     * Return true if a block is chosen; false otherwise
+     */
+    private boolean chooseBlock() {
+      // iterate all source's blocks until find a good one    
+      for(Iterator<BalancerBlock> blocks=
+        source.getBlockIterator(); blocks.hasNext();) {
+        if(isGoodBlock(blocks.next())) {
+          blocks.remove();
+          return true;
+        }
+      }
+      return false;
+    }
+    
+    /* Decide if the given block is good for the tentatvie move
+     * A block is good iff
+     * 1. it is a good candidate; see isGoodBlockCandidate
+     * 2. can find a proxy source for this move
+     */
+    private boolean isGoodBlock(BalancerBlock block) {
+      synchronized(block) {
+        synchronized(movedBlocks) {
+          if(isGoodBlockCandidate(source, target, block)) {
+            this.block = block;
+            if( chooseProxySource() ) {
+              addToMoved(block);
+              LOG.info("Decided to move block "+ block.getBlockId()
+                  +" with a length of "+FsShell.byteDesc(block.getNumBytes())
+                  + " bytes from " + source.getName() 
+                  + " to " + target.getName()
+                  + " using proxy source " + proxySource.getName() );
+              return true;
+            }
+          }
+        }
+      }
+      return false;
+    }
+    
+    /* Now we find out source, target, and block, we need to find a proxy
+     * 
+     * @return true if a proxy is found; otherwise false
+     */
+    private boolean chooseProxySource() {
+      // check if there is replica which is on the same rack with the target
+      for(BalancerDatanode loc:block.getLocations()) {
+        if(cluster.isOnSameRack(loc.getDatanode(), target.getDatanode())) {
+          if(loc.addPendingBlock(this)) {
+            proxySource = loc;
+            return true;
+          }
+        }
+      }
+      // find out a non-busy replica
+      for(BalancerDatanode loc:block.getLocations()) {
+        if(loc.addPendingBlock(this)) {
+          proxySource = loc;
+          return true;
+        }
+      }
+      return false;
+    }
+    
+    /* Dispatch the block move task to the proxy source & wait for the response
+     */
+    private void dispatch() {
+      Socket sock = new Socket();
+      DataOutputStream out = null;
+      DataInputStream in = null;
+      try {
+        sock.connect(DataNode.createSocketAddr(
+            proxySource.datanode.getName()), FSConstants.READ_TIMEOUT);
+        //System.out.println("Dispatching block "+block.getBlockId());
+        out = new DataOutputStream( new BufferedOutputStream(
+            sock.getOutputStream(), FSConstants.BUFFER_SIZE));
+        long start = FSNamesystem.now();
+        sendRequest(out);
+        in = new DataInputStream( new BufferedInputStream(
+            sock.getInputStream(), FSConstants.BUFFER_SIZE));
+        receiveResponse(in);
+        bytesMoved.inc(block.getNumBytes());
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Moving block "+block.getBlock().getBlockId()+
+              " from "+source.getName() + " to " +
+              target.getName() + " through " +
+              proxySource.getName() +
+              " succeeded: " +
+              "taken " + (FSNamesystem.now()-start)+" ms");
+        }
+      } catch (IOException e) {
+        LOG.warn("Error moving block "+block.getBlockId()+
+            " from "+source.getName() + " to " +
+            target.getName() + " through " +
+            proxySource.getName() +
+            ": "+e.getMessage()+ "\n" +
+            StringUtils.stringifyException(e) );
+      } finally {
+        IOUtils.closeStream(out);
+        IOUtils.closeStream(in);
+        IOUtils.closeSocket(sock);
+        
+        proxySource.removePendingBlock(this);
+        synchronized(target) {
+          target.removePendingBlock(this);
+        }
+
+        synchronized (this ) {
+          reset();
+        }
+        synchronized (Balancer.this) {
+          Balancer.this.notifyAll();
+        }
+      }
+    }
+    
+    /* Send a block copy request to the outputstream*/
+    private void sendRequest(DataOutputStream out) throws IOException {
+      out.writeShort(FSConstants.DATA_TRANFER_VERSION);
+      out.writeByte(FSConstants.OP_COPY_BLOCK);
+      out.writeLong(block.getBlock().getBlockId());
+      Text.writeString(out, source.getStorageID());
+      target.write(out);
+      out.flush();
+    }
+    
+    /* Receive a block copy response from the input stream */ 
+    private void receiveResponse(DataInputStream in) throws IOException {
+      short status = in.readShort();
+      if (status != FSConstants.OP_STATUS_SUCCESS) {
+        throw new IOException("Moving block "+block.getBlockId()+
+            " from "+source.getName() + " to " +
+            target.getName() + " through " +
+            proxySource.getName() +
+        "failed");
+      }
+    }
+
+    /* reset the object */
+    private void reset() {
+      block = null;
+      source = null;
+      proxySource = null;
+      target = null;
+    }
+    
+    /* start a thread to dispatch the block move */
+    private void moveBlock() {
+      BlockMover blockMover = new BlockMover();
+      blockMover.setDaemon(true);
+      blockMover.start();
+    }
+    
+    /* A thread for moving a block */
+    private class BlockMover extends Thread {
+      BlockMover() {
+      }
+      
+      public void run() {
+        dispatch();
+      }
+    }
+  }
+  
+  /* A class for keeping track of blocks in the Balancer */
+  private Map<Block, BalancerBlock> movedBlocks 
+             = new HashMap<Block, BalancerBlock>();
+
+  static private class BalancerBlock {
+    private Block block; // the block
+    private List<BalancerDatanode> locations
+            = new ArrayList<BalancerDatanode>(3); // its locations
+    
+    /* Constructor */
+    private BalancerBlock(Block block) {
+      this.block = block;
+    }
+    
+    /* clean block locations */
+    private synchronized void clearLocations() {
+      locations.clear();
+    }
+    
+    /* add a location */
+    private synchronized void addLocation(BalancerDatanode datanode) {
+      if(!locations.contains(datanode)) {
+        locations.add(datanode);
+      }
+    }
+    
+    /* Return if the block is located on <code>datanode</code> */
+    private synchronized boolean isLocatedOnDatanode(BalancerDatanode datanode) {
+      return locations.contains(datanode);
+    }
+    
+    /* Return its locations */
+    private synchronized List<BalancerDatanode> getLocations() {
+      return locations;
+    }
+    
+    /* Return the block */
+    private Block getBlock() {
+      return block;
+    }
+    
+    /* Return the block id */
+    private long getBlockId() {
+      return block.getBlockId();
+    }
+    
+    /* Return the length of the block */
+    private long getNumBytes() {
+      return block.getNumBytes();
+    }
+  }
+  
+  /* A class that represents a node to be involved in balancing 
+   * and the total number of bytes to be moved from/to this node.
+   */
+  static private class NodeTask {
+    private BalancerDatanode datanode; //source or target node
+    private long size;  //bytes scheduled to move
+    
+    /* constructor */
+    private NodeTask(BalancerDatanode datanode, long size) {
+      this.datanode = datanode;
+      this.size = size;
+    }
+    
+    /* Get the node */
+    private BalancerDatanode getDatanode() {
+      return datanode;
+    }
+    
+    /* Get the number of bytes that need to be moved */
+    private long getSize() {
+      return size;
+    }
+  }
+  
+  /* Return the utilization of a datanode */
+  static private double getUtilization(DatanodeInfo datanode) {
+    return ((double)datanode.getDfsUsed())/datanode.getCapacity()*100;
+  }
+  
+  /* A class that keeps track of a datanode in Balancer */
+  private static class BalancerDatanode implements Writable {
+    final private static long MAX_SIZE_TO_MOVE = 2*1024*1024*1024L; //2GB
+    final protected static short MAX_NUM_CONCURRENT_MOVES =
+      DataNode.MAX_BALANCING_THREADS;
+    protected DatanodeInfo datanode;
+    private double utilization;
+    protected long sizeToMove;
+    protected long scheduledSize = 0L;
+    //  blocks being moved but not confirmed yet
+    private List<PendingBlockMove> pendingBlocks = 
+      new ArrayList<PendingBlockMove>(MAX_NUM_CONCURRENT_MOVES); 
+    
+    /* Constructor 
+     * Depending on avgutil & threshold, decide how many bytes to move 
+     * for the next iteration
+     */
+    private BalancerDatanode(
+        DatanodeInfo node, double avgUtil, double threshold) {
+      datanode = node;
+      utilization = Balancer.getUtilization(node);
+        
+      if(utilization >= avgUtil+threshold
+          || utilization <= avgUtil-threshold) { 
+        sizeToMove = (long)(threshold*datanode.getCapacity()/100);
+      } else {
+        sizeToMove = 
+          (long)(Math.abs(avgUtil-utilization)*datanode.getCapacity()/100);
+      }
+      if(this.utilization < avgUtil ) {
+        long remaining = datanode.getRemaining();
+        sizeToMove = (remaining<sizeToMove)?remaining:sizeToMove;
+      }
+      sizeToMove = (MAX_SIZE_TO_MOVE<sizeToMove)?MAX_SIZE_TO_MOVE:sizeToMove;
+    }
+    
+    /** Get the datanode */
+    protected DatanodeInfo getDatanode() {
+      return datanode;
+    }
+    
+    /** Get the name of the datanode */
+    protected String getName() {
+      return datanode.getName();
+    }
+    
+    /* Get the storage id of the datanode */
+    protected String getStorageID() {
+      return datanode.getStorageID();
+    }
+    
+    /** Decide if still need to move more bytes */
+    protected boolean needMove() {
+      return scheduledSize<sizeToMove;
+    }
+
+    /** Return the total number of bytes that need to be moved */
+    protected long availableSizeToMove() {
+      return sizeToMove-scheduledSize;
+    }
+    
+    /* Add a node task's size to the node */
+    protected void addNodeTask(NodeTask task) {
+      scheduledSize += task.getSize();
+    }
+    
+    /* Check if the node can schedule more blocks to move */
+    private boolean isNotBusy() {
+      if( pendingBlocks.size() < MAX_NUM_CONCURRENT_MOVES ) {
+        return true;
+      }
+      return false;
+    }
+    
+    /* Check if all the dispatched moves are done */
+    private boolean isMoveCompleted() {
+      return pendingBlocks.isEmpty();
+    }
+    
+    /* Add a scheduled block move to the node */
+    private synchronized boolean addPendingBlock(
+        PendingBlockMove pendingBlock) {
+      if( isNotBusy() ) {
+        return pendingBlocks.add(pendingBlock);
+      }
+      return false;
+    }
+    
+    /* Remove a scheduled block move from the node */
+    private synchronized boolean  removePendingBlock(
+        PendingBlockMove pendingBlock) {
+      return pendingBlocks.remove(pendingBlock);
+    }
+
+    /** The following two methods support the Writable interface */
+    /** Deserialize */
+    public void readFields(DataInput in) throws IOException {
+      datanode.readFields(in);
+    }
+
+    /** Serialize */
+    public void write(DataOutput out) throws IOException {
+      datanode.write(out);
+    }
+  }
+  
+  /** A node that can be the sources of a block move */
+  private class Source extends BalancerDatanode {
+    
+    /* A thread that initates a block move 
+     * and waits for block move to complete */
+    private class Dispatcher extends Thread {
+      public void run() {
+        Source.this.dispatchBlocks();
+      }
+    }
+    
+    private ArrayList<NodeTask> nodeTasks = new ArrayList<NodeTask>(2);
+    private long blocksToReceive = 0L;
+    private List<BalancerBlock> blocks
+            = new ArrayList<BalancerBlock>();
+    
+    /* constructor */
+    private Source(DatanodeInfo node, double avgUtil, double threshold) {
+      super(node, avgUtil, threshold);
+    }
+    
+    /** Add a node task */
+    protected void addNodeTask(NodeTask task) {
+      super.addNodeTask(task);
+      nodeTasks.add(task);
+    }
+    
+    /* Return an iterator to this source's blocks */
+    private Iterator<BalancerBlock> getBlockIterator() {
+      return blocks.iterator();
+    }
+    
+    /* fetch new blocks of this source from namenode and
+     *  update this source's block list & the global block list
+     */
+    private void getBlockList() throws IOException {
+      BlockWithLocations[] newBlocks = namenode.getBlocks(datanode, 
+        (long)Math.min(MAX_BLOCKS_SIZE_TO_FETCH, blocksToReceive)).getBlocks();
+      for(BlockWithLocations blk:newBlocks) {
+        blocksToReceive -= blk.getBlock().getNumBytes();
+        BalancerBlock block;
+        synchronized(blockList) {
+          block = blockList.get(blk.getBlock());
+          if(block==null) {
+            block = new BalancerBlock(blk.getBlock());
+            blockList.put(blk.getBlock(), block);
+          } else {
+            block.clearLocations();
+          }
+        
+          synchronized (block) {
+            // update locations
+            for ( String location :blk.getDatanodes() ) {
+              BalancerDatanode datanode = datanodes.get(location);
+              if (datanode != null) { // not an unknown datanode
+                block.addLocation(datanode);
+              }
+            }
+          }
+          if(!blocks.contains(block) && isGoodBlockCandidate(block)) {
+            // filter bad candidates
+            blocks.add(block);
+          }
+        }
+      }
+    }
+
+    /* Decide if the given block is a good candiate to move or not */
+    private boolean isGoodBlockCandidate(BalancerBlock block) {
+      for(NodeTask nodeTask:nodeTasks) {
+        if(Balancer.this.isGoodBlockCandidate(this, nodeTask.datanode, block)) {
+          return true;
+        }
+      }
+      return false;
+    }
+
+    /* Return a block that's good for move */
+    private PendingBlockMove chooseBlock() {
+      for( Iterator<NodeTask> tasks=nodeTasks.iterator(); tasks.hasNext(); ) {
+        NodeTask task = tasks.next();
+        BalancerDatanode target = task.getDatanode();
+        PendingBlockMove pendingBlock = new PendingBlockMove();
+        if( target.addPendingBlock(pendingBlock) ) { 
+          // target is not busy, so do a tentative block allocation
+          pendingBlock.source = this;
+          pendingBlock.target = target;
+          if( pendingBlock.chooseBlock() ) {
+            long blockSize = pendingBlock.block.getNumBytes(); 
+            scheduledSize -= blockSize;
+            task.size -= blockSize;
+            if(task.size == 0) {
+              tasks.remove();
+            }
+            return pendingBlock;
+          } else {
+            // cancel the tentative move
+            target.removePendingBlock(pendingBlock);
+          }
+        }
+      }
+      return null;
+    }
+
+    /* iterate all source's blocks to remove moved ones */    
+    private void filterMovedBlocks() {
+      for(Iterator<BalancerBlock> blocks=getBlockIterator();
+            blocks.hasNext();) {
+        if(isMoved(blocks.next())) {
+          blocks.remove();
+        }
+      }
+    }
+    
+    /* Return if should fetch more blocks from namenode */
+    private boolean shouldFetchMoreBlocks() {
+      return blocks.size()<5 && blocksToReceive>0;
+    }
+    
+    /* This method iteratively does the following:
+     * it first selects a block to move,
+     * then sends a request to the proxy source to start the block move
+     * when the source's block list falls below a threshold, it asks
+     * the namenode for more blocks.
+     * It terminates when it has dispatch enough block move tasks or
+     * it has received enough blocks from the namenode
+     */ 
+    private void dispatchBlocks() {
+      // the number of consecutive iterations without block move
+      this.blocksToReceive = 2*scheduledSize;
+      while(scheduledSize>0 && (!blocks.isEmpty() || blocksToReceive>0)) {
+        PendingBlockMove pendingBlock = chooseBlock();
+        if(pendingBlock != null){
+          // move the block
+          pendingBlock.moveBlock();
+        } else { // no blocks to choose from
+          filterMovedBlocks(); // filter already moved blocks
+          if (shouldFetchMoreBlocks()) {// fetch new blocks
+            try {
+              getBlockList();
+            } catch (IOException e) {
+              LOG.warn(StringUtils.stringifyException(e));
+              return;
+            }
+          } else { // wait for targets to be idle
+            try {
+              synchronized(Balancer.this) {
+                Balancer.this.wait(1000);
+              }
+            } catch (InterruptedException ignored) {
+            }
+          }
+        }
+      }
+    }
+  }
+  
+  /** Default constructor */
+  Balancer() {
+  }
+  
+  /** Construct a balancer from the given configuration */
+  Balancer(Configuration conf) {
+    setConf(conf);
+  } 
+
+  /** Construct a balancer from the given configuration and threshold */
+  Balancer(Configuration conf, double threshold) {
+    setConf(conf);
+    this.threshold = threshold;
+  }
+
+  /**
+   * Run a balancer
+   * @param args
+   */
+  public static void main(String[] args) {
+    try {
+      System.exit( ToolRunner.run(null, new Balancer(), args) );
+    } catch (Throwable e) {
+      LOG.error(StringUtils.stringifyException(e));
+      System.exit(-1);
+    }
+
+  }
+
+  private static void printUsage() {
+    System.out.println("Usage: java Balancer");
+    System.out.println("          [-threshold <threshold>]\t" 
+        +"percentage of disk capacity");
+  }
+  
+  void init(String[] args) throws IOException {
+    // parse argument to get the threshold
+    int argsLen = (args == null) ? 0 : args.length;
+    if(argsLen==0) {
+      threshold = 10;
+    } else {
+      if(argsLen != 2 || !"-threshold".equalsIgnoreCase(args[0])) {
+        printUsage();
+        System.exit(-1);
+      } else {
+        try {
+          threshold = Double.parseDouble(args[1]);
+          if(threshold < 0) {
+            throw new NumberFormatException();
+          }
+          LOG.info( "Using a threshold of " + threshold );
+        } catch(NumberFormatException e) {
+          System.err.println(
+              "Expect a double parameter in the range of [0, 100]: "+ args[1]);
+          printUsage();
+          System.exit(-1);
+        }
+      }
+    }
+
+    // get name node address 
+    InetSocketAddress nameNodeAddr = DataNode.createSocketAddr(
+        conf.get("fs.default.name", "local"));
+    // connect to name node
+    this.namenode = createNamenode(nameNodeAddr, conf);
+    this.client = DFSClient.createNamenode(nameNodeAddr, conf);
+  }
+  
+  /* Build a NamenodeProtocol connection to the namenode and
+   * set up the retry policy */ 
+  private static NamenodeProtocol createNamenode(
+      InetSocketAddress nameNodeAddr, Configuration conf) throws IOException {
+    RetryPolicy timeoutPolicy = RetryPolicies.exponentialBackoffRetry(
+        5, 200, TimeUnit.MILLISECONDS);
+    Map<Class<? extends Exception>,RetryPolicy> exceptionToPolicyMap =
+      new HashMap<Class<? extends Exception>, RetryPolicy>();
+    RetryPolicy methodPolicy = RetryPolicies.retryByException(
+        timeoutPolicy, exceptionToPolicyMap);
+    Map<String,RetryPolicy> methodNameToPolicyMap = new HashMap<String,RetryPolicy>();
+    methodNameToPolicyMap.put("getBlocks", methodPolicy);
+
+    return (NamenodeProtocol) RetryProxy.create(
+        NamenodeProtocol.class,
+        RPC.waitForProxy(NamenodeProtocol.class,
+                         NamenodeProtocol.versionID,
+                         nameNodeAddr, 
+                         conf),
+        methodNameToPolicyMap);
+    
+  }
+  
+  /* get all live datanodes of a cluster and their disk usage
+   * decide the number of bytes need to be moved
+   */
+  private long initNodes() throws IOException {
+    return initNodes(client.getDatanodeReport(DatanodeReportType.LIVE));
+  }
+  
+  /* For the given data node, build a network topology, decide
+   * overloaded datanodes, above average loaded datanodes, below average loaded
+   * datanodes, and underloaded datanodes. Return the total number of bytes
+   * that are needed to move to make the cluster balanced.
+   */
+  private long initNodes(DatanodeInfo[] datanodes) {
+    // compute average utilization
+    long totalCapacity=0L, totalUsedSpace=0L;
+    for(DatanodeInfo datanode:datanodes) {
+      totalCapacity += datanode.getCapacity();
+      totalUsedSpace += datanode.getDfsUsed();
+    }
+    this.avgUtilization = ((double)totalUsedSpace)/totalCapacity*100;
+
+    /*create network topology and all data node lists: 
+     * overloaded, above-average, below-average, and underloaded 
+     */  
+    long overLoadedBytes = 0L, underLoadedBytes = 0L;
+    for(DatanodeInfo datanode:datanodes) {
+      cluster.add(datanode);
+      BalancerDatanode datanodeS;
+      if(getUtilization(datanode) > avgUtilization) {
+        datanodeS = new Source(datanode, avgUtilization, threshold);
+        if( isAboveAvgUtilized(datanodeS)) {
+          this.aboveAvgUtilizedDatanodes.add((Source)datanodeS);
+        } else if( isOverUtilized(datanodeS) ) {
+          this.overUtilizedDatanodes.add((Source)datanodeS);
+          overLoadedBytes += (long)((datanodeS.utilization-avgUtilization
+              -threshold)*datanodeS.datanode.getCapacity()/100.0);
+        }
+      } else {
+        datanodeS = new BalancerDatanode(datanode, avgUtilization, threshold);
+        if( isBelowAvgUtilized(datanodeS)) {
+          this.belowAvgUtilizedDatanodes.add(datanodeS);
+        } else if( isUnderUtilized(datanodeS) ) {
+          this.underUtilizedDatanodes.add(datanodeS);
+          underLoadedBytes += (long)((avgUtilization-threshold-
+              datanodeS.utilization)*datanodeS.datanode.getCapacity()/100.0);
+        }
+      }
+      this.datanodes.put(datanode.getStorageID(), datanodeS);
+    }
+    
+    //logging
+    logImbalancedNodes();
+    
+    // return number of bytes to be moved in order to make the cluster balanced
+    return (overLoadedBytes>underLoadedBytes)?overLoadedBytes:underLoadedBytes;
+  }
+
+  private void logImbalancedNodes() {
+    StringBuilder msg = new StringBuilder();
+    msg.append(overUtilizedDatanodes.size());
+    msg.append(" over utilized nodes:");
+    for(Source node:overUtilizedDatanodes) {
+      msg.append( " " );
+      msg.append( node.getName() );
+    }
+    LOG.info(msg);
+    msg = new StringBuilder();
+    msg.append(underUtilizedDatanodes.size());
+    msg.append(" under utilized nodes: ");
+    for(BalancerDatanode node:underUtilizedDatanodes) {
+      msg.append( " " );
+      msg.append( node.getName() );
+    }
+    LOG.info(msg);
+  }
+  
+  private enum NodeType {
+    OnRackOffBand, OnRackInBand, OffRackOffBand, OffRackInBand
+  }
+  
+  /* Decide all <source, target> pairs and
+   * the number of bytes to move from a source to a target
+   * Return total number of bytes to move in this iteration
+   */
+  private long chooseNodes() {
+    // matching nodes in the following priority: 
+    // 1. on rack over/under utilized nodes
+    chooseNodes(NodeType.OnRackOffBand);
+    // 2. on rack above/below average utilized nodes
+    chooseNodes(NodeType.OnRackInBand);
+    // 3. off rack over/under utilized nodes
+    chooseNodes(NodeType.OffRackOffBand);
+    // 4. off rack above/below average utilized nodes
+    chooseNodes(NodeType.OffRackInBand);
+    
+    long bytesToMove = 0L;
+    for(Source src:sources) {
+      bytesToMove += src.scheduledSize;
+    }
+    return bytesToMove;
+  }
+  
+  /* Decide all <source, target> pairs and
+   * the number of bytes to move from a source to a target for each node type
+   */
+  private boolean chooseNodes( NodeType nodeType) {
+    // first choose targets for all overutilized nodes
+    boolean nodeChosen = chooseTargets(nodeType);
+    // then choose sources for remaining underutilized nodes
+    return chooseSources(nodeType) || nodeChosen;
+  }
+  
+  /* choose targets of type <code>nodeType</code> */
+  private boolean chooseTargets( NodeType nodeType ) {
+    boolean nodeChosen = false;
+    for(Iterator<Source> iterator = overUtilizedDatanodes.iterator();
+        iterator.hasNext();) {
+      Source source = iterator.next();
+      Iterator<BalancerDatanode> candidates = null;;
+      switch( nodeType ) {
+      case OnRackOffBand:
+      case OffRackOffBand:  // choose from underUtilized nodes
+        candidates = underUtilizedDatanodes.iterator();
+        break;
+      case OnRackInBand:
+      case OffRackInBand:   // choose from belowAvgUtilized nodes
+        candidates = belowAvgUtilizedDatanodes.iterator();
+        break;
+      }
+      while( chooseTarget(source, candidates, nodeType) ) {
+        nodeChosen = true;
+      }
+      if(!source.needMove()) {
+        iterator.remove();
+      }
+    }
+    return nodeChosen;
+  }
+  
+  /* choose sources of type <code>nodeType</code> */ 
+  private boolean chooseSources(NodeType nodeType) {
+    boolean nodeChosen = false;
+    for(Iterator<BalancerDatanode> iterator = 
+      underUtilizedDatanodes.iterator(); iterator.hasNext();) {
+      BalancerDatanode target = iterator.next();
+      Iterator<Source> candidates = null;;
+      switch( nodeType ) {
+      case OnRackOffBand:
+      case OffRackOffBand:  // choose from overUtilized nodes
+        candidates = overUtilizedDatanodes.iterator();
+        break;
+      case OnRackInBand:
+      case OffRackInBand:   // choose from aboveAvgUtilized nodes
+        candidates = aboveAvgUtilizedDatanodes.iterator();
+        break;
+      }
+      while( chooseSource(target, candidates, nodeType) ) {
+        nodeChosen = true;
+      }
+      if(!target.needMove()) {
+        iterator.remove();
+      }
+    }
+    return nodeChosen;
+  }
+
+  /* choose a target of type <code>nodeType</code> for <code>source</code>
+   * from candidates <code>candidates</code>
+   */
+  private boolean chooseTarget(Source source,
+      Iterator<BalancerDatanode> candidates, NodeType nodeType) {
+    if(!source.needMove()) {
+      return false;
+    }
+    boolean foundTarget = false;
+    BalancerDatanode target = null;
+    while(!foundTarget && candidates.hasNext()) {
+      target = candidates.next();
+      if( !target.needMove()) {
+        candidates.remove();
+        continue;
+      }
+      switch( nodeType ) {
+      case OnRackOffBand:
+      case OnRackInBand:    // choose from on-rack nodes
+        if( cluster.isOnSameRack(source.datanode, target.datanode)) {
+          foundTarget = true;
+        }
+        break;
+      case OffRackOffBand:
+      case OffRackInBand:   // choose from off-rack nodes
+        if( !cluster.isOnSameRack(source.datanode, target.datanode)) {
+          foundTarget = true;
+        }
+        break;
+      }
+    }
+    if(foundTarget) {
+      assert(target != null):"Choose a null target";
+      long size = source.availableSizeToMove();
+      long targetSize = target.availableSizeToMove();
+      size = (size>targetSize)?targetSize:size;
+      NodeTask nodeTask = new NodeTask(target, size);
+      source.addNodeTask(nodeTask);
+      target.addNodeTask(nodeTask);
+      sources.add(source);
+      targets.add(target);
+      if( !target.needMove()) {
+        candidates.remove();
+      }
+      LOG.info("Decided to move "+FsShell.byteDesc(size)+" bytes from "
+          +source.datanode.getName() + " to " + target.datanode.getName());
+      return true;
+    }
+    return false;
+  }
+  
+  /* choose a source of type <code>nodeType</code> for <code>target</code>
+   * from candidates <code>candidates</code>
+   */
+  private boolean chooseSource(BalancerDatanode target,
+      Iterator<Source> candidates, NodeType nodeType) {
+    if(!target.needMove()) {
+      return false;
+    }
+    boolean foundSource = false;
+    Source source = null;
+    while(!foundSource && candidates.hasNext()) {
+      source = candidates.next();
+      if( !source.needMove()) {
+        candidates.remove();
+        continue;
+      }
+      switch( nodeType ) {
+      case OnRackOffBand:
+      case OnRackInBand:    // choose from on-rack nodes
+        if( cluster.isOnSameRack(source.getDatanode(), target.getDatanode())) {
+          foundSource = true;
+        }
+        break;
+      case OffRackOffBand:
+      case OffRackInBand:   // choose from off-rack nodes
+        if( !cluster.isOnSameRack(source.datanode, target.datanode)) {
+          foundSource = true;
+        }
+        break;
+      }
+    }
+    if(foundSource) {
+      assert(target != null):"Choose a null target";
+      long size = source.availableSizeToMove();
+      long targetSize = target.availableSizeToMove();
+      size = (size>targetSize)?targetSize:size;
+      NodeTask nodeTask = new NodeTask(target, size);
+      source.addNodeTask(nodeTask);
+      target.addNodeTask(nodeTask);
+      sources.add(source);
+      targets.add(target);
+      if( !source.needMove()) {
+        candidates.remove();
+      }
+      LOG.info("Decided to move "+FsShell.byteDesc(size)+" bytes from "
+          +source.datanode.getName() + " to " + target.datanode.getName());
+      return true;
+    }
+    return false;
+  }
+
+  /* Start a thread to dispatch block moves for each source. 
+   * The thread selects blocks to move & sends request to proxy source to
+   * initiate block move. The process is flow controled. Block selection is
+   * blocked if there are too many un-confirmed block moves.
+   */
+  private static class BytesMoved {
+    private long bytesMoved = 0L;;
+    private synchronized void inc( long bytes ) {
+      bytesMoved += bytes;
+    }
+
+    private long get() {
+      return bytesMoved;
+    }
+  };
+  private BytesMoved bytesMoved = new BytesMoved();
+  private int notChangedIterations = 0;
+  private long lastBytesMoved = 0L;
+  
+  /* The method decides blocks to move and initiates moves 
+   * Return false if no progress has been made for 3 consectutive iterations;
+   * otherwise true */
+  private boolean dispatchBlockMoves() {
+    lastBytesMoved = bytesMoved.get();
+    Source.Dispatcher dispatchers[] = new Source.Dispatcher[sources.size()];
+    int i=0;
+    for(Source source:sources) {
+      dispatchers[i] = source.new Dispatcher();
+      dispatchers[i++].start();
+    }
+    for(Source.Dispatcher dispatcher:dispatchers) {
+      try {
+        dispatcher.join();
+      } catch (InterruptedException e) {
+        LOG.info(StringUtils.stringifyException(e));
+      }
+    }
+    waitForMoveCompletion();
+    if(lastBytesMoved < bytesMoved.get()) {
+      notChangedIterations = 0;
+    } else {
+      notChangedIterations++;
+      if(notChangedIterations >= 3) {
+        return false;
+      }
+    }
+    return true;
+  }
+  
+  /* waite for all block move confirmations 
+   * by checking each target's pendingMove queue 
+   */
+  private void waitForMoveCompletion() {
+    boolean shouldWait;
+    do {
+      shouldWait = false;
+      for(BalancerDatanode target:targets) {
+        if(!target.isMoveCompleted()) {
+          shouldWait = true;
+        }
+      }
+      if(shouldWait) {
+        try {
+          Thread.sleep(60000);
+        } catch (InterruptedException ignored) {
+        }
+      }
+    } while (shouldWait);
+  }
+
+  /* mark a block to be moved */
+  private void addToMoved(BalancerBlock block) {
+    synchronized(movedBlocks) {
+      movedBlocks.put(block.getBlock(), block);
+    }
+  }
+  
+  /* check if a block is marked as moved */
+  private boolean isMoved(BalancerBlock block) {
+    synchronized(movedBlocks) {
+      return movedBlocks.containsKey(block.getBlock());
+    }
+  }
+  
+  /* Decide if it is ok to move the given block from source to target
+   * A block is a good candiate if
+   * 1. the block is not in the process of being moved/has not been moved;
+   * 2. the block does not have a replica on the target;
+   * 3. doing the move does not reduce the number of racks that the block has
+   */
+  private boolean isGoodBlockCandidate(Source source, 
+      BalancerDatanode target, BalancerBlock block) {
+    // check if the block is moved or not
+    if(isMoved(block)) {
+        return false;
+    }
+    if(block.isLocatedOnDatanode(target)) {
+      return false;
+    }
+
+    boolean goodBlock = false;
+    if(cluster.isOnSameRack(source.getDatanode(), target.getDatanode())) {
+      goodBlock = true;
+    } else {
+      boolean notOnSameRack = true;
+      synchronized (block) {
+        for(BalancerDatanode loc:block.locations) {
+          if(cluster.isOnSameRack(loc.datanode, target.datanode)) {
+            notOnSameRack = false;
+            break;
+          }
+        }
+      }
+      if(notOnSameRack) {
+        goodBlock = true;
+      } else {
+        for(BalancerDatanode loc:block.locations) {
+          if(loc != source && 
+              cluster.isOnSameRack(loc.datanode, source.datanode)) {
+            goodBlock = true;
+            break;
+          }
+        }
+      }
+    }
+    return goodBlock;
+  }
+  
+  /* reset all fields in a balancer preparing for the next iteration */
+  private void resetData() {
+    this.cluster = new NetworkTopology();
+    this.overUtilizedDatanodes.clear();
+    this.aboveAvgUtilizedDatanodes.clear();
+    this.belowAvgUtilizedDatanodes.clear();
+    this.underUtilizedDatanodes.clear();
+    this.datanodes.clear();
+    this.sources.clear();
+    this.targets.clear();  
+    this.avgUtilization = 0.0D;
+    this.movedBlocks.clear();
+    this.blockList.clear();
+  }
+  
+  private boolean isOverUtilized(BalancerDatanode datanode) {
+    return datanode.utilization > (avgUtilization+threshold);
+  }
+  
+  private boolean isAboveAvgUtilized(BalancerDatanode datanode) {
+    return (datanode.utilization <= (avgUtilization+threshold))
+        && (datanode.utilization > avgUtilization);
+  }
+  
+  private boolean isUnderUtilized(BalancerDatanode datanode) {
+    return datanode.utilization < (avgUtilization-threshold);
+  }
+
+  private boolean isBelowAvgUtilized(BalancerDatanode datanode) {
+        return (datanode.utilization >= (avgUtilization-threshold))
+                 && (datanode.utilization < avgUtilization);
+  }
+
+  /** main method of Balancer
+   * @param args arguments to a Balancer
+   * @exception any exception occurs during datanode balancing
+   */
+  public int run(String[] args) throws Exception {
+    long startTime = FSNamesystem.now();
+    try {
+      // initialize a blancer
+      init(args);
+
+      System.out.println("Iteration#\tBytes Already Moved\tBytes Left to move\tBytes Being Moved");
+      int iterations = 0;
+      while (true ) {
+        /* get all live datanodes of a cluster and their disk usage
+         * decide the number of bytes need to be moved
+         */
+        long bytesLeftToMove = initNodes();
+        if(bytesLeftToMove == 0) {
+          System.out.println("The cluster is balanced. Exiting...");
+          return 1;
+        } else {
+          LOG.info( "Need to move "+ FsShell.byteDesc(bytesLeftToMove)
+              +" bytes to make the cluster balanced." );
+        }
+        
+        /* Decide all the nodes that will participate in the block move and
+         * the number of bytes that need to be moved from one node to another
+         */
+        long bytesToMove = chooseNodes();
+        if(bytesToMove == 0) {
+          System.out.println("No node can receive any block. Exiting...");
+          return 1;
+        } else {
+          LOG.info( "Will move " + FsShell.byteDesc(bytesToMove) +
+              "bytes in this iteration");
+        }
+   
+        System.out.println(iterations + "\t" +
+            FsShell.byteDesc(bytesMoved.get()) + "\t" +
+            FsShell.byteDesc(bytesLeftToMove)+"\t" +
+            FsShell.byteDesc(bytesToMove));
+        
+        /* For each pair of <source, target>, start a thread that repeatedly 
+         * decide a block to be moved and its proxy source, 
+         * then initiates the move util all bytes are moved or no more block
+         * available to move.
+         */
+        if(!dispatchBlockMoves()) {
+          System.out.println(
+            "No progress has been made for 3 iterations. Exting...");
+          return 1;
+        }
+
+        // clean all lists
+        resetData();
+        
+        try {
+          Thread.sleep(2*conf.getLong("dfs.heartbeat.interval", 3));
+        } catch (InterruptedException ignored) {
+        }
+        
+        iterations++;
+      }
+    } catch (IOException e) {
+      return -1;
+    }finally {
+    System.out.println("Balancing took " + (FSNamesystem.now()-startTime)/1000
+        + "sec");
+    }
+  }
+
+  /** return this balancer's configuration */
+  public Configuration getConf() {
+    return conf;
+  }
+
+  /** set this balancer's configuration */
+  public void setConf(Configuration conf) {
+    this.conf = conf;
+  }
+
+}
Index: src/test/org/apache/hadoop/dfs/TestBalancer.java
===================================================================
--- src/test/org/apache/hadoop/dfs/TestBalancer.java	(revision 0)
+++ src/test/org/apache/hadoop/dfs/TestBalancer.java	(revision 0)
@@ -0,0 +1,285 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.dfs;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.dfs.FSConstants.DatanodeReportType;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import junit.framework.TestCase;
+/**
+ * This class tests if a balancer schedules tasks correctly.
+ */
+public class TestBalancer extends TestCase {
+  private static final Configuration CONF = new Configuration();
+  final private static long CAPACITY = 500L;
+  final private static String RACK0 = "/rack0";
+  final private static String RACK1 = "/rack1";
+  final private static String RACK2 = "/rack2";
+  final static private String fileName = "/tmp.txt";
+  final static private Path filePath = new Path(fileName);
+  private MiniDFSCluster cluster;
+
+  ClientProtocol client;
+
+  static final int DEFAULT_BLOCK_SIZE = 10;
+  private Balancer balancer;
+  private Random r = new Random();
+
+  static {
+    CONF.setLong("dfs.block.size", DEFAULT_BLOCK_SIZE);
+    CONF.setInt("io.bytes.per.checksum", DEFAULT_BLOCK_SIZE);
+    //CONF.setLong("dfs.blockreport.intantervalMsec", 100L);
+    CONF.setLong("dfs.heartbeat.interval", 1L);
+    CONF.setBoolean(SimulatedFSDataset.CONFIG_PROPERTY_SIMULATED, true);
+
+  }
+
+  /* create a file with a length of <code>fileLen</code> */
+  private void createFile(long fileLen, short replicationFactor)
+  throws IOException {
+    FileSystem fs = cluster.getFileSystem();
+    DFSTestUtil.createFile(fs, filePath, fileLen, 
+        replicationFactor, r.nextLong());
+    DFSTestUtil.waitReplication(fs, filePath, replicationFactor);
+  }
+
+
+  /* fill up a cluster with <code>numNodes</code> datanodes 
+   * whose used space to be <code>size</code>
+   */
+  private Block[] generateBlocks(long size, short numNodes) throws IOException {
+    cluster = new MiniDFSCluster( CONF, numNodes, true, null);
+    try {
+      cluster.waitActive();
+      client = DFSClient.createNamenode(
+          DataNode.createSocketAddr(CONF.get("fs.default.name")), CONF);
+
+      short replicationFactor = (short)(numNodes-1);
+      long fileLen = size/replicationFactor;
+      createFile(fileLen, replicationFactor);
+
+      List<LocatedBlock> locatedBlocks = cluster.getNameNode().
+      getBlockLocations(fileName, 0, fileLen).getLocatedBlocks();
+
+      int numOfBlocks = locatedBlocks.size();
+      Block[] blocks = new Block[numOfBlocks];
+      for(int i=0; i<numOfBlocks; i++) {
+        Block b = locatedBlocks.get(i).getBlock();
+        blocks[i] = new Block(b.getBlockId(), b.getNumBytes());
+      }
+
+      return blocks;
+    } finally {
+      cluster.shutdown();
+    }
+  }
+
+  /* Distribute all blocks according to the given distribution */
+  Block[][] distributeBlocks(Block[] blocks, short replicationFactor, 
+      final long[] distribution ) {
+    // make a copy
+    long[] usedSpace = new long[distribution.length];
+    System.arraycopy(distribution, 0, usedSpace, 0, distribution.length);
+
+    List<Block>[] blockReports = (List<Block>[])new List<?>[usedSpace.length];
+    Block[][] results = new Block[usedSpace.length][];
+    for(int i=0; i<usedSpace.length; i++) {
+      blockReports[i] = new ArrayList<Block>();
+    }
+    for(int i=0; i<blocks.length; i++) {
+      for(int j=0; j<replicationFactor; j++) {
+        boolean notChosen = true;
+        while(notChosen) {
+          int chosenIndex = r.nextInt(usedSpace.length);
+          if( usedSpace[chosenIndex]>0 ) {
+            notChosen = false;
+            blockReports[chosenIndex].add(blocks[i]);
+            usedSpace[chosenIndex] -= blocks[i].getNumBytes();
+          }
+        }
+      }
+    }
+    for(int i=0; i<usedSpace.length; i++) {
+      results[i] = blockReports[i].toArray(new Block[blockReports[i].size()]);
+    }
+    return results;
+  }
+
+  /* we first start a cluster and fill the cluster up to a certain size.
+   * then redistribute blocks according the requried distribution.
+   * Afterwards a balnacer is running to balance the cluster.
+   */
+  private void testUnevenDistribution(
+      long distribution[], long capacities[], String[] racks) throws Exception {
+    int numDatanodes = distribution.length;
+    if (capacities.length != numDatanodes || racks.length != numDatanodes) {
+      throw new IllegalArgumentException("Array length is not the same");
+    }
+
+    // calculate total space that need to be filled
+    long totalUsedSpace=0L;
+    for(int i=0; i<distribution.length; i++) {
+      totalUsedSpace += distribution[i];
+    }
+
+    // fill the cluster
+    Block[] blocks = generateBlocks(totalUsedSpace, (short)numDatanodes);
+
+    // redistribute blocks
+    Block[][] blocksDN = distributeBlocks(
+        blocks, (short)(numDatanodes-1), distribution);
+
+    // restart the cluster: do NOT format the cluster
+    CONF.set("dfs.safemode.threshold.pct", "0.0f"); 
+    cluster = new MiniDFSCluster(0, CONF, numDatanodes,
+        false, true, null, racks, capacities);
+    cluster.waitActive();
+    client = DFSClient.createNamenode(
+        DataNode.createSocketAddr(CONF.get("fs.default.name")), CONF);
+
+    cluster.injectBlocks(blocksDN);
+
+    long totalCapacity = 0L;
+    for(long capacity:capacities) {
+      totalCapacity += capacity;
+    }
+    runBalancer(totalUsedSpace, totalCapacity);
+  }
+
+  /* wait for one heartbeat */
+  private void waitForHeartBeat( long expectedUsedSpace, long expectedTotalSpace )
+  throws IOException {
+    long[] status = client.getStats();
+    while(status[0] != expectedTotalSpace || status[1] != expectedUsedSpace ) {
+      try {
+        Thread.sleep(100L);
+      } catch(InterruptedException ignored) {
+      }
+      status = client.getStats();
+    }
+  }
+
+  /* This test start a one-node cluster, fill the node to be 30% full;
+   * It then adds an empty node and start balancing.
+   * @param newCapacity new node's capacity
+   * @param new 
+   */
+  private void test(long[] capacities, String[] racks, 
+      long newCapacity, String newRack) throws Exception {
+    int numOfDatanodes = capacities.length;
+    assertEquals(numOfDatanodes, racks.length);
+    cluster = new MiniDFSCluster(0, CONF, capacities.length, true, true, null, 
+        racks, capacities);
+    try {
+      cluster.waitActive();
+      client = DFSClient.createNamenode(
+          DataNode.createSocketAddr(CONF.get("fs.default.name")), CONF);
+
+      long totalCapacity=0L;
+      for(long capacity:capacities) {
+        totalCapacity += capacity;
+      }
+      // fill up the cluster to be 30% full
+      long totalUsedSpace = totalCapacity*3/10;
+      createFile(totalUsedSpace/numOfDatanodes, (short)numOfDatanodes);
+      // start up an empty node with the same capacity and on the same rack
+      cluster.startDataNodes(CONF, 1, true, null,
+          new String[]{newRack}, new long[]{newCapacity});
+
+      totalCapacity += newCapacity;
+
+      // run balancer and validate results
+      runBalancer(totalUsedSpace, totalCapacity);
+    } finally {
+      cluster.shutdown();
+    }
+  }
+
+  /* Start balancer and check if the cluster is balanced after the run */
+  private void runBalancer( long totalUsedSpace, long totalCapacity )
+  throws Exception {
+    waitForHeartBeat(totalUsedSpace, totalCapacity);
+
+    // start rebalancing
+    balancer = new Balancer(CONF);
+    balancer.run(new String[0]);
+
+    waitForHeartBeat(totalUsedSpace, totalCapacity);
+    boolean balanced;
+    do {
+      DatanodeInfo[] datanodeReport = 
+        client.getDatanodeReport(DatanodeReportType.ALL);
+      assertEquals(datanodeReport.length, cluster.getDataNodes().size());
+      balanced = true;
+      double avgUtilization = ((double)totalUsedSpace)/totalCapacity*100;
+      for(DatanodeInfo datanode:datanodeReport) {
+        if(Math.abs(avgUtilization-
+            ((double)datanode.getDfsUsed())/datanode.getCapacity()*100)>10) {
+          balanced = false;
+          try {
+            Thread.sleep(100);
+          } catch(InterruptedException ignored) {
+          }
+          break;
+        }
+      }
+    } while(!balanced);
+
+  }
+  /** Test a cluster with even distribution, 
+   * then a new empty node is added to the cluster*/
+  public void testBalancer0() throws Exception {
+    /** one-node cluste test*/
+    // add an empty node with the same CAPACITY & rack
+    //test(new long[]{CAPACITY}, new String[]{RACK0}, CAPACITY, RACK0);
+    // add an empty node with the same CAPACITY but different rack
+    //test(new long[]{CAPACITY}, new String[]{RACK0}, CAPACITY, RACK1);
+    // add an empty node with half of the CAPACITY & the same rack
+    //test(new long[]{CAPACITY}, new String[]{RACK0}, CAPACITY/2, RACK0);
+
+    /** two-node cluster test */
+    test(new long[]{CAPACITY, CAPACITY}, new String[]{RACK0, RACK0},
+        CAPACITY, RACK0);
+    test(new long[]{CAPACITY, CAPACITY}, new String[]{RACK0, RACK1},
+        CAPACITY, RACK2);
+  }
+
+  /** Test unEven distributed cluster */
+  public void testBalancer1() throws Exception {
+    testUnevenDistribution(
+        new long[] {50*CAPACITY/100, 10*CAPACITY/100},
+        new long[]{CAPACITY, CAPACITY},
+        new String[] {RACK0, RACK1});
+  }
+
+  /**
+   * @param args
+   */
+  public static void main(String[] args) throws Exception {
+    (new TestBalancer()).testBalancer0();
+
+  }
+
+}