HBASE-28158 handle balancer throttling

Author: ukumawat

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

@@ -2007,7 +2007,7 @@ private void balanceThrottling(long nextBalanceStartTime, int maxRegionsInTransi
     // But if there are zero regions in transition, it can skip sleep to speed up.
     while (
       !interrupted && EnvironmentEdgeManager.currentTime() < nextBalanceStartTime
-        && this.assignmentManager.hasRegionsInTransition()
+        && this.assignmentManager.getOngoingTRSPCount()>0
     ) {
       try {
         Thread.sleep(100);
@@ -2019,7 +2019,7 @@ private void balanceThrottling(long nextBalanceStartTime, int maxRegionsInTransi
     // Throttling by max number regions in transition
     while (
       !interrupted && maxRegionsInTransition > 0
-        && this.assignmentManager.getRegionsInTransitionCount() >= maxRegionsInTransition
+        && this.assignmentManager.getOngoingTRSPCount() >= maxRegionsInTransition
         && EnvironmentEdgeManager.currentTime() <= cutoffTime
     ) {
       try {
@@ -2098,7 +2098,7 @@ public BalanceResponse balance(BalanceRequest request) throws IOException {
 
     synchronized (this.balancer) {
       // Only allow one balance run at at time.
-      if (this.assignmentManager.hasRegionsInTransition()) {
+      if (this.assignmentManager.getOngoingTRSPCount() >0) {
         List<RegionStateNode> regionsInTransition = assignmentManager.getRegionsInTransition();
         // if hbase:meta region is in transition, result of assignment cannot be recorded
         // ignore the force flag in that case
@@ -2113,7 +2113,7 @@ public BalanceResponse balance(BalanceRequest request) throws IOException {
 
         if (!request.isIgnoreRegionsInTransition() || metaInTransition) {
           LOG.info("Not running balancer (ignoreRIT=false" + ", metaRIT=" + metaInTransition
-            + ") because " + regionsInTransition.size() + " region(s) in transition: " + toPrint
+            + ") because " + assignmentManager.getOngoingTRSPCount() + " region(s) in transition: " + toPrint
             + (truncated ? "(truncated list)" : ""));
           return responseBuilder.build();
         }
@@ -2250,7 +2250,7 @@ public boolean normalizeRegions(final NormalizeTableFilterParams ntfp,
     if (skipRegionManagementAction("region normalizer")) {
       return false;
     }
-    if (assignmentManager.hasRegionsInTransition()) {
+    if (assignmentManager.getOngoingTRSPCount() > 0) {
       return false;
     }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignmentManager.java

@@ -2080,14 +2080,16 @@ public boolean hasRegionsInTransition() {
   }
 
   public List<RegionStateNode> getRegionsInTransition() {
-    return new ArrayList<RegionStateNode>(
-      regionInTransitionTracker.getRegionsInTransition().values());
+    return regionInTransitionTracker.getRegionsInTransition();
   }
 
   public boolean isRegionInTransition(final RegionInfo regionInfo) {
     return regionInTransitionTracker.isRegionInTransition(regionInfo);
   }
 
+  public int getOngoingTRSPCount() {
+    return regionStates.getOngoingTRSPCount();
+  }
   /**
    * Get the number of regions in transition.
    */
@@ -2096,11 +2098,7 @@ public int getRegionsInTransitionCount() {
   }
 
   public List<RegionState> getRegionsStateInTransition() {
-    final List<RegionState> rit = new ArrayList<RegionState>(getRegionsInTransitionCount());
-    for (RegionStateNode node : getRegionsInTransition()) {
-      rit.add(node.toRegionState());
-    }
-    return rit;
+    return getRegionsInTransition().stream().map(RegionStateNode::toRegionState).toList();
   }
 
   public SortedSet<RegionState> getRegionsInTransitionOrderedByTimestamp() {
@@ -2365,16 +2363,14 @@ public void markRegionAsSplit(final RegionInfo parent, final ServerName serverNa
   public void markRegionAsMerged(final RegionInfo child, final ServerName serverName,
     RegionInfo[] mergeParents) throws IOException {
     final RegionStateNode node = regionStates.getOrCreateRegionStateNode(child);
-    node.setState(State.MERGED);
     for (RegionInfo ri : mergeParents) {
       regionStates.deleteRegion(ri);
+      regionInTransitionTracker.handleRegionDelete(ri);
     }
     // TODO need to handle delete and new region
     TableDescriptor td = master.getTableDescriptors().get(child.getTable());
     regionStateStore.mergeRegions(child, mergeParents, serverName, td);
     regionInTransitionTracker.handleRegionStateNodeOperation(node);
-    Arrays.stream(mergeParents).forEach(regionInfo -> regionInTransitionTracker
-      .handleRegionStateNodeOperation(regionStates.getRegionStateNode(regionInfo)));
     if (shouldAssignFavoredNodes(child)) {
       getFavoredNodePromoter().generateFavoredNodesForMergedRegion(child, mergeParents);
     }

hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignmentManagerUtil.java

@@ -90,6 +90,7 @@ private static void unlock(List<RegionStateNode> regionNodes) {
 
   static TransitRegionStateProcedure[] createUnassignProceduresForSplitOrMerge(
     MasterProcedureEnv env, Stream<RegionInfo> regions, int regionReplication) throws IOException {
+    //TODO code to get all the replica
     List<RegionStateNode> regionNodes = regions
       .flatMap(hri -> IntStream.range(0, regionReplication)
         .mapToObj(i -> RegionReplicaUtil.getRegionInfoForReplica(hri, i)))
@@ -281,6 +282,7 @@ static void reopenRegionsForRollback(MasterProcedureEnv env, List<RegionInfo> re
   static void removeNonDefaultReplicas(MasterProcedureEnv env, Stream<RegionInfo> regions,
     int regionReplication) {
     // Remove from in-memory states
+    //TODO should we not confirm here that replica region are closed or not ?
     regions.flatMap(hri -> IntStream.range(1, regionReplication)
       .mapToObj(i -> RegionReplicaUtil.getRegionInfoForReplica(hri, i))).forEach(hri -> {
         env.getAssignmentManager().getRegionStates().deleteRegion(hri);

hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionInTransitionTracker.java

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hbase.master.assignment;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ConcurrentSkipListMap;
 import org.apache.hadoop.hbase.client.RegionInfo;
@@ -51,7 +52,11 @@ public boolean isRegionInTransition(final RegionInfo regionInfo) {
 
   public void handleRegionStateNodeOperation(RegionStateNode regionStateNode) {
     RegionState.State currentState = regionStateNode.getState();
-    // if reiong is merged or split it should not be in RIT list
+    //only consider default replica for availability
+    if(regionStateNode.getRegionInfo().getReplicaId() != RegionInfo.DEFAULT_REPLICA_ID){
+      return;
+    }
+    // if region is merged or split it should not be in RIT list
     if (
       currentState == RegionState.State.SPLIT || currentState == RegionState.State.MERGED
         || regionStateNode.getRegionInfo().isSplit()
@@ -64,8 +69,8 @@ public void handleRegionStateNodeOperation(RegionStateNode regionStateNode) {
     }
   }
 
-  public void handleRegionDelete(RegionStateNode regionStateNode) {
-    removeRegionInTransition(regionStateNode.getRegionInfo());
+  public void handleRegionDelete(RegionInfo regionInfo) {
+    removeRegionInTransition(regionInfo);
   }
 
   private List<RegionState.State> getExceptedRegionStates(RegionStateNode regionStateNode) {
@@ -99,8 +104,8 @@ public boolean hasRegionsInTransition() {
     return !regionInTransition.isEmpty();
   }
 
-  public ConcurrentSkipListMap<RegionInfo, RegionStateNode> getRegionsInTransition() {
-    return regionInTransition;
+  public List<RegionStateNode> getRegionsInTransition() {
+    return new ArrayList<>(regionInTransition.values());
   }
 
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionStateNode.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hbase.master.assignment;
 
 import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.TableName;
@@ -67,6 +68,7 @@
 public class RegionStateNode implements Comparable<RegionStateNode> {
 
   private static final Logger LOG = LoggerFactory.getLogger(RegionStateNode.class);
+  private final AtomicInteger trspCounter;
 
   private static final class AssignmentProcedureEvent extends ProcedureEvent<RegionInfo> {
     public AssignmentProcedureEvent(final RegionInfo regionInfo) {
@@ -100,10 +102,11 @@ public AssignmentProcedureEvent(final RegionInfo regionInfo) {
 
   private volatile long openSeqNum = HConstants.NO_SEQNUM;
 
-  RegionStateNode(RegionInfo regionInfo) {
+  RegionStateNode(RegionInfo regionInfo, AtomicInteger trspCounter) {
     this.regionInfo = regionInfo;
     this.event = new AssignmentProcedureEvent(regionInfo);
     this.lock = new RegionStateNodeLock(regionInfo);
+    this.trspCounter = trspCounter;
   }
 
   /**
@@ -204,11 +207,13 @@ public ServerName setRegionLocation(final ServerName serverName) {
   public TransitRegionStateProcedure setProcedure(TransitRegionStateProcedure proc) {
     assert this.procedure == null;
     this.procedure = proc;
+    trspCounter.incrementAndGet();
     return proc;
   }
 
   public void unsetProcedure(TransitRegionStateProcedure proc) {
     assert this.procedure == proc;
+    trspCounter.decrementAndGet();
     this.procedure = null;
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionStates.java

@@ -53,6 +53,8 @@ public class RegionStates {
 
   private final Object regionsMapLock = new Object();
 
+  private final AtomicInteger trspCounter = new AtomicInteger(0);
+
   // TODO: Replace the ConcurrentSkipListMaps
   /**
    * A Map from {@link RegionInfo#getRegionName()} to {@link RegionStateNode}
@@ -103,7 +105,7 @@ public boolean isRegionInRegionStates(final RegionInfo hri) {
   RegionStateNode createRegionStateNode(RegionInfo regionInfo) {
     synchronized (regionsMapLock) {
       RegionStateNode node = regionsMap.computeIfAbsent(regionInfo.getRegionName(),
-        key -> new RegionStateNode(regionInfo));
+        key -> new RegionStateNode(regionInfo,trspCounter));
 
       if (encodedRegionsMap.get(regionInfo.getEncodedName()) != node) {
         encodedRegionsMap.put(regionInfo.getEncodedName(), node);
@@ -599,6 +601,10 @@ public void addToOfflineRegions(final RegionStateNode regionNode) {
     regionOffline.put(regionNode.getRegionInfo(), regionNode);
   }
 
+  public int getOngoingTRSPCount() {
+    return trspCounter.get();
+  }
+
   // ==========================================================================
   // Region FAIL_OPEN helpers
   // ==========================================================================

hbase-server/src/main/resources/hbase-webapps/master/assignmentManagerStatus.jsp

@@ -32,7 +32,7 @@
   AssignmentManager assignmentManager = master.getAssignmentManager();
   int limit = 100;
 
-  SortedSet<RegionState> rit = assignmentManager.getRegionStates().getRegionsInTransitionOrderedByTimestamp();
+  SortedSet<RegionState> rit = assignmentManager.getRegionsInTransitionOrderedByTimestamp();
 
 if (!rit.isEmpty()) {
   long currentTime = System.currentTimeMillis();

hbase-server/src/test/java/org/apache/hadoop/hbase/master/http/TestMasterStatusUtil.java

@@ -97,7 +97,7 @@ public void setupBasicMocks() {
     regionsInTransition
       .add(new RegionState(FAKE_HRI, RegionState.State.CLOSING, 12345L, FAKE_HOST));
     Mockito.doReturn(rs).when(am).getRegionStates();
-    Mockito.doReturn(regionsInTransition).when(rs).getRegionsInTransition();
+    Mockito.doReturn(regionsInTransition).when(am).getRegionsInTransition();
     Mockito.doReturn(am).when(master).getAssignmentManager();
     Mockito.doReturn(serverManager).when(master).getServerManager();