HBASE-29168 Add configurable throttling of region moves in CacheAwareLoadBalancer. (#6763)

wchevreuil · web-flow · commit d98c63a2eef8 · 2025-03-27T14:21:41.000Z
Signed-off-by: Tak Lon (Stephen) Wu &lt;taklwu@apache.org&gt;
diff --git a/hbase-balancer/src/main/java/org/apache/hadoop/hbase/master/LoadBalancer.java b/hbase-balancer/src/main/java/org/apache/hadoop/hbase/master/LoadBalancer.java
@@ -19,6 +19,7 @@
 
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.io.IOException;
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import org.apache.hadoop.conf.Configuration;
@@ -76,6 +77,18 @@ public interface LoadBalancer extends Stoppable, ConfigurationObserver {
   @Deprecated
   String HBASE_RSGROUP_LOADBALANCER_CLASS = "hbase.rsgroup.grouploadbalancer.class";
 
+  /**
+   * Configuration to determine the time to sleep when throttling (if throttling is implemented by
+   * the underlying implementation).
+   */
+  String MOVE_THROTTLING = "hbase.master.balancer.move.throttlingMillis";
+
+  /**
+   * The default value, in milliseconds, for the hbase.master.balancer.move.throttlingMillis if
+   * throttling is implemented.
+   */
+  Duration MOVE_THROTTLING_DEFAULT = Duration.ofMillis(60 * 1000);
+
   /**
    * Set the current cluster status. This allows a LoadBalancer to map host name to a server
    */
@@ -154,4 +167,8 @@ Map<ServerName, List<RegionInfo>> retainAssignment(Map<RegionInfo, ServerName> r
   default void
     updateBalancerLoadInfo(Map<TableName, Map<ServerName, List<RegionInfo>>> loadOfAllTable) {
   }
+
+  default void throttle(RegionPlan plan) throws Exception {
+    // noop
+  }
 }
diff --git a/hbase-balancer/src/main/java/org/apache/hadoop/hbase/master/balancer/CacheAwareLoadBalancer.java b/hbase-balancer/src/main/java/org/apache/hadoop/hbase/master/balancer/CacheAwareLoadBalancer.java
@@ -43,7 +43,9 @@
 import org.apache.hadoop.hbase.ServerMetrics;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.Size;
+import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.master.RegionPlan;
 import org.apache.hadoop.hbase.util.Pair;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
@@ -53,6 +55,13 @@
 public class CacheAwareLoadBalancer extends StochasticLoadBalancer {
   private static final Logger LOG = LoggerFactory.getLogger(CacheAwareLoadBalancer.class);
 
+  public static final String CACHE_RATIO_THRESHOLD =
+    "hbase.master.balancer.stochastic.throttling.cacheRatio";
+  public static final float CACHE_RATIO_THRESHOLD_DEFAULT = 0.8f;
+
+  public Float ratioThreshold;
+
+  private Long sleepTime;
   private Configuration configuration;
 
   public enum GeneratorFunctionType {
@@ -65,6 +74,9 @@ public synchronized void loadConf(Configuration configuration) {
     this.configuration = configuration;
     this.costFunctions = new ArrayList<>();
     super.loadConf(configuration);
+    ratioThreshold =
+      this.configuration.getFloat(CACHE_RATIO_THRESHOLD, CACHE_RATIO_THRESHOLD_DEFAULT);
+    sleepTime = configuration.getLong(MOVE_THROTTLING, MOVE_THROTTLING_DEFAULT.toMillis());
   }
 
   @Override
@@ -160,6 +172,53 @@ private RegionInfo getRegionInfoByEncodedName(BalancerClusterState cluster, Stri
     return null;
   }
 
+  @Override
+  public void throttle(RegionPlan plan) {
+    Pair<ServerName, Float> rsRatio = this.regionCacheRatioOnOldServerMap.get(plan.getRegionName());
+    if (
+      rsRatio != null && plan.getDestination().equals(rsRatio.getFirst())
+        && rsRatio.getSecond() >= ratioThreshold
+    ) {
+      LOG.debug("Moving region {} to server {} with cache ratio {}. No throttling needed.",
+        plan.getRegionInfo().getEncodedName(), plan.getDestination(), rsRatio.getSecond());
+    } else {
+      if (rsRatio != null) {
+        LOG.debug("Moving region {} to server {} with cache ratio: {}. Throttling move for {}ms.",
+          plan.getRegionInfo().getEncodedName(), plan.getDestination(),
+          plan.getDestination().equals(rsRatio.getFirst()) ? rsRatio.getSecond() : "unknown",
+          sleepTime);
+      } else {
+        LOG.debug(
+          "Moving region {} to server {} with no cache ratio info for the region. "
+            + "Throttling move for {}ms.",
+          plan.getRegionInfo().getEncodedName(), plan.getDestination(), sleepTime);
+      }
+      try {
+        Thread.sleep(sleepTime);
+      } catch (InterruptedException e) {
+        throw new RuntimeException(e);
+      }
+    }
+  }
+
+  @Override
+  protected List<RegionPlan> balanceTable(TableName tableName,
+    Map<ServerName, List<RegionInfo>> loadOfOneTable) {
+    final Map<String, Pair<ServerName, Float>> snapshot = new HashMap<>();
+    snapshot.putAll(this.regionCacheRatioOnOldServerMap);
+    List<RegionPlan> plans = super.balanceTable(tableName, loadOfOneTable);
+    plans.sort((p1, p2) -> {
+      Pair<ServerName, Float> pair1 = snapshot.get(p1.getRegionName());
+      Float ratio1 =
+        pair1 == null ? 0 : pair1.getFirst().equals(p1.getDestination()) ? pair1.getSecond() : 0f;
+      Pair<ServerName, Float> pair2 = snapshot.get(p2.getRegionName());
+      Float ratio2 =
+        pair2 == null ? 0 : pair2.getFirst().equals(p2.getDestination()) ? pair2.getSecond() : 0f;
+      return ratio1.compareTo(ratio2) * (-1);
+    });
+    return plans;
+  }
+
   private class CacheAwareCandidateGenerator extends CandidateGenerator {
     @Override
     protected BalanceAction generate(BalancerClusterState cluster) {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
@@ -621,6 +621,7 @@ protected void cacheBlockWithWaitInternal(BlockCacheKey cacheKey, Cacheable cach
       try {
         successfulAddition = bq.offer(re, queueAdditionWaitTime, TimeUnit.MILLISECONDS);
       } catch (InterruptedException e) {
+        LOG.error("Thread interrupted: ", e);
         Thread.currentThread().interrupt();
       }
     } else {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java
@@ -2188,10 +2188,14 @@ public List<RegionPlan> executeRegionPlansWithThrottling(List<RegionPlan> plans)
         // TODO: bulk assign
         try {
           this.assignmentManager.balance(plan);
+          this.balancer.updateClusterMetrics(getClusterMetricsWithoutCoprocessor());
+          this.balancer.throttle(plan);
         } catch (HBaseIOException hioe) {
           // should ignore failed plans here, avoiding the whole balance plans be aborted
           // later calls of balance() can fetch up the failed and skipped plans
           LOG.warn("Failed balance plan {}, skipping...", plan, hioe);
+        } catch (Exception e) {
+          LOG.warn("Failed throttling assigning a new plan.", e);
         }
         // rpCount records balance plans processed, does not care if a plan succeeds
         rpCount++;
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestBlockEvictionOnRegionMovement.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestBlockEvictionOnRegionMovement.java
@@ -18,6 +18,7 @@
 package org.apache.hadoop.hbase.io.hfile;
 
 import static org.apache.hadoop.hbase.HConstants.BUCKET_CACHE_IOENGINE_KEY;
+import static org.apache.hadoop.hbase.io.hfile.CacheConfig.CACHE_BLOCKS_ON_WRITE_KEY;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertTrue;
@@ -82,6 +83,7 @@ public void setup() throws Exception {
     conf.set("hbase.bucketcache.persistent.path", testDir + "/bucket.persistence");
     conf.setLong(CacheConfig.BUCKETCACHE_PERSIST_INTERVAL_KEY, 100);
     conf.setBoolean(CacheConfig.EVICT_BLOCKS_ON_CLOSE_KEY, true);
+    conf.setBoolean(CACHE_BLOCKS_ON_WRITE_KEY, true);
     zkCluster = TEST_UTIL.startMiniZKCluster();
     cluster = TEST_UTIL.startMiniHBaseCluster(option);
     cluster.setConf(conf);
@@ -99,7 +101,7 @@ public void testBlockEvictionOnRegionMove() throws Exception {
     assertTrue(regionServingRS.getBlockCache().isPresent());
     long oldUsedCacheSize =
       regionServingRS.getBlockCache().get().getBlockCaches()[1].getCurrentSize();
-    assertNotEquals(0, regionServingRS.getBlockCache().get().getBlockCaches()[1].getBlockCount());
+    assertNotEquals(0, oldUsedCacheSize);
 
     Admin admin = TEST_UTIL.getAdmin();
     RegionInfo regionToMove = regionServingRS.getRegions(tableRegionMove).get(0).getRegionInfo();
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCache.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCache.java
@@ -19,6 +19,7 @@
 
 import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.ACCEPT_FACTOR_CONFIG_NAME;
 import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.BLOCK_ORPHAN_GRACE_PERIOD;
+import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.DEFAULT_ERROR_TOLERATION_DURATION;
 import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.EXTRA_FREE_FACTOR_CONFIG_NAME;
 import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.MIN_FACTOR_CONFIG_NAME;
 import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.QUEUE_ADDITION_WAIT_TIME;
@@ -870,7 +871,7 @@ public void testBlockAdditionWaitWhenCache() throws Exception {
       config.setLong(QUEUE_ADDITION_WAIT_TIME, 1000);
 
       bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
-        constructedBlockSizes, 1, 1, persistencePath);
+        constructedBlockSizes, 1, 1, persistencePath, DEFAULT_ERROR_TOLERATION_DURATION, config);
       assertTrue(bucketCache.waitForCacheInitialization(10000));
       long usedByteSize = bucketCache.getAllocator().getUsedSize();
       assertEquals(0, usedByteSize);
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/balancer/TestCacheAwareLoadBalancer.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/balancer/TestCacheAwareLoadBalancer.java
@@ -18,6 +18,7 @@
 package org.apache.hadoop.hbase.master.balancer;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -47,6 +48,8 @@
 import org.apache.hadoop.hbase.master.RegionPlan;
 import org.apache.hadoop.hbase.testclassification.LargeTests;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.Pair;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.Test;
@@ -234,6 +237,157 @@ public void testRegionsPartiallyCachedOnOldServerAndNotCachedOnCurrentServer() t
     assertTrue(targetServers.get(server1).containsAll(oldCachedRegions));
   }
 
+  @Test
+  public void testThrottlingRegionBeyondThreshold() throws Exception {
+    Configuration conf = HBaseConfiguration.create();
+    CacheAwareLoadBalancer balancer = new CacheAwareLoadBalancer();
+    balancer.setClusterInfoProvider(new DummyClusterInfoProvider(conf));
+    balancer.loadConf(conf);
+    balancer.initialize();
+    ServerName server0 = servers.get(0);
+    ServerName server1 = servers.get(1);
+    Pair<ServerName, Float> regionRatio = new Pair<>();
+    regionRatio.setFirst(server0);
+    regionRatio.setSecond(1.0f);
+    balancer.regionCacheRatioOnOldServerMap.put("region1", regionRatio);
+    RegionInfo mockedInfo = mock(RegionInfo.class);
+    when(mockedInfo.getEncodedName()).thenReturn("region1");
+    RegionPlan plan = new RegionPlan(mockedInfo, server1, server0);
+    long startTime = EnvironmentEdgeManager.currentTime();
+    balancer.throttle(plan);
+    long endTime = EnvironmentEdgeManager.currentTime();
+    assertTrue((endTime - startTime) < 10);
+  }
+
+  @Test
+  public void testThrottlingRegionBelowThreshold() throws Exception {
+    Configuration conf = HBaseConfiguration.create();
+    conf.setLong(CacheAwareLoadBalancer.MOVE_THROTTLING, 100);
+    CacheAwareLoadBalancer balancer = new CacheAwareLoadBalancer();
+    balancer.setClusterInfoProvider(new DummyClusterInfoProvider(conf));
+    balancer.loadConf(conf);
+    balancer.initialize();
+    ServerName server0 = servers.get(0);
+    ServerName server1 = servers.get(1);
+    Pair<ServerName, Float> regionRatio = new Pair<>();
+    regionRatio.setFirst(server0);
+    regionRatio.setSecond(0.1f);
+    balancer.regionCacheRatioOnOldServerMap.put("region1", regionRatio);
+    RegionInfo mockedInfo = mock(RegionInfo.class);
+    when(mockedInfo.getEncodedName()).thenReturn("region1");
+    RegionPlan plan = new RegionPlan(mockedInfo, server1, server0);
+    long startTime = EnvironmentEdgeManager.currentTime();
+    balancer.throttle(plan);
+    long endTime = EnvironmentEdgeManager.currentTime();
+    assertTrue((endTime - startTime) >= 100);
+  }
+
+  @Test
+  public void testThrottlingCacheRatioUnknownOnTarget() throws Exception {
+    Configuration conf = HBaseConfiguration.create();
+    conf.setLong(CacheAwareLoadBalancer.MOVE_THROTTLING, 100);
+    CacheAwareLoadBalancer balancer = new CacheAwareLoadBalancer();
+    balancer.setClusterInfoProvider(new DummyClusterInfoProvider(conf));
+    balancer.loadConf(conf);
+    balancer.initialize();
+    ServerName server0 = servers.get(0);
+    ServerName server1 = servers.get(1);
+    ServerName server3 = servers.get(2);
+    // setting region cache ratio 100% on server 3, though this is not the target in the region plan
+    Pair<ServerName, Float> regionRatio = new Pair<>();
+    regionRatio.setFirst(server3);
+    regionRatio.setSecond(1.0f);
+    balancer.regionCacheRatioOnOldServerMap.put("region1", regionRatio);
+    RegionInfo mockedInfo = mock(RegionInfo.class);
+    when(mockedInfo.getEncodedName()).thenReturn("region1");
+    RegionPlan plan = new RegionPlan(mockedInfo, server1, server0);
+    long startTime = EnvironmentEdgeManager.currentTime();
+    balancer.throttle(plan);
+    long endTime = EnvironmentEdgeManager.currentTime();
+    assertTrue((endTime - startTime) >= 100);
+  }
+
+  @Test
+  public void testThrottlingCacheRatioUnknownForRegion() throws Exception {
+    Configuration conf = HBaseConfiguration.create();
+    conf.setLong(CacheAwareLoadBalancer.MOVE_THROTTLING, 100);
+    CacheAwareLoadBalancer balancer = new CacheAwareLoadBalancer();
+    balancer.setClusterInfoProvider(new DummyClusterInfoProvider(conf));
+    balancer.loadConf(conf);
+    balancer.initialize();
+    ServerName server0 = servers.get(0);
+    ServerName server1 = servers.get(1);
+    ServerName server3 = servers.get(2);
+    // No cache ratio available for region1
+    RegionInfo mockedInfo = mock(RegionInfo.class);
+    when(mockedInfo.getEncodedName()).thenReturn("region1");
+    RegionPlan plan = new RegionPlan(mockedInfo, server1, server0);
+    long startTime = EnvironmentEdgeManager.currentTime();
+    balancer.throttle(plan);
+    long endTime = EnvironmentEdgeManager.currentTime();
+    assertTrue((endTime - startTime) >= 100);
+  }
+
+  @Test
+  public void testRegionPlansSortedByCacheRatioOnTarget() throws Exception {
+    // The regions are fully cached on old server
+
+    Map<ServerName, List<RegionInfo>> clusterState = new HashMap<>();
+    ServerName server0 = servers.get(0);
+    ServerName server1 = servers.get(1);
+    ServerName server2 = servers.get(2);
+
+    // Simulate on RS with all regions, and two RSes with no regions
+    List<RegionInfo> regionsOnServer0 = randomRegions(15);
+    List<RegionInfo> regionsOnServer1 = randomRegions(0);
+    List<RegionInfo> regionsOnServer2 = randomRegions(0);
+
+    clusterState.put(server0, regionsOnServer0);
+    clusterState.put(server1, regionsOnServer1);
+    clusterState.put(server2, regionsOnServer2);
+
+    // Mock cluster metrics
+    // Mock 5 regions from server0 were previously hosted on server1
+    List<RegionInfo> oldCachedRegions1 = regionsOnServer0.subList(5, 10);
+    List<RegionInfo> oldCachedRegions2 = regionsOnServer0.subList(10, regionsOnServer0.size());
+    Map<ServerName, ServerMetrics> serverMetricsMap = new TreeMap<>();
+    // mock server metrics to set cache ratio as 0 in the RS 0
+    serverMetricsMap.put(server0, mockServerMetricsWithRegionCacheInfo(server0, regionsOnServer0,
+      0.0f, new ArrayList<>(), 0, 10));
+    // mock server metrics to set cache ratio as 1 in the RS 1
+    serverMetricsMap.put(server1, mockServerMetricsWithRegionCacheInfo(server1, regionsOnServer1,
+      0.0f, oldCachedRegions1, 10, 10));
+    // mock server metrics to set cache ratio as .8 in the RS 2
+    serverMetricsMap.put(server2, mockServerMetricsWithRegionCacheInfo(server2, regionsOnServer2,
+      0.0f, oldCachedRegions2, 8, 10));
+    ClusterMetrics clusterMetrics = mock(ClusterMetrics.class);
+    when(clusterMetrics.getLiveServerMetrics()).thenReturn(serverMetricsMap);
+    loadBalancer.updateClusterMetrics(clusterMetrics);
+
+    Map<TableName, Map<ServerName, List<RegionInfo>>> LoadOfAllTable =
+      (Map) mockClusterServersWithTables(clusterState);
+    List<RegionPlan> plans = loadBalancer.balanceCluster(LoadOfAllTable);
+    LOG.debug("plans size: {}", plans.size());
+    LOG.debug("plans: {}", plans);
+    LOG.debug("server1 name: {}", server1.getServerName());
+    // assert the plans are in descending order from the most cached to the least cached
+    int highCacheCount = 0;
+    for (RegionPlan plan : plans) {
+      LOG.debug("plan region: {}, target server: {}", plan.getRegionInfo().getEncodedName(),
+        plan.getDestination().getServerName());
+      if (highCacheCount < 5) {
+        LOG.debug("Count: {}", highCacheCount);
+        assertTrue(oldCachedRegions1.contains(plan.getRegionInfo()));
+        assertFalse(oldCachedRegions2.contains(plan.getRegionInfo()));
+        highCacheCount++;
+      } else {
+        assertTrue(oldCachedRegions2.contains(plan.getRegionInfo()));
+        assertFalse(oldCachedRegions1.contains(plan.getRegionInfo()));
+      }
+    }
+
+  }
+
   @Test
   public void testRegionsFullyCachedOnOldServerAndNotCachedOnCurrentServers() throws Exception {
     // The regions are fully cached on old server

Original file line number	Diff line number	Diff line change
`@@ -621,6 +621,7 @@ protected void cacheBlockWithWaitInternal(BlockCacheKey cacheKey, Cacheable cach`
`621`	`621`	`try {`
`622`	`622`	`successfulAddition = bq.offer(re, queueAdditionWaitTime, TimeUnit.MILLISECONDS);`
`623`	`623`	`} catch (InterruptedException e) {`
	`624`	`+ LOG.error("Thread interrupted: ", e);`
`624`	`625`	`Thread.currentThread().interrupt();`
`625`	`626`	`}`
`626`	`627`	`} else {`