@Override
  protected List<RegionPlan> balanceTable(TableName tableName,
    Map<ServerName, List<RegionInfo>> loadOfOneTable) {
    long startTime = EnvironmentEdgeManager.currentTime();

    // construct a Cluster object with clusterMap and rest of the
    // argument as defaults
    BalancerClusterState c =
      new BalancerClusterState(loadOfOneTable, null, this.regionFinder, this.rackManager);
    if (!needsBalance(c) && !this.overallNeedsBalance()) {
      return null;
    }
    ClusterLoadState cs = new ClusterLoadState(loadOfOneTable);
    int numServers = cs.getNumServers();
    NavigableMap<ServerAndLoad, List<RegionInfo>> serversByLoad = cs.getServersByLoad();
    int numRegions = cs.getNumRegions();
    float average = cs.getLoadAverage();
    int max = (int) Math.ceil(average);
    int min = (int) average;

    // Using to check balance result.
    StringBuilder strBalanceParam = new StringBuilder();
    strBalanceParam.append("Balance parameter: numRegions=").append(numRegions)
      .append(", numServers=").append(numServers).append(", max=").append(max).append(", min=")
      .append(min);
    LOG.debug(strBalanceParam.toString());

    // Balance the cluster
    // TODO: Look at data block locality or a more complex load to do this
    MinMaxPriorityQueue<RegionPlan> regionsToMove =
      MinMaxPriorityQueue.orderedBy(rpComparator).create();
    List<RegionPlan> regionsToReturn = new ArrayList<>();

    // Walk down most loaded, pruning each to the max
    int serversOverloaded = 0;
    // flag used to fetch regions from head and tail of list, alternately
    boolean fetchFromTail = false;
    Map<ServerName, BalanceInfo> serverBalanceInfo = new TreeMap<>();
    for (Map.Entry<ServerAndLoad, List<RegionInfo>> server : serversByLoad.descendingMap()
      .entrySet()) {
      ServerAndLoad sal = server.getKey();
      int load = sal.getLoad();
      if (load <= max) {
        serverBalanceInfo.put(sal.getServerName(), new BalanceInfo(0, 0, server.getValue()));
        continue;
      }
      serversOverloaded++;
      List<RegionInfo> regions = server.getValue();
      int numToOffload = Math.min(load - max, regions.size());
      // account for the out-of-band regions which were assigned to this server
      // after some other region server crashed
      Collections.sort(regions, riComparator);
      int numTaken = 0;
      for (int i = 0; i <= numToOffload;) {
        RegionInfo hri = regions.get(i); // fetch from head
        if (fetchFromTail) {
          hri = regions.get(regions.size() - 1 - i);
        }
        i++;
        regionsToMove.add(new RegionPlan(hri, sal.getServerName(), null));
        numTaken++;
        if (numTaken >= numToOffload) {
          break;
        }
      }
      serverBalanceInfo.put(sal.getServerName(),
        new BalanceInfo(numToOffload, -numTaken, server.getValue()));
    }
    int totalNumMoved = regionsToMove.size();

    // Walk down least loaded, filling each to the min
    int neededRegions = 0; // number of regions needed to bring all up to min
    fetchFromTail = false;

    Map<ServerName, Integer> underloadedServers = new HashMap<>();
    int maxToTake = numRegions - min;
    for (Map.Entry<ServerAndLoad, List<RegionInfo>> server : serversByLoad.entrySet()) {
      if (maxToTake == 0) {
        break; // no more to take
      }
      int load = server.getKey().getLoad();
      if (load >= min) {
        continue; // look for other servers which haven't reached min
      }
      int regionsToPut = min - load;
      maxToTake -= regionsToPut;
      underloadedServers.put(server.getKey().getServerName(), regionsToPut);
    }
    // number of servers that get new regions
    int serversUnderloaded = underloadedServers.size();
    int incr = 1;
    List<ServerName> sns =
      Arrays.asList(underloadedServers.keySet().toArray(new ServerName[serversUnderloaded]));
    Collections.shuffle(sns);
    while (regionsToMove.size() > 0) {
      int cnt = 0;
      int i = incr > 0 ? 0 : underloadedServers.size() - 1;
      for (; i >= 0 && i < underloadedServers.size(); i += incr) {
        if (regionsToMove.isEmpty()) {
          break;
        }
        ServerName si = sns.get(i);
        int numToTake = underloadedServers.get(si);
        if (numToTake == 0) {
          continue;
        }

        addRegionPlan(regionsToMove, fetchFromTail, si, regionsToReturn);

        underloadedServers.put(si, numToTake - 1);
        cnt++;
        BalanceInfo bi = serverBalanceInfo.get(si);
        bi.setNumRegionsAdded(bi.getNumRegionsAdded() + 1);
      }
      if (cnt == 0) {
        break;
      }
      // iterates underloadedServers in the other direction
      incr = -incr;
    }
    for (Integer i : underloadedServers.values()) {
      // If we still want to take some, increment needed
      neededRegions += i;
    }

    // Need to do a second pass.
    // Either more regions to assign out or servers that are still underloaded

    // If we need more to fill min, grab one from each most loaded until enough
    if (neededRegions != 0) {
      // Walk down most loaded, grabbing one from each until we get enough
      for (Map.Entry<ServerAndLoad, List<RegionInfo>> server : serversByLoad.descendingMap()
        .entrySet()) {
        BalanceInfo balanceInfo = serverBalanceInfo.get(server.getKey().getServerName());
        int idx = balanceInfo == null ? 0 : balanceInfo.getNextRegionForUnload();
        if (idx >= server.getValue().size()) {
          break;
        }
        RegionInfo region = server.getValue().get(idx);
        if (region.isMetaRegion()) {
          continue; // Don't move meta regions.
        }
        regionsToMove.add(new RegionPlan(region, server.getKey().getServerName(), null));
        balanceInfo.setNumRegionsAdded(balanceInfo.getNumRegionsAdded() - 1);
        balanceInfo.setNextRegionForUnload(balanceInfo.getNextRegionForUnload() + 1);
        totalNumMoved++;
        if (--neededRegions == 0) {
          // No more regions needed, done shedding
          break;
        }
      }
    }

    // Now we have a set of regions that must be all assigned out
    // Assign each underloaded up to the min, then if leftovers, assign to max

    // Walk down least loaded, assigning to each to fill up to min
    for (Map.Entry<ServerAndLoad, List<RegionInfo>> server : serversByLoad.entrySet()) {
      int regionCount = server.getKey().getLoad();
      if (regionCount >= min) {
        break;
      }
      BalanceInfo balanceInfo = serverBalanceInfo.get(server.getKey().getServerName());
      if (balanceInfo != null) {
        regionCount += balanceInfo.getNumRegionsAdded();
      }
      if (regionCount >= min) {
        continue;
      }
      int numToTake = min - regionCount;
      int numTaken = 0;
      while (numTaken < numToTake && 0 < regionsToMove.size()) {
        addRegionPlan(regionsToMove, fetchFromTail, server.getKey().getServerName(),
          regionsToReturn);
        numTaken++;
        balanceInfo.setNumRegionsAdded(balanceInfo.getNumRegionsAdded() + 1);
      }
    }

    if (min != max) {
      balanceOverall(regionsToReturn, serverBalanceInfo, fetchFromTail, regionsToMove, max, min);
    }

    long endTime = EnvironmentEdgeManager.currentTime();

    if (!regionsToMove.isEmpty() || neededRegions != 0) {
      // Emit data so can diagnose how balancer went astray.
      LOG.warn(
        "regionsToMove=" + totalNumMoved + ", numServers=" + numServers + ", serversOverloaded="
          + serversOverloaded + ", serversUnderloaded=" + serversUnderloaded);
      StringBuilder sb = new StringBuilder();
      for (Map.Entry<ServerName, List<RegionInfo>> e : loadOfOneTable.entrySet()) {
        if (sb.length() > 0) {
          sb.append(", ");
        }
        sb.append(e.getKey().toString());
        sb.append(" ");
        sb.append(e.getValue().size());
      }
      LOG.warn("Input " + sb.toString());
    }

    // All done!
    LOG.info("Done. Calculated a load balance in " + (endTime - startTime) + "ms. " + "Moving "
      + totalNumMoved + " regions off of " + serversOverloaded + " overloaded servers onto "
      + serversUnderloaded + " less loaded servers");

    return regionsToReturn;
  }
Generate a global load balancing plan according to the specified map of server information to
the most loaded regions of each server. The load balancing invariant is that all servers are
within 1 region of the average number of regions per server. If the average is an integer
number, all servers will be balanced to the average. Otherwise, all servers will have either
floor(average) or ceiling(average) regions. HBASE-3609 Modeled regionsToMove using Guava's
MinMaxPriorityQueue so that we can fetch from both ends of the queue. At the beginning, we
check whether there was empty region server just discovered by Master. If so, we alternately
choose new / old regions from head / tail of regionsToMove, respectively. This alternation
avoids clustering young regions on the newly discovered region server. Otherwise, we choose new
regions from head of regionsToMove. Another improvement from HBASE-3609 is that we assign
regions from regionsToMove to underloaded servers in round-robin fashion. Previously one
underloaded server would be filled before we move onto the next underloaded server, leading to
clustering of young regions. Finally, we randomly shuffle underloaded servers so that they
receive offloaded regions relatively evenly across calls to balanceCluster(). The algorithm is
currently implemented as such:
<ol>
<li>Determine the two valid numbers of regions each server should have,
<b>MIN</b>=floor(average) and <b>MAX</b>=ceiling(average).
<li>Iterate down the most loaded servers, shedding regions from each so each server hosts
exactly <b>MAX</b> regions. Stop once you reach a server that already has &lt;= <b>MAX</b>
regions.
<p>
Order the regions to move from most recent to least.
<li>Iterate down the least loaded servers, assigning regions so each server has exactly
<b>MIN</b> regions. Stop once you reach a server that already has &gt;= <b>MIN</b> regions.
Regions being assigned to underloaded servers are those that were shed in the previous step. It
is possible that there were not enough regions shed to fill each underloaded server to
<b>MIN</b>. If so we end up with a number of regions required to do so, <b>neededRegions</b>.
It is also possible that we were able to fill each underloaded but ended up with regions that
were unassigned from overloaded servers but that still do not have assignment. If neither of
these conditions hold (no regions needed to fill the underloaded servers, no regions leftover
from overloaded servers), we are done and return. Otherwise we handle these cases below.
<li>If <b>neededRegions</b> is non-zero (still have underloaded servers), we iterate the most
loaded servers again, shedding a single server from each (this brings them from having
<b>MAX</b> regions to having <b>MIN</b> regions).
<li>We now definitely have more regions that need assignment, either from the previous step or
from the original shedding from overloaded servers. Iterate the least loaded servers filling
each to <b>MIN</b>.
<li>If we still have more regions that need assignment, again iterate the least loaded servers,
this time giving each one (filling them to <b>MAX</b>) until we run out.
<li>All servers will now either host <b>MIN</b> or <b>MAX</b> regions. In addition, any server
hosting &gt;= <b>MAX</b> regions is guaranteed to end up with <b>MAX</b> regions at the end of
the balancing. This ensures the minimal number of regions possible are moved.
</ol>
TODO: We can at-most reassign the number of regions away from a particular server to be how
many they report as most loaded. Should we just keep all assignment in memory? Any objections?
Does this mean we need HeapSize on HMaster? Or just careful monitor? (current thinking is we
will hold all assignments in memory)
@param loadOfOneTable Map of regionservers and their load/region information to a list of their
                      most loaded regions
@return a list of regions to be moved, including source and destination, or null if cluster is
        already balanced
hbase-balancer/src/main/java/org/apache/hadoop/hbase/master/balancer/SimpleLoadBalancer.java