HBASE-25492: Create table with rsgroup

Author: arshadmohammad

hbase-client/src/main/java/org/apache/hadoop/hbase/HTableDescriptor.java

@@ -22,6 +22,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -987,4 +988,9 @@ public ColumnFamilyDescriptor getColumnFamily(byte[] name) {
   protected ModifyableTableDescriptor getDelegateeForModification() {
     return delegatee;
   }
+
+  @Override
+  public Optional<String> getRegionServerGroup() {
+    return delegatee.getRegionServerGroup();
+  }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptor.java

@@ -23,6 +23,7 @@
 import java.util.Comparator;
 import java.util.Iterator;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -360,4 +361,11 @@ default boolean matchReplicationScope(boolean enabled) {
     }
     return !enabled;
   }
+
+  /**
+   * Get the region server group this table belongs to. The regions of this table will be placed
+   * only on the region servers within this group. If not present, will be placed on
+   * {@link org.apache.hadoop.hbase.rsgroup.RSGroupInfo#DEFAULT_GROUP}.
+   */
+  Optional<String> getRegionServerGroup();
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptorBuilder.java

@@ -41,6 +41,7 @@
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.exceptions.DeserializationException;
 import org.apache.hadoop.hbase.exceptions.HBaseException;
+import org.apache.hadoop.hbase.rsgroup.RSGroupInfo;
 import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.PrettyPrinter;
@@ -192,6 +193,9 @@ public class TableDescriptorBuilder {
   private static final Bytes PRIORITY_KEY
           = new Bytes(Bytes.toBytes(PRIORITY));
 
+  private static final Bytes RSGROUP_KEY =
+      new Bytes(Bytes.toBytes(RSGroupInfo.TABLE_DESC_PROP_GROUP));
+
   /**
    * Relative priority of the table used for rpc scheduling
    */
@@ -594,6 +598,11 @@ public TableDescriptorBuilder setReplicationScope(int scope) {
     return this;
   }
 
+  public TableDescriptorBuilder setRegionServerGroup(String group) {
+    desc.setValue(RSGROUP_KEY, group);
+    return this;
+  }
+
   public TableDescriptor build() {
     return new ModifyableTableDescriptor(desc);
   }
@@ -1647,6 +1656,16 @@ private static TableDescriptor parseFrom(final byte[] bytes)
     public int getColumnFamilyCount() {
       return families.size();
     }
+
+    @Override
+    public Optional<String> getRegionServerGroup() {
+      Bytes value = values.get(RSGROUP_KEY);
+      if (value != null) {
+        return Optional.of(Bytes.toString(value.get(), value.getOffset(), value.getLength()));
+      } else {
+        return Optional.empty();
+      }
+    }
   }
 
   private static Optional<CoprocessorDescriptor> toCoprocessorDescriptor(String spec) {

hbase-common/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupInfo.java

@@ -38,6 +38,7 @@
 public class RSGroupInfo {
   public static final String DEFAULT_GROUP = "default";
   public static final String NAMESPACE_DESC_PROP_GROUP = "hbase.rsgroup.name";
+  public static final String TABLE_DESC_PROP_GROUP = "hbase.rsgroup.name";
 
   private final String name;
   // Keep servers in a sorted set so has an expected ordering when displayed.

hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupAdminEndpoint.java

@@ -485,7 +485,7 @@ boolean rsgroupHasServersOnline(TableDescriptor desc) throws IOException {
   }
 
   void assignTableToGroup(TableDescriptor desc) throws IOException {
-    RSGroupInfo rsGroupInfo = groupInfoManager.determineRSGroupInfoForTable(desc.getTableName());
+    RSGroupInfo rsGroupInfo = determineRSGroupInfoForTable(desc);
     if (rsGroupInfo == null) {
       throw new ConstraintException("Default RSGroup for this table " + desc.getTableName()
         + " does not exist.");
@@ -508,7 +508,7 @@ public void preCreateTableAction(
     if (desc.getTableName().isSystemTable()) {
       return;
     }
-    RSGroupInfo rsGroupInfo = groupInfoManager.determineRSGroupInfoForTable(desc.getTableName());
+    RSGroupInfo rsGroupInfo = determineRSGroupInfoForTable(desc);
     if (rsGroupInfo == null) {
       throw new ConstraintException("Default RSGroup for this table " + desc.getTableName()
         + " does not exist.");
@@ -523,6 +523,20 @@ public void preCreateTableAction(
     }
   }
 
+  private RSGroupInfo determineRSGroupInfoForTable(final TableDescriptor desc) throws IOException {
+    Optional<String> optGroupNameOfTable = desc.getRegionServerGroup();
+    if (optGroupNameOfTable.isPresent()) {
+      final RSGroupInfo rsGroup = groupInfoManager.getRSGroup(optGroupNameOfTable.get());
+      if (rsGroup == null) {
+        // When rs group is set in table descriptor then it must exist
+        throw new ConstraintException(
+            "Region server group " + optGroupNameOfTable.get() + " does not exist.");
+      } else {
+        return rsGroup;
+      }
+    }
+    return groupInfoManager.determineRSGroupInfoForTable(desc.getTableName());
+  }
   // Remove table from its RSGroup.
   @Override
   public void postDeleteTable(ObserverContext<MasterCoprocessorEnvironment> ctx,

hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupAdminServer.java

@@ -27,11 +27,15 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.Future;
+import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.NamespaceDescriptor;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
 import org.apache.hadoop.hbase.client.TableState;
 import org.apache.hadoop.hbase.constraint.ConstraintException;
 import org.apache.hadoop.hbase.master.HMaster;
@@ -43,7 +47,9 @@
 import org.apache.hadoop.hbase.master.TableStateManager;
 import org.apache.hadoop.hbase.master.assignment.AssignmentManager;
 import org.apache.hadoop.hbase.master.assignment.RegionStateNode;
+import org.apache.hadoop.hbase.master.procedure.ProcedureSyncWait;
 import org.apache.hadoop.hbase.net.Address;
+import org.apache.hadoop.hbase.procedure2.Procedure;
 import org.apache.hadoop.hbase.util.Pair;
 import org.apache.hbase.thirdparty.com.google.common.collect.Maps;
 import org.apache.yetus.audience.InterfaceAudience;
@@ -302,78 +308,6 @@ private boolean waitForRegionMovement(List<Pair<RegionInfo, Future<byte[]>>> reg
     return allRegionsMoved;
   }
 
-  /**
-   * Moves regions of tables which are not on target group servers.
-   *
-   * @param tables    the tables that will move to new group
-   * @param targetGrp the target group
-   * @throws IOException if moving the region fails
-   */
-  private void moveTableRegionsToGroup(Set<TableName> tables, RSGroupInfo targetGrp)
-    throws IOException {
-    List<ServerName> targetGrpSevers = new ArrayList<>(targetGrp.getServers().size());
-    for (ServerName serverName : master.getServerManager().getOnlineServers().keySet()) {
-      if (targetGrp.getServers().contains(serverName.getAddress())) {
-        targetGrpSevers.add(serverName);
-      }
-    }
-    //Set true to indicate at least one region movement failed
-    boolean errorInRegionMove;
-    int retry = 0;
-    List<Pair<RegionInfo, Future<byte[]>>> assignmentFutures = new ArrayList<>();
-    do {
-      errorInRegionMove = false;
-      for (TableName table : tables) {
-        if (master.getTableStateManager().isTableState(table, TableState.State.DISABLED,
-          TableState.State.DISABLING)) {
-          LOG.debug("Skipping move regions because the table {} is disabled", table);
-          continue;
-        }
-        LOG.info("Moving region(s) for table {} to RSGroup {}", table, targetGrp.getName());
-        for (RegionInfo region : master.getAssignmentManager().getRegionStates()
-          .getRegionsOfTable(table)) {
-          ServerName sn =
-            master.getAssignmentManager().getRegionStates().getRegionServerOfRegion(region);
-          if (!targetGrp.containsServer(sn.getAddress())) {
-            LOG.info("Moving region {} to RSGroup {}", region.getShortNameToLog(),
-              targetGrp.getName());
-            ServerName dest =
-              this.master.getLoadBalancer().randomAssignment(region, targetGrpSevers);
-            if (dest == null) {
-              errorInRegionMove = true;
-              continue;
-            }
-            RegionPlan rp = new RegionPlan(region, sn, dest);
-            try {
-              Future<byte[]> future = this.master.getAssignmentManager().moveAsync(rp);
-              assignmentFutures.add(Pair.newPair(region, future));
-            } catch (Exception ioe) {
-              errorInRegionMove = true;
-              LOG.error("Move region {} to group failed, will retry, current retry time is {}",
-                region.getShortNameToLog(), retry, ioe);
-            }
-
-          }
-        }
-      }
-      boolean allRegionsMoved =
-        waitForRegionMovement(assignmentFutures, targetGrp.getName(), retry);
-      if (allRegionsMoved && !errorInRegionMove) {
-        LOG.info("All regions from table(s) {} moved to target group {}.", tables,
-          targetGrp.getName());
-        return;
-      } else {
-        retry++;
-        try {
-          rsGroupInfoManager.wait(1000);
-        } catch (InterruptedException e) {
-          LOG.warn("Sleep interrupted", e);
-          Thread.currentThread().interrupt();
-        }
-      }
-    } while (retry <= 50);
-  }
-
   @edu.umd.cs.findbugs.annotations.SuppressWarnings(
       value="RCN_REDUNDANT_NULLCHECK_WOULD_HAVE_BEEN_A_NPE",
       justification="Ignoring complaint because don't know what it is complaining about")
@@ -459,7 +393,7 @@ public void moveTables(Set<TableName> tables, String targetGroup) throws IOExcep
       // targetGroup is null when a table is being deleted. In this case no further
       // action is required.
       if (targetGroup != null) {
-        moveTableRegionsToGroup(tables, rsGroupInfoManager.getRSGroup(targetGroup));
+        moveTablesAndWait(tables, targetGroup);
       }
     }
   }
@@ -582,7 +516,7 @@ public void moveServersAndTables(Set<Address> servers, Set<TableName> tables, St
         rsGroupInfoManager.getRSGroup(srcGroup).getServers(),
         targetGroup, srcGroup);
       //move regions of these tables which are not on group servers
-      moveTableRegionsToGroup(tables, rsGroupInfoManager.getRSGroup(targetGroup));
+      moveTablesAndWait(tables, targetGroup);
     }
     LOG.info("Move servers and tables done. Severs: {}, Tables: {} => {}", servers, tables,
         targetGroup);
@@ -609,6 +543,11 @@ public void removeServers(Set<Address> servers) throws IOException {
   public void renameRSGroup(String oldName, String newName) throws IOException {
     synchronized (rsGroupInfoManager) {
       rsGroupInfoManager.renameRSGroup(oldName, newName);
+      Set<TableName> updateTables = master.getTableDescriptors().getAll().values().stream()
+          .filter(t -> oldName.equals(t.getRegionServerGroup().orElse(null)))
+          .map(TableDescriptor::getTableName).collect(Collectors.toSet());
+      // Update rs group info into table descriptors, here move is not relevant
+      moveTablesAndWait(updateTables, newName);
     }
   }
 
@@ -718,4 +657,25 @@ private void checkForDeadOrOnlineServers(Set<Address> servers) throws Constraint
       }
     }
   }
+
+  private void moveTablesAndWait(Set<TableName> tables, String targetGroup) throws IOException {
+    List<Long> procIds = new ArrayList<>();
+    for (TableName tableName : tables) {
+      TableDescriptor oldTd = master.getTableDescriptors().get(tableName);
+      if (oldTd == null) {
+        continue;
+      }
+      TableDescriptor newTd =
+          TableDescriptorBuilder.newBuilder(oldTd).setRegionServerGroup(targetGroup).build();
+      procIds.add(master.modifyTable(tableName, newTd, HConstants.NO_NONCE, HConstants.NO_NONCE));
+    }
+    for (long procId : procIds) {
+      Procedure<?> proc = master.getMasterProcedureExecutor().getProcedure(procId);
+      if (proc == null) {
+        continue;
+      }
+      ProcedureSyncWait
+          .waitForProcedureToCompleteIOE(master.getMasterProcedureExecutor(), proc, Long.MAX_VALUE);
+    }
+  }
 }

hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroupsAdmin2.java

@@ -578,74 +578,41 @@ public boolean evaluate() throws Exception {
       }
     }
     assert srcServer != null;
-    RegionInfo ri = TEST_UTIL.getMiniHBaseCluster().getMaster().getAssignmentManager().
-        getRegionInfo(Bytes.toBytesBinary(rregion));
-    RegionStateNode rsn =
-        TEST_UTIL.getMiniHBaseCluster().getMaster().getAssignmentManager().getRegionStates()
-            .getRegionStateNode(ri);
+    RegionInfo ri = master.getAssignmentManager().
+      getRegionInfo(Bytes.toBytesBinary(rregion));
+    RegionStateNode rsn = master.getAssignmentManager().getRegionStates().getRegionStateNode(ri);
     rsn.setState(RegionState.State.SPLITTING);
 
-    // move table to group
-    Thread t2 = new Thread(() -> {
-      LOG.info("thread2 start running, to move regions");
-      try {
-        rsGroupAdmin.moveTables(Sets.newHashSet(tableName), newGroup.getName());
-      } catch (IOException e) {
-        LOG.error("move server error", e);
-      }
-    });
-    t2.start();
+    rsGroupAdmin.moveTables(Sets.newHashSet(tableName), newGroup.getName());
 
-    // start thread to recover region state
-    final ServerName ss = srcServer;
-    final String sregion = rregion;
-    AtomicBoolean changed = new AtomicBoolean(false);
-    Thread t1 = new Thread(() -> {
-      LOG.info("thread1 start running, will recover region state");
-      long current = System.currentTimeMillis();
-      while (System.currentTimeMillis() - current <= 50000) {
-        List<RegionInfo> regions = master.getAssignmentManager().getRegionsOnServer(ss);
-        List<RegionInfo> tableRegions = new ArrayList<>();
-        for (RegionInfo regionInfo : regions) {
-          if (regionInfo.getTable().equals(tableName)) {
-            tableRegions.add(regionInfo);
-          }
+    // Except splitting region all must have moved.
+    final Address groupServer = newGroup.getServers().iterator().next();
+    for (RegionInfo regionInfo : master.getAssignmentManager().getAssignedRegions()) {
+      if (regionInfo.getTable().equals(tableName)) {
+        if (regionInfo.equals(rsn.getRegionInfo())) {
+          // This is splitting region
+          assertEquals(
+            master.getAssignmentManager().getRegionStates().getRegionServerOfRegion(regionInfo),
+            srcServer);
+        } else {
+          // These are normal regions
+          assertEquals(
+            master.getAssignmentManager().getRegionStates().getRegionServerOfRegion(regionInfo)
+              .getAddress(), groupServer);
         }
-        LOG.debug("server table region size is:{}", tableRegions.size());
-        assert tableRegions.size() >= 1;
-        // when there is exactly one region left, we can determine the move operation encountered
-        // exception caused by the strange region state.
-        if (tableRegions.size() == 1) {
-          assertEquals(tableRegions.get(0).getRegionNameAsString(), sregion);
-          rsn.setState(RegionState.State.OPEN);
-          LOG.info("set region {} state OPEN", sregion);
-          changed.set(true);
-          break;
-        }
-        sleep(5000);
       }
-    });
-    t1.start();
-
-    t1.join();
-    t2.join();
+    }
 
-    TEST_UTIL.waitFor(WAIT_TIMEOUT, new Waiter.Predicate<Exception>() {
-      @Override
-      public boolean evaluate() {
-        if (changed.get()) {
-          boolean serverHasTableRegions = false;
-          for (RegionInfo regionInfo : master.getAssignmentManager().getRegionsOnServer(ss)) {
-            if (regionInfo.getTable().equals(tableName)) {
-              serverHasTableRegions = true;
-              break;
-            }
-          }
-          return !serverHasTableRegions && !rsn.getRegionLocation().equals(ss);
-        }
-        return false;
+    rsn.setState(RegionState.State.OPEN);
+    // Retry table movement
+    rsGroupAdmin.moveTables(Sets.newHashSet(tableName), newGroup.getName());
+    for (RegionInfo regionInfo : master.getAssignmentManager().getAssignedRegions()) {
+      if (regionInfo.getTable().equals(tableName)) {
+        assertEquals(
+          master.getAssignmentManager().getRegionStates().getRegionServerOfRegion(regionInfo)
+            .getAddress(), groupServer);
       }
-    });
+    }
   }
 
   @Test