HBASE-28158 Decouple RIT list management from TRSP

Author: ukumawat

hbase-server/src/main/jamon/org/apache/hadoop/hbase/tmpl/master/AssignmentManagerStatusTmpl.jamon

@@ -1033,6 +1033,8 @@ private void finishActiveMasterInitialization() throws IOException, InterruptedE
     Map<Class<?>, List<Procedure<MasterProcedureEnv>>> procsByType = procedureExecutor
       .getActiveProceduresNoCopy().stream().collect(Collectors.groupingBy(p -> p.getClass()));
 
+    // This manager must be accessed AFTER hbase:meta is confirmed on line..
+    this.tableStateManager = new TableStateManager(this);
     // Create Assignment Manager
     this.assignmentManager = createAssignmentManager(this, masterRegion);
     this.assignmentManager.start();
@@ -1056,8 +1058,6 @@ private void finishActiveMasterInitialization() throws IOException, InterruptedE
         .map(p -> (ServerCrashProcedure) p).map(p -> p.getServerName()).collect(Collectors.toSet()),
       Sets.union(rsListStorage.getAll(), walManager.getLiveServersFromWALDir()),
       walManager.getSplittingServersFromWALDir());
-    // This manager must be accessed AFTER hbase:meta is confirmed on line..
-    this.tableStateManager = new TableStateManager(this);
 
     startupTaskGroup.addTask("Initializing ZK system trackers");
     initializeZKBasedSystemTrackers();
@@ -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.getRegionStates().hasRegionsInTransition()
+        && this.assignmentManager.hasRegionsInTransition()
     ) {
       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.getRegionStates().getRegionsInTransitionCount()
+        && this.assignmentManager.getRegionsInTransitionCount()
             >= maxRegionsInTransition
         && EnvironmentEdgeManager.currentTime() <= cutoffTime
     ) {
@@ -3081,7 +3081,7 @@ public ClusterMetrics getClusterMetricsWithoutCoprocessor(EnumSet<Option> option
         case REGIONS_IN_TRANSITION: {
           if (assignmentManager != null) {
             builder.setRegionsInTransition(
-              assignmentManager.getRegionStates().getRegionsStateInTransition());
+              assignmentManager.getRegionsStateInTransition());
           }
           break;
         }

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

@@ -22,11 +22,14 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -232,12 +235,15 @@ public class AssignmentManager {
 
   private final int forceRegionRetainmentRetries;
 
+  private final RegionInTransitionTracker regionInTransitionTracker;
+
   public AssignmentManager(MasterServices master, MasterRegion masterRegion) {
     this(master, masterRegion, new RegionStateStore(master, masterRegion));
   }
 
   AssignmentManager(MasterServices master, MasterRegion masterRegion, RegionStateStore stateStore) {
     this.master = master;
+    regionInTransitionTracker = new RegionInTransitionTracker(master.getTableStateManager());
     this.regionStateStore = stateStore;
     this.metrics = new MetricsAssignmentManager();
     this.masterRegion = masterRegion;
@@ -411,6 +417,7 @@ public void stop() {
 
     // Stop the RegionStateStore
     regionStates.clear();
+    regionInTransitionTracker.clear();
 
     // Update meta events (for testing)
     if (hasProcExecutor) {
@@ -1093,7 +1100,7 @@ private int submitUnassignProcedure(TableName tableName,
       regionNode.lock();
       try {
         if (shouldSubmit.apply(regionNode)) {
-          if (regionNode.isInTransition()) {
+          if (regionNode.isOngoingTRSP()) {
             logRIT.accept(regionNode);
             inTransitionCount++;
             continue;
@@ -1704,7 +1711,7 @@ public boolean isRegionTwiceOverThreshold(final RegionInfo regionInfo) {
     protected void update(final AssignmentManager am) {
       final RegionStates regionStates = am.getRegionStates();
       this.statTimestamp = EnvironmentEdgeManager.currentTime();
-      update(regionStates.getRegionsStateInTransition(), statTimestamp);
+      update(am.getRegionsStateInTransition(), statTimestamp);
       update(regionStates.getRegionFailedOpen(), statTimestamp);
 
       if (LOG.isDebugEnabled() && ritsOverThreshold != null && !ritsOverThreshold.isEmpty()) {
@@ -1794,6 +1801,7 @@ public void joinCluster() throws IOException {
    */
   // Public so can be run by the Master as part of the startup. Needs hbase:meta to be online.
   // Needs to be done after the table state manager has been started.
+  //TODO umesh we can update the rit map here
   public void processOfflineRegions() {
     TransitRegionStateProcedure[] procs =
       regionStates.getRegionStateNodes().stream().filter(rsn -> rsn.isInState(State.OFFLINE))
@@ -1873,6 +1881,7 @@ public void visitRegionState(Result result, final RegionInfo regionInfo, final S
       if (regionNode.getProcedure() != null) {
         regionNode.getProcedure().stateLoaded(AssignmentManager.this, regionNode);
       }
+      regionInTransitionTracker.handleRegionStateNodeOperation(regionNode);
     }
   };
 
@@ -2046,17 +2055,59 @@ public Pair<Integer, Integer> getReopenStatus(TableName tableName) {
     return new Pair<Integer, Integer>(ritCount, states.size());
   }
 
+  // This comparator sorts the RegionStates by time stamp then Region name.
+  // Comparing by timestamp alone can lead us to discard different RegionStates that happen
+  // to share a timestamp.
+  private static class RegionStateStampComparator implements Comparator<RegionState> {
+    @Override
+    public int compare(final RegionState l, final RegionState r) {
+      int stampCmp = Long.compare(l.getStamp(), r.getStamp());
+      return stampCmp != 0 ? stampCmp : RegionInfo.COMPARATOR.compare(l.getRegion(), r.getRegion());
+    }
+  }
+
+  public final static RegionStateStampComparator REGION_STATE_STAMP_COMPARATOR =
+    new RegionStateStampComparator();
+
   // ============================================================================================
   // TODO: Region State In Transition
   // ============================================================================================
   public boolean hasRegionsInTransition() {
-    return regionStates.hasRegionsInTransition();
+    return regionInTransitionTracker.hasRegionsInTransition();
   }
 
   public List<RegionStateNode> getRegionsInTransition() {
-    return regionStates.getRegionsInTransition();
+    return new ArrayList<RegionStateNode>(regionInTransitionTracker.getRegionsInTransition().values());
   }
 
+  public boolean isRegionInTransition(final RegionInfo regionInfo) {
+    return regionInTransitionTracker.isRegionInTransition(regionInfo);
+  }
+
+  /**
+   * Get the number of regions in transition.
+   */
+  public int getRegionsInTransitionCount() {
+    return regionInTransitionTracker.getRegionsInTransition().size();
+  }
+
+  public List<RegionState> getRegionsStateInTransition() {
+    final List<RegionState> rit = new ArrayList<RegionState>(getRegionsInTransitionCount());
+    for (RegionStateNode node : getRegionsInTransition()) {
+      rit.add(node.toRegionState());
+    }
+    return rit;
+  }
+
+  public SortedSet<RegionState> getRegionsInTransitionOrderedByTimestamp() {
+    final SortedSet<RegionState> rit = new TreeSet<RegionState>(REGION_STATE_STAMP_COMPARATOR);
+    for (RegionStateNode node : getRegionsInTransition()) {
+      rit.add(node.toRegionState());
+    }
+    return rit;
+  }
+
+
   public List<RegionInfo> getAssignedRegions() {
     return regionStates.getAssignedRegions();
   }
@@ -2122,6 +2173,8 @@ private CompletableFuture<Void> transitStateAndUpdate(RegionStateNode regionNode
       if (e != null) {
         // revert
         regionNode.setState(state);
+      }else{
+        regionInTransitionTracker.handleRegionStateNodeOperation(regionNode);
       }
     });
     return future;
@@ -2170,6 +2223,7 @@ CompletableFuture<Void> regionFailedOpen(RegionStateNode regionNode, boolean giv
         if (regionLocation != null) {
           regionStates.removeRegionFromServer(regionLocation, regionNode);
         }
+        regionInTransitionTracker.handleRegionStateNodeOperation(regionNode);
       } else {
         // revert
         regionNode.setState(state);
@@ -2230,6 +2284,7 @@ CompletableFuture<Void> persistToMeta(RegionStateNode regionNode) {
         // on table that contains state.
         setMetaAssigned(regionInfo, true);
       }
+      regionInTransitionTracker.handleRegionStateNodeOperation(regionNode);
     });
   }
 
@@ -2247,6 +2302,7 @@ public CompletableFuture<Void> regionClosedAbnormally(RegionStateNode regionNode
           regionNode.setLastHost(regionLocation);
           regionStates.removeRegionFromServer(regionLocation, regionNode);
         }
+        regionInTransitionTracker.handleRegionStateNodeOperation(regionNode);
       } else {
         // revert
         regionNode.setState(state);
@@ -2307,6 +2363,7 @@ public void markRegionAsMerged(final RegionInfo child, final ServerName serverNa
     for (RegionInfo ri : mergeParents) {
       regionStates.deleteRegion(ri);
     }
+    //TODO need to handle delete and new region
     TableDescriptor td = master.getTableDescriptors().get(child.getTable());
     regionStateStore.mergeRegions(child, mergeParents, serverName, td);
     if (shouldAssignFavoredNodes(child)) {

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

@@ -154,15 +154,15 @@ private static TransitRegionStateProcedure[] createAssignProcedures(MasterProced
           regionNode.lock();
           try {
             if (ignoreIfInTransition) {
-              if (regionNode.isInTransition()) {
+              if (regionNode.isOngoingTRSP()) {
                 return null;
               }
             } else {
               // should never fail, as we have the exclusive region lock, and the region is newly
               // created, or has been successfully closed so should not be on any servers, so SCP
               // will
               // not process it either.
-              assert !regionNode.isInTransition();
+              assert !regionNode.isOngoingTRSP();
             }
             regionNode.setProcedure(proc);
           } finally {
@@ -184,7 +184,7 @@ private static TransitRegionStateProcedure[] createAssignProcedures(MasterProced
         // apply ignoreRITs to replica regions as well.
         if (
           !ignoreIfInTransition || !env.getAssignmentManager().getRegionStates()
-            .getOrCreateRegionStateNode(ri).isInTransition()
+            .getOrCreateRegionStateNode(ri).isOngoingTRSP()
         ) {
           replicaRegionInfos.add(ri);
         }
@@ -232,7 +232,7 @@ private static TransitRegionStateProcedure[] createRoundRobinAssignProcedures(
       for (RegionInfo region : regionsAndReplicas) {
         if (
           env.getAssignmentManager().getRegionStates().getOrCreateRegionStateNode(region)
-            .isInTransition()
+            .isOngoingTRSP()
         ) {
           return null;
         }

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

@@ -0,0 +1,82 @@
+package org.apache.hadoop.hbase.master.assignment;
+
+import java.util.List;
+import java.util.concurrent.ConcurrentSkipListMap;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.client.TableState;
+import org.apache.hadoop.hbase.master.RegionState;
+import org.apache.hadoop.hbase.master.TableStateManager;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@InterfaceAudience.Private
+public class RegionInTransitionTracker {
+
+  private final List<RegionState.State> DISABLE_TABLE_REGION_STATE =
+    List.of(RegionState.State.OFFLINE, RegionState.State.CLOSED);
+  private final List<RegionState.State> ENABLE_TABLE_REGION_STATE =
+    List.of(RegionState.State.OPEN, RegionState.State.SPLIT, RegionState.State.MERGED);
+
+  private static final Logger LOG = LoggerFactory.getLogger(RegionInTransitionTracker.class);
+
+  private final ConcurrentSkipListMap<RegionInfo, RegionStateNode> regionInTransition =
+    new ConcurrentSkipListMap<>(RegionInfo.COMPARATOR);
+  private final TableStateManager tableStateManager;
+
+  public RegionInTransitionTracker(TableStateManager tableStateManager) {
+    this.tableStateManager = tableStateManager;
+  }
+
+  public boolean isRegionInTransition(final RegionInfo regionInfo) {
+    return regionInTransition.containsKey(regionInfo);
+  }
+
+  public void handleRegionStateNodeOperation(RegionStateNode regionStateNode) {
+    RegionState.State currentState = regionStateNode.getState();
+
+    if (!getExceptedRegionStates(regionStateNode).contains(currentState)) {
+      addRegionInTransition(regionStateNode);
+    } else {
+      removeRegionInTransition(regionStateNode.getRegionInfo());
+    }
+  }
+
+  public void handleRegionDelete(RegionStateNode regionStateNode) {
+    removeRegionInTransition(regionStateNode.getRegionInfo());
+  }
+
+  private List<RegionState.State> getExceptedRegionStates(RegionStateNode regionStateNode) {
+    if (tableStateManager.isTableState(regionStateNode.getTable(), TableState.State.ENABLED,
+      TableState.State.ENABLING)) {
+      return ENABLE_TABLE_REGION_STATE;
+    } else {
+      return DISABLE_TABLE_REGION_STATE;
+    }
+  }
+
+  public void addRegionInTransition(final RegionStateNode regionStateNode) {
+    if (regionInTransition.putIfAbsent(regionStateNode.getRegionInfo(), regionStateNode) == null) {
+      LOG.info("Added to RIT list" + regionStateNode.getRegionInfo().getEncodedName());
+    }
+  }
+
+  public void removeRegionInTransition(final RegionInfo regionInfo) {
+    if (regionInTransition.remove(regionInfo) != null) {
+      LOG.info("Removed from RIT list" + regionInfo);
+    }
+  }
+
+  public void clear() {
+    regionInTransition.clear();
+  }
+
+  public boolean hasRegionsInTransition() {
+    return !regionInTransition.isEmpty();
+  }
+
+  public ConcurrentSkipListMap<RegionInfo, RegionStateNode> getRegionsInTransition() {
+    return regionInTransition;
+  }
+
+}

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

@@ -18,7 +18,6 @@
 package org.apache.hadoop.hbase.master.assignment;
 
 import java.util.Arrays;
-import java.util.concurrent.ConcurrentMap;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.TableName;
@@ -78,7 +77,6 @@ public AssignmentProcedureEvent(final RegionInfo regionInfo) {
   private final RegionStateNodeLock lock;
   private final RegionInfo regionInfo;
   private final ProcedureEvent<?> event;
-  private final ConcurrentMap<RegionInfo, RegionStateNode> ritMap;
 
   // volatile only for getLastUpdate and test usage, the upper layer should sync on the
   // RegionStateNode before accessing usually.
@@ -102,10 +100,9 @@ public AssignmentProcedureEvent(final RegionInfo regionInfo) {
 
   private volatile long openSeqNum = HConstants.NO_SEQNUM;
 
-  RegionStateNode(RegionInfo regionInfo, ConcurrentMap<RegionInfo, RegionStateNode> ritMap) {
+  RegionStateNode(RegionInfo regionInfo) {
     this.regionInfo = regionInfo;
     this.event = new AssignmentProcedureEvent(regionInfo);
-    this.ritMap = ritMap;
     this.lock = new RegionStateNodeLock(regionInfo);
   }
 
@@ -161,7 +158,7 @@ public boolean isStuck() {
     return isInState(State.FAILED_OPEN) && getProcedure() != null;
   }
 
-  public boolean isInTransition() {
+  public boolean isOngoingTRSP() {
     return getProcedure() != null;
   }
 
@@ -207,14 +204,12 @@ public ServerName setRegionLocation(final ServerName serverName) {
   public TransitRegionStateProcedure setProcedure(TransitRegionStateProcedure proc) {
     assert this.procedure == null;
     this.procedure = proc;
-    ritMap.put(regionInfo, this);
     return proc;
   }
 
   public void unsetProcedure(TransitRegionStateProcedure proc) {
     assert this.procedure == proc;
     this.procedure = null;
-    ritMap.remove(regionInfo, this);
   }
 
   public TransitRegionStateProcedure getProcedure() {