HBASE-30058 Eliminate unnecessary connection creation in snapshot operations

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

@@ -1792,8 +1792,8 @@ public SnapshotResponse snapshot(RpcController controller, SnapshotRequest reque
       LOG.info(server.getClientIdAuditPrefix() + " snapshot request for:"
         + ClientSnapshotDescriptionUtils.toString(request.getSnapshot()));
       // get the snapshot information
-      SnapshotDescription snapshot =
-        SnapshotDescriptionUtils.validate(request.getSnapshot(), server.getConfiguration());
+      SnapshotDescription snapshot = SnapshotDescriptionUtils.validate(server.getConnection(),
+        request.getSnapshot(), server.getConfiguration());
       // send back the max amount of time the client should wait for the snapshot to complete
       long waitTime = SnapshotDescriptionUtils.getMaxMasterTimeout(server.getConfiguration(),
         snapshot.getType(), SnapshotDescriptionUtils.DEFAULT_MAX_WAIT_TIME);

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

@@ -133,7 +133,7 @@ private void restoreSnapshotAcl(MasterProcedureEnv env) throws IOException {
     Configuration conf = env.getMasterServices().getConfiguration();
     if (
       restoreAcl && snapshot.hasUsersAndPermissions() && snapshot.getUsersAndPermissions() != null
-        && SnapshotDescriptionUtils.isSecurityAvailable(conf)
+        && SnapshotDescriptionUtils.isSecurityAvailable(env.getMasterServices().getConnection())
     ) {
       RestoreSnapshotHelper.restoreSnapshotAcl(snapshot, tableDescriptor.getTableName(), conf);
     }

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

@@ -556,7 +556,7 @@ private void addRegionsToInMemoryStates(List<RegionInfo> regionInfos, MasterProc
   private void restoreSnapshotAcl(final MasterProcedureEnv env) throws IOException {
     if (
       restoreAcl && snapshot.hasUsersAndPermissions() && snapshot.getUsersAndPermissions() != null
-        && SnapshotDescriptionUtils.isSecurityAvailable(env.getMasterServices().getConfiguration())
+        && SnapshotDescriptionUtils.isSecurityAvailable(env.getMasterServices().getConnection())
     ) {
       // restore acl of snapshot to table.
       RestoreSnapshotHelper.restoreSnapshotAcl(snapshot, TableName.valueOf(snapshot.getTable()),

hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/PermissionStorage.java

@@ -486,6 +486,12 @@ public static ListMultimap<String, UserPermission> getTablePermissions(Configura
       null, false);
   }
 
+  public static ListMultimap<String, UserPermission> getTablePermissions(Configuration conf,
+    TableName tableName, Table t) throws IOException {
+    return getPermissions(conf, tableName != null ? tableName.getName() : null, t, null, null, null,
+      false);
+  }
+
   public static ListMultimap<String, UserPermission> getNamespacePermissions(Configuration conf,
     String namespace) throws IOException {
     return getPermissions(conf, Bytes.toBytes(toNamespaceEntry(namespace)), null, null, null, null,

hbase-server/src/main/java/org/apache/hadoop/hbase/snapshot/SnapshotDescriptionUtils.java

@@ -33,7 +33,7 @@
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.Connection;
-import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.ipc.RpcServer;
 import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.security.access.AccessChecker;
@@ -294,14 +294,16 @@ private static Path getDefaultWorkingSnapshotDir(final Path rootDir) {
    * Convert the passed snapshot description into a 'full' snapshot description based on default
    * parameters, if none have been supplied. This resolves any 'optional' parameters that aren't
    * supplied to their default values.
+   * @param conn     connection to use for reading ACL information. Can be null if security is not
+   *                 enabled.
    * @param snapshot general snapshot descriptor
    * @param conf     Configuration to read configured snapshot defaults if snapshot is not complete
    * @return a valid snapshot description
    * @throws IllegalArgumentException if the {@link SnapshotDescription} is not a complete
    *                                  {@link SnapshotDescription}.
    */
-  public static SnapshotDescription validate(SnapshotDescription snapshot, Configuration conf)
-    throws IllegalArgumentException, IOException {
+  public static SnapshotDescription validate(Connection conn, SnapshotDescription snapshot,
+    Configuration conf) throws IllegalArgumentException, IOException {
     if (!snapshot.hasTable()) {
       throw new IllegalArgumentException(
         "Descriptor doesn't apply to a table, so we can't build it.");
@@ -350,8 +352,8 @@ public static SnapshotDescription validate(SnapshotDescription snapshot, Configu
     snapshot = builder.build();
 
     // set the acl to snapshot if security feature is enabled.
-    if (isSecurityAvailable(conf)) {
-      snapshot = writeAclToSnapshotDescription(snapshot, conf);
+    if (isSecurityAvailable(conn)) {
+      snapshot = writeAclToSnapshotDescription(conn, snapshot, conf);
     }
     return snapshot;
   }
@@ -474,21 +476,22 @@ public static boolean isSnapshotOwner(org.apache.hadoop.hbase.client.SnapshotDes
     return user.getShortName().equals(snapshot.getOwner());
   }
 
-  public static boolean isSecurityAvailable(Configuration conf) throws IOException {
-    try (Connection conn = ConnectionFactory.createConnection(conf);
-      Admin admin = conn.getAdmin()) {
+  public static boolean isSecurityAvailable(Connection conn) throws IOException {
+    try (Admin admin = conn.getAdmin()) {
       return admin.tableExists(PermissionStorage.ACL_TABLE_NAME);
     }
   }
 
-  private static SnapshotDescription writeAclToSnapshotDescription(SnapshotDescription snapshot,
-    Configuration conf) throws IOException {
+  private static SnapshotDescription writeAclToSnapshotDescription(Connection conn,
+    SnapshotDescription snapshot, Configuration conf) throws IOException {
     ListMultimap<String, UserPermission> perms =
       User.runAsLoginUser(new PrivilegedExceptionAction<ListMultimap<String, UserPermission>>() {
         @Override
         public ListMultimap<String, UserPermission> run() throws Exception {
-          return PermissionStorage.getTablePermissions(conf,
-            TableName.valueOf(snapshot.getTable()));
+          try (Table aclTable = conn.getTable(PermissionStorage.ACL_TABLE_NAME)) {
+            return PermissionStorage.getTablePermissions(conf,
+              TableName.valueOf(snapshot.getTable()), aclTable);
+          }
         }
       });
     return snapshot.toBuilder()

hbase-server/src/test/java/org/apache/hadoop/hbase/master/assignment/TestMergeTableRegionsProcedure.java

@@ -371,7 +371,7 @@ public void testMergingRegionWhileTakingSnapshot() throws Exception {
       new SnapshotDescription("SnapshotProcedureTest", tableName, SnapshotType.FLUSH);
     SnapshotProtos.SnapshotDescription snapshotProto =
       ProtobufUtil.createHBaseProtosSnapshotDesc(snapshot);
-    snapshotProto = SnapshotDescriptionUtils.validate(snapshotProto,
+    snapshotProto = SnapshotDescriptionUtils.validate(null, snapshotProto,
       UTIL.getHBaseCluster().getMaster().getConfiguration());
     long snapshotProcId = procExec.submitProcedure(
       new TestSnapshotProcedure.DelaySnapshotProcedure(procExec.getEnvironment(), snapshotProto));

hbase-server/src/test/java/org/apache/hadoop/hbase/master/assignment/TestSplitTableRegionProcedure.java

@@ -549,7 +549,7 @@ public void testSplitRegionWhileTakingSnapshot() throws Exception {
       new SnapshotDescription("SnapshotProcedureTest", tableName, SnapshotType.FLUSH);
     SnapshotProtos.SnapshotDescription snapshotProto =
       ProtobufUtil.createHBaseProtosSnapshotDesc(snapshot);
-    snapshotProto = SnapshotDescriptionUtils.validate(snapshotProto,
+    snapshotProto = SnapshotDescriptionUtils.validate(null, snapshotProto,
       UTIL.getHBaseCluster().getMaster().getConfiguration());
     long snapshotProcId = procExec.submitProcedure(
       new TestSnapshotProcedure.DelaySnapshotProcedure(procExec.getEnvironment(), snapshotProto));

hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestSnapshotProcedure.java

@@ -117,7 +117,8 @@ public void setup() throws Exception {
     SNAPSHOT_NAME = "SnapshotProcedureTest";
     snapshot = new SnapshotDescription(SNAPSHOT_NAME, TABLE_NAME, SnapshotType.FLUSH);
     snapshotProto = ProtobufUtil.createHBaseProtosSnapshotDesc(snapshot);
-    snapshotProto = SnapshotDescriptionUtils.validate(snapshotProto, master.getConfiguration());
+    snapshotProto =
+      SnapshotDescriptionUtils.validate(null, snapshotProto, master.getConfiguration());
     final byte[][] splitKeys = new RegionSplitter.HexStringSplit().split(10);
     Table table = TEST_UTIL.createTable(TABLE_NAME, CF, splitKeys);
     TEST_UTIL.loadTable(table, CF, false);

hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestSnapshotProcedureEarlyExpiration.java

@@ -81,7 +81,8 @@ public void setup() throws Exception { // Copied from TestSnapshotProcedure with
       properties);
 
     snapshotProto = ProtobufUtil.createHBaseProtosSnapshotDesc(snapshot);
-    snapshotProto = SnapshotDescriptionUtils.validate(snapshotProto, master.getConfiguration());
+    snapshotProto =
+      SnapshotDescriptionUtils.validate(null, snapshotProto, master.getConfiguration());
     final byte[][] splitKeys = new RegionSplitter.HexStringSplit().split(10);
     Table table = TEST_UTIL.createTable(TABLE_NAME, CF, splitKeys);
     TEST_UTIL.loadTable(table, CF, false);

hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestSnapshotRegionProcedure.java

@@ -89,7 +89,8 @@ public void setup() throws Exception {
     SnapshotDescription snapshot =
       new SnapshotDescription(SNAPSHOT_NAME, tableName, SnapshotType.FLUSH);
     snapshotProto = ProtobufUtil.createHBaseProtosSnapshotDesc(snapshot);
-    snapshotProto = SnapshotDescriptionUtils.validate(snapshotProto, master.getConfiguration());
+    snapshotProto =
+      SnapshotDescriptionUtils.validate(null, snapshotProto, master.getConfiguration());
     final byte[][] splitKeys = new RegionSplitter.HexStringSplit().split(10);
     Table table = TEST_UTIL.createTable(tableName, cf, splitKeys);
     TEST_UTIL.loadTable(table, cf, false);

hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestSnapshotVerifyProcedure.java

@@ -86,7 +86,7 @@ public void setup() throws Exception {
     TEST_UTIL.getAdmin().flush(tableName);
 
     // prepare unverified snapshot
-    snapshotProto = SnapshotDescriptionUtils.validate(snapshotProto, conf);
+    snapshotProto = SnapshotDescriptionUtils.validate(null, snapshotProto, conf);
     Path rootDir = CommonFSUtils.getRootDir(conf);
     Path workingDir = SnapshotDescriptionUtils.getWorkingSnapshotDir(snapshotProto, rootDir, conf);
     FileSystem workingDirFs = workingDir.getFileSystem(conf);

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRSSnapshotVerifier.java

@@ -74,7 +74,7 @@ public void setup() throws Exception {
 
     // prepare unverified snapshot
     Configuration conf = TEST_UTIL.getConfiguration();
-    snapshotProto = SnapshotDescriptionUtils.validate(snapshotProto, conf);
+    snapshotProto = SnapshotDescriptionUtils.validate(null, snapshotProto, conf);
     Path rootDir = CommonFSUtils.getRootDir(conf);
     Path workingDir = SnapshotDescriptionUtils.getWorkingSnapshotDir(snapshotProto, rootDir, conf);
     FileSystem workingDirFs = workingDir.getFileSystem(conf);

hbase-server/src/test/java/org/apache/hadoop/hbase/snapshot/TestSnapshotDescriptionUtils.java

@@ -82,8 +82,8 @@ public void cleanupFS() throws Exception {
   public void testValidateMissingTableName() throws IOException {
     Configuration conf = new Configuration(false);
     try {
-      SnapshotDescriptionUtils.validate(SnapshotDescription.newBuilder().setName("fail").build(),
-        conf);
+      SnapshotDescriptionUtils.validate(null,
+        SnapshotDescription.newBuilder().setName("fail").build(), conf);
       fail("Snapshot was considered valid without a table name");
     } catch (IllegalArgumentException e) {
       LOG.debug("Correctly failed when snapshot doesn't have a tablename");