HBASE-15515 Improve LocalityBasedCandidateGenerator in Balancer (Guanghao Zhang)

hbase-server/src/main/java/org/apache/hadoop/hbase/master/balancer/BaseLoadBalancer.java

@@ -826,7 +826,7 @@ public abstract class BaseLoadBalancer implements LoadBalancer {
     int getLowestLocalityRegionOnServer(int serverIndex) {
       if (regionFinder != null) {
         float lowestLocality = 1.0f;
-        int lowestLocalityRegionIndex = 0;
+        int lowestLocalityRegionIndex = -1;
         if (regionsPerServer[serverIndex].length == 0) {
           // No regions on that region server
           return -1;
@@ -836,13 +836,22 @@ public abstract class BaseLoadBalancer implements LoadBalancer {
           HDFSBlocksDistribution distribution = regionFinder
               .getBlockDistribution(regions[regionIndex]);
           float locality = distribution.getBlockLocalityIndex(servers[serverIndex].getHostname());
+          // skip empty region
+          if (distribution.getUniqueBlocksTotalWeight() == 0) {
+            continue;
+          }
           if (locality < lowestLocality) {
             lowestLocality = locality;
             lowestLocalityRegionIndex = j;
           }
         }
+        if (lowestLocalityRegionIndex == -1) {
+          return -1;
+        }
         if (LOG.isTraceEnabled()) {
-          LOG.trace(" Lowest locality region index is " + lowestLocalityRegionIndex
+          LOG.trace("Lowest locality region is "
+              + regions[regionsPerServer[serverIndex][lowestLocalityRegionIndex]]
+                  .getRegionNameAsString() + " with locality " + lowestLocality
               + " and its region server contains " + regionsPerServer[serverIndex].length
               + " regions");
         }
@@ -861,9 +870,14 @@ public abstract class BaseLoadBalancer implements LoadBalancer {
       }
     }
 
-    int getLeastLoadedTopServerForRegion(int region) {
+    /**
+     * Returns a least loaded server which has better locality for this region
+     * than the current server.
+     */
+    int getLeastLoadedTopServerForRegion(int region, int currentServer) {
       if (regionFinder != null) {
-        List<ServerName> topLocalServers = regionFinder.getTopBlockLocations(regions[region]);
+        List<ServerName> topLocalServers = regionFinder.getTopBlockLocations(regions[region],
+          servers[currentServer].getHostname());
         int leastLoadedServerIndex = -1;
         int load = Integer.MAX_VALUE;
         for (ServerName sn : topLocalServers) {
@@ -880,6 +894,10 @@ public abstract class BaseLoadBalancer implements LoadBalancer {
             load = tempLoad;
           }
         }
+        if (leastLoadedServerIndex != -1) {
+          LOG.debug("Pick the least loaded server " + servers[leastLoadedServerIndex].getHostname()
+            + " with better locality for region " + regions[region]);
+        }
         return leastLoadedServerIndex;
       } else {
         return -1;

hbase-server/src/main/java/org/apache/hadoop/hbase/master/balancer/RegionLocationFinder.java

@@ -170,6 +170,22 @@ class RegionLocationFinder {
     return mapHostNameToServerName(topHosts);
   }
 
+  /**
+   * Returns an ordered list of hosts which have better locality for this region
+   * than the current host.
+   */
+  protected List<ServerName> getTopBlockLocations(HRegionInfo region, String currentHost) {
+    HDFSBlocksDistribution blocksDistribution = getBlockDistribution(region);
+    List<String> topHosts = new ArrayList<String>();
+    for (String host : blocksDistribution.getTopHosts()) {
+      if (host.equals(currentHost)) {
+        break;
+      }
+      topHosts.add(host);
+    }
+    return mapHostNameToServerName(topHosts);
+  }
+
   /**
    * Returns an ordered list of hosts that are hosting the blocks for this
    * region. The weight of each host is the sum of the block lengths of all

hbase-server/src/main/java/org/apache/hadoop/hbase/master/balancer/StochasticLoadBalancer.java

@@ -705,9 +705,7 @@ public class StochasticLoadBalancer extends BaseLoadBalancer {
         return pickRandomRegions(cluster, thisServer, otherServer);
       }
 
-      cluster.calculateRegionServerLocalities();
+      int thisServer = pickRandomServer(cluster);
-      // Pick server with lowest locality
-      int thisServer = pickLowestLocalityServer(cluster);
       int thisRegion;
       if (thisServer == -1) {
         LOG.warn("Could not pick lowest locality region server");
@@ -722,7 +720,7 @@ public class StochasticLoadBalancer extends BaseLoadBalancer {
       }
 
       // Pick the least loaded server with good locality for the region
-      int otherServer = cluster.getLeastLoadedTopServerForRegion(thisRegion);
+      int otherServer = cluster.getLeastLoadedTopServerForRegion(thisRegion, thisServer);
 
       if (otherServer == -1) {
         return Cluster.NullAction;