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

@@ -482,8 +482,13 @@ static Map<byte[], ListMultimap<String, UserPermission>> loadAll(Configuration c
 
   public static ListMultimap<String, UserPermission> getTablePermissions(Configuration conf,
     TableName tableName) throws IOException {
-    return getPermissions(conf, tableName != null ? tableName.getName() : null, null, null, null,
-      null, false);
+    return getTablePermissions(conf, tableName, null);
+  }
+
+  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,

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

@@ -34,6 +34,7 @@
 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;
@@ -302,6 +303,27 @@ private static Path getDefaultWorkingSnapshotDir(final Path rootDir) {
    */
   public static SnapshotDescription validate(SnapshotDescription snapshot, Configuration conf)
     throws IllegalArgumentException, IOException {
+    try (Connection conn = ConnectionFactory.createConnection(conf)) {
+      return validate(conn, snapshot, conf);
+    }
+  }
+
+  /**
+   * 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. Prefer this overload over
+   * {@link #validate(SnapshotDescription, Configuration)} when the caller already holds a
+   * {@link Connection}, to avoid the cost of creating a fresh connection (which involves a ZK
+   * session and, in Kerberos environments, a TGS request to the KDC) for each call.
+   * @param conn     connection to use for reading ACL information when security is 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(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 +372,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;
   }
@@ -475,20 +497,32 @@ public static boolean isSnapshotOwner(org.apache.hadoop.hbase.client.SnapshotDes
   }
 
   public static boolean isSecurityAvailable(Configuration conf) throws IOException {
-    try (Connection conn = ConnectionFactory.createConnection(conf);
-      Admin admin = conn.getAdmin()) {
+    try (Connection conn = ConnectionFactory.createConnection(conf)) {
+      return isSecurityAvailable(conn);
+    }
+  }
+
+  /**
+   * Prefer this overload over {@link #isSecurityAvailable(Configuration)} when the caller already
+   * holds a {@link Connection}, to avoid the cost of creating a fresh connection (which involves a
+   * ZK session and, in Kerberos environments, a TGS request to the KDC) for each call.
+   */
+  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()