HBASE-25173 Remove owner related methods in TableDescriptor/TableDescriptorBuilder (#2541)

Signed-off-by: Duo Zhang <zhangduo@apache.org>

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

@@ -177,13 +177,6 @@ public interface TableDescriptor {
    */
   TableName getTableName();
 
-  /**
-   * @deprecated since 2.0.0 and will be removed in 3.0.0.
-   * @see <a href="https://issues.apache.org/jira/browse/HBASE-15583">HBASE-15583</a>
-   */
-  @Deprecated
-  String getOwnerString();
-
   /**
    * 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

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

@@ -42,7 +42,6 @@ import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.exceptions.DeserializationException;
 import org.apache.hadoop.hbase.rsgroup.RSGroupInfo;
-import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
@@ -71,12 +70,6 @@ public class TableDescriptorBuilder {
   private static final Bytes MAX_FILESIZE_KEY
           = new Bytes(Bytes.toBytes(MAX_FILESIZE));
 
-  @InterfaceAudience.Private
-  public static final String OWNER = "OWNER";
-  @InterfaceAudience.Private
-  public static final Bytes OWNER_KEY
-          = new Bytes(Bytes.toBytes(OWNER));
-
   /**
    * Used by rest interface to access this metadata attribute
    * which denotes if the table is Read Only.
@@ -485,26 +478,6 @@ public class TableDescriptorBuilder {
     return this;
   }
 
-  /**
-   * @deprecated since 2.0.0 and will be removed in 3.0.0.
-   * @see <a href="https://issues.apache.org/jira/browse/HBASE-15583">HBASE-15583</a>
-   */
-  @Deprecated
-  public TableDescriptorBuilder setOwner(User owner) {
-    desc.setOwner(owner);
-    return this;
-  }
-
-  /**
-   * @deprecated since 2.0.0 and will be removed in 3.0.0.
-   * @see <a href="https://issues.apache.org/jira/browse/HBASE-15583">HBASE-15583</a>
-   */
-  @Deprecated
-  public TableDescriptorBuilder setOwnerString(String ownerString) {
-    desc.setOwnerString(ownerString);
-    return this;
-  }
-
   public TableDescriptorBuilder setPriority(int priority) {
     desc.setPriority(priority);
     return this;
@@ -1550,38 +1523,6 @@ public class TableDescriptorBuilder {
       }
     }
 
-    /**
-     * @deprecated since 2.0.0 and will be removed in 3.0.0.
-     * @see <a href="https://issues.apache.org/jira/browse/HBASE-15583">HBASE-15583</a>
-     */
-    @Deprecated
-    public ModifyableTableDescriptor setOwner(User owner) {
-      return setOwnerString(owner != null ? owner.getShortName() : null);
-    }
-
-    /**
-     * @deprecated since 2.0.0 and will be removed in 3.0.0.
-     * @see <a href="https://issues.apache.org/jira/browse/HBASE-15583">HBASE-15583</a>
-     */
-    // used by admin.rb:alter(table_name,*args) to update owner.
-    @Deprecated
-    public ModifyableTableDescriptor setOwnerString(String ownerString) {
-      return setValue(OWNER_KEY, ownerString);
-    }
-
-    /**
-     * @deprecated since 2.0.0 and will be removed in 3.0.0.
-     * @see <a href="https://issues.apache.org/jira/browse/HBASE-15583">HBASE-15583</a>
-     */
-    @Override
-    @Deprecated
-    public String getOwnerString() {
-      // Note that every table should have an owner (i.e. should have OWNER_KEY set).
-      // hbase:meta should return system user as owner, not null (see
-      // MasterFileSystem.java:bootstrap()).
-      return getOrDefault(OWNER_KEY, Function.identity(), null);
-    }
-
     /**
      * @return the bytes in pb format
      */

hbase-endpoint/src/test/java/org/apache/hadoop/hbase/coprocessor/TestSecureExport.java

@@ -216,6 +216,7 @@ public class TestSecureExport {
             Permission.Action.EXEC,
             Permission.Action.READ,
             Permission.Action.WRITE);
+    SecureTestUtil.grantGlobal(UTIL, USER_OWNER, Permission.Action.CREATE);
     addLabels(UTIL.getConfiguration(), Arrays.asList(USER_OWNER),
             Arrays.asList(PRIVATE, CONFIDENTIAL, SECRET, TOPSECRET));
   }
@@ -236,11 +237,11 @@ public class TestSecureExport {
   public void testAccessCase() throws Throwable {
     final String exportTable = name.getMethodName();
     TableDescriptor exportHtd = TableDescriptorBuilder
-            .newBuilder(TableName.valueOf(name.getMethodName()))
+            .newBuilder(TableName.valueOf(exportTable))
             .setColumnFamily(ColumnFamilyDescriptorBuilder.of(FAMILYA))
-            .setOwnerString(USER_OWNER)
             .build();
-    SecureTestUtil.createTable(UTIL, exportHtd, new byte[][]{Bytes.toBytes("s")});
+    User owner = User.createUserForTesting(UTIL.getConfiguration(), USER_OWNER, new String[0]);
+    SecureTestUtil.createTable(UTIL, owner, exportHtd, new byte[][]{Bytes.toBytes("s")});
     SecureTestUtil.grantOnTable(UTIL, USER_RO,
             TableName.valueOf(exportTable), null, null,
             Permission.Action.READ);
@@ -340,9 +341,9 @@ public class TestSecureExport {
     final TableDescriptor exportHtd = TableDescriptorBuilder
             .newBuilder(TableName.valueOf(exportTable))
             .setColumnFamily(ColumnFamilyDescriptorBuilder.of(FAMILYA))
-            .setOwnerString(USER_OWNER)
             .build();
-    SecureTestUtil.createTable(UTIL, exportHtd, new byte[][]{Bytes.toBytes("s")});
+    User owner = User.createUserForTesting(UTIL.getConfiguration(), USER_OWNER, new String[0]);
+    SecureTestUtil.createTable(UTIL, owner, exportHtd, new byte[][]{Bytes.toBytes("s")});
     AccessTestAction putAction = () -> {
       Put p1 = new Put(ROW1);
       p1.addColumn(FAMILYA, QUAL, NOW, QUAL);
@@ -398,9 +399,8 @@ public class TestSecureExport {
       final TableDescriptor importHtd = TableDescriptorBuilder
               .newBuilder(TableName.valueOf(importTable))
               .setColumnFamily(ColumnFamilyDescriptorBuilder.of(FAMILYB))
-              .setOwnerString(USER_OWNER)
               .build();
-      SecureTestUtil.createTable(UTIL, importHtd, new byte[][]{Bytes.toBytes("s")});
+      SecureTestUtil.createTable(UTIL, owner, importHtd, new byte[][]{Bytes.toBytes("s")});
       AccessTestAction importAction = () -> {
         String[] args = new String[]{
           "-D" + Import.CF_RENAME_PROP + "=" + FAMILYA_STRING + ":" + FAMILYB_STRING,

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

@@ -804,10 +804,7 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
             + PermissionStorage.ACL_TABLE_NAME + " is not yet created. "
             + getClass().getSimpleName() + " should be configured as the first Coprocessor");
       } else {
-        String owner = desc.getOwnerString();
+        String owner = getActiveUser(c).getShortName();
-        // default the table owner to current user, if not specified.
-        if (owner == null)
-          owner = getActiveUser(c).getShortName();
         final UserPermission userPermission = new UserPermission(owner,
             Permission.newBuilder(desc.getTableName()).withActions(Action.values()).build());
         // switch to the real hbase master user for doing the RPC on the ACL table
@@ -906,8 +903,7 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
     TableDescriptor oldDesc, TableDescriptor currentDesc) throws IOException {
     final Configuration conf = c.getEnvironment().getConfiguration();
     // default the table owner to current user, if not specified.
-    final String owner = (currentDesc.getOwnerString() != null) ? currentDesc.getOwnerString() :
+    final String owner = getActiveUser(c).getShortName();
-      getActiveUser(c).getShortName();
     User.runAsLoginUser(new PrivilegedExceptionAction<Void>() {
       @Override
       public Void run() throws Exception {

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

@@ -179,8 +179,7 @@ public class SnapshotScannerHDFSAclController implements MasterCoprocessor, Mast
       // 1. Create table directories to make HDFS acls can be inherited
       hdfsAclHelper.createTableDirectories(tableName);
       // 2. Add table owner HDFS acls
-      String owner =
+      String owner = getActiveUser(c).getShortName();
-          desc.getOwnerString() == null ? getActiveUser(c).getShortName() : desc.getOwnerString();
       hdfsAclHelper.addTableAcl(tableName, Sets.newHashSet(owner), "create");
       // 3. Record table owner permission is synced to HDFS in acl table
       SnapshotScannerHDFSAclStorage.addUserTableHdfsAcl(c.getEnvironment().getConnection(), owner,

hbase-server/src/test/java/org/apache/hadoop/hbase/HBaseTestingUtility.java

@@ -2985,16 +2985,26 @@ public class HBaseTestingUtility extends HBaseZKTestingUtility {
 
   /**
    * Get a shared Connection to the cluster.
-   * this method is threadsafe.
+   * this method is thread safe.
    * @return A Connection that can be shared. Don't close. Will be closed on shutdown of cluster.
    */
   public Connection getConnection() throws IOException {
     return getAsyncConnection().toConnection();
   }
 
+  /**
+   * Get a assigned Connection to the cluster.
+   * this method is thread safe.
+   * @param user assigned user
+   * @return A Connection with assigned user.
+   */
+  public Connection getConnection(User user) throws IOException {
+    return getAsyncConnection(user).toConnection();
+  }
+
   /**
    * Get a shared AsyncClusterConnection to the cluster.
-   * this method is threadsafe.
+   * this method is thread safe.
    * @return An AsyncClusterConnection that can be shared. Don't close. Will be closed on shutdown of cluster.
    */
   public AsyncClusterConnection getAsyncConnection() throws IOException {
@@ -3003,7 +3013,7 @@ public class HBaseTestingUtility extends HBaseZKTestingUtility {
         if (connection == null) {
           try {
             User user = UserProvider.instantiate(conf).getCurrent();
-            connection = ClusterConnectionFactory.createAsyncClusterConnection(conf, null, user);
+            connection = getAsyncConnection(user);
           } catch(IOException ioe) {
             throw new UncheckedIOException("Failed to create connection", ioe);
           }
@@ -3015,6 +3025,16 @@ public class HBaseTestingUtility extends HBaseZKTestingUtility {
     }
   }
 
+  /**
+   * Get a assigned AsyncClusterConnection to the cluster.
+   * this method is thread safe.
+   * @param user assigned user
+   * @return An AsyncClusterConnection with assigned user.
+   */
+  public AsyncClusterConnection getAsyncConnection(User user) throws IOException {
+    return ClusterConnectionFactory.createAsyncClusterConnection(conf, null, user);
+  }
+
   public void closeConnection() throws IOException {
     if (hbaseAdmin != null) {
       Closeables.close(hbaseAdmin, true);

hbase-server/src/test/java/org/apache/hadoop/hbase/client/SnapshotWithAclTestBase.java

@@ -120,14 +120,17 @@ public abstract class SnapshotWithAclTestBase extends SecureTestUtil {
     USER_RW = User.createUserForTesting(conf, "rwuser", new String[0]);
     USER_RO = User.createUserForTesting(conf, "rouser", new String[0]);
     USER_NONE = User.createUserForTesting(conf, "usernone", new String[0]);
+
+    // Grant table creation permission to USER_OWNER
+    grantGlobal(TEST_UTIL, USER_OWNER.getShortName(), Permission.Action.CREATE);
   }
 
   @Before
   public void setUp() throws Exception {
-    TEST_UTIL.createTable(TableDescriptorBuilder.newBuilder(TEST_TABLE)
+    TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(TEST_TABLE)
       .setColumnFamily(
-        ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY).setMaxVersions(100).build())
+        ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY).setMaxVersions(100).build()).build();
-      .setOwner(USER_OWNER).build(), new byte[][] { Bytes.toBytes("s") });
+    createTable(TEST_UTIL, USER_OWNER, tableDescriptor, new byte[][] { Bytes.toBytes("s") });
     TEST_UTIL.waitTableEnabled(TEST_TABLE);
 
     grantOnTable(TEST_UTIL, USER_RW.getShortName(), TEST_TABLE, TEST_FAMILY, null,
@@ -200,9 +203,9 @@ public abstract class SnapshotWithAclTestBase extends SecureTestUtil {
     TableName tableName2 = TableName.valueOf(TEST_UTIL.getRandomUUID().toString());
     cloneSnapshot(snapshotName1, tableName2, false);
     verifyRows(tableName2);
-    verifyAllowed(new AccessReadAction(tableName2), USER_OWNER);
+    verifyDenied(new AccessReadAction(tableName2), USER_OWNER);
     verifyDenied(new AccessReadAction(tableName2), USER_NONE, USER_RO, USER_RW);
-    verifyAllowed(new AccessWriteAction(tableName2), USER_OWNER);
+    verifyDenied(new AccessWriteAction(tableName2), USER_OWNER);
     verifyDenied(new AccessWriteAction(tableName2), USER_RO, USER_RW, USER_NONE);
 
     // remove read permission for USER_RO.

hbase-server/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroupsWithACL.java

@@ -131,6 +131,9 @@ public class TestRSGroupsWithACL extends SecureTestUtil {
     USER_GROUP_WRITE =
         User.createUserForTesting(conf, "user_group_write", new String[] { GROUP_WRITE });
 
+    // Grant table creation permission to USER_OWNER
+    grantGlobal(TEST_UTIL, USER_OWNER.getShortName(), Permission.Action.CREATE);
+
     systemUserConnection = TEST_UTIL.getConnection();
     setUpTableAndUserPermissions();
     master = TEST_UTIL.getHBaseCluster().getMaster();
@@ -156,8 +159,7 @@ public class TestRSGroupsWithACL extends SecureTestUtil {
     ColumnFamilyDescriptorBuilder cfd = ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY);
     cfd.setMaxVersions(100);
     tableBuilder.setColumnFamily(cfd.build());
-    tableBuilder.setValue(TableDescriptorBuilder.OWNER, USER_OWNER.getShortName());
+    createTable(TEST_UTIL, USER_OWNER, tableBuilder.build(), new byte[][] { Bytes.toBytes("s") });
-    createTable(TEST_UTIL, tableBuilder.build(), new byte[][] { Bytes.toBytes("s") });
 
     // Set up initial grants
     grantGlobal(TEST_UTIL, USER_ADMIN.getShortName(), Permission.Action.ADMIN,

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/SecureTestUtil.java

@@ -775,6 +775,18 @@ public class SecureTestUtil {
     testUtil.waitUntilAllRegionsAssigned(htd.getTableName());
   }
 
+  public static void createTable(HBaseTestingUtility testUtil, User user, TableDescriptor htd)
+    throws Exception {
+    createTable(testUtil, user, htd, null);
+  }
+
+  public static void createTable(HBaseTestingUtility testUtil, User user, TableDescriptor htd,
+    byte[][] splitKeys) throws Exception {
+    try (Connection con = testUtil.getConnection(user); Admin admin = con.getAdmin()) {
+      createTable(testUtil, admin, htd, splitKeys);
+    }
+  }
+
   public static void deleteTable(HBaseTestingUtility testUtil, TableName tableName)
       throws Exception {
     deleteTable(testUtil, testUtil.getAdmin(), tableName);

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestAccessController.java

@@ -270,6 +270,9 @@ public class TestAccessController extends SecureTestUtil {
     USER_GROUP_WRITE =
         User.createUserForTesting(conf, "user_group_write", new String[] { GROUP_WRITE });
 
+    // Grant table creation permission to USER_OWNER
+    grantGlobal(TEST_UTIL, USER_OWNER.getShortName(), Action.CREATE);
+
     systemUserConnection = TEST_UTIL.getConnection();
     setUpTableAndUserPermissions();
   }
@@ -283,9 +286,8 @@ public class TestAccessController extends SecureTestUtil {
   private static void setUpTableAndUserPermissions() throws Exception {
     TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(TEST_TABLE)
       .setColumnFamily(
-        ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY).setMaxVersions(100).build())
+        ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY).setMaxVersions(100).build()).build();
-      .setOwner(USER_OWNER).build();
+    createTable(TEST_UTIL, USER_OWNER, tableDescriptor, new byte[][] { Bytes.toBytes("s") });
-    createTable(TEST_UTIL, tableDescriptor, new byte[][] { Bytes.toBytes("s") });
 
     HRegion region = TEST_UTIL.getHBaseCluster().getRegions(TEST_TABLE).get(0);
     RegionCoprocessorHost rcpHost = region.getCoprocessorHost();
@@ -1670,8 +1672,8 @@ public class TestAccessController extends SecureTestUtil {
     }
     TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(tableName)
       .setColumnFamily(ColumnFamilyDescriptorBuilder.of(family1))
-      .setColumnFamily(ColumnFamilyDescriptorBuilder.of(family2)).setOwner(USER_OWNER).build();
+      .setColumnFamily(ColumnFamilyDescriptorBuilder.of(family2)).build();
-    createTable(TEST_UTIL, tableDescriptor);
+    createTable(TEST_UTIL, USER_OWNER, tableDescriptor);
     try {
       List<UserPermission> perms =
           admin.getUserPermissions(GetUserPermissionsRequest.newBuilder(tableName).build());
@@ -1724,13 +1726,9 @@ public class TestAccessController extends SecureTestUtil {
       assertFalse("User should not be granted permission: " + upToVerify.toString(),
         hasFoundUserPermission(upToVerify, perms));
 
-      // disable table before modification
-      admin.disableTable(tableName);
-
       User newOwner = User.createUserForTesting(conf, "new_owner", new String[] {});
-      tableDescriptor =
+      grantOnTable(TEST_UTIL, newOwner.getShortName(), tableName,
-        TableDescriptorBuilder.newBuilder(tableDescriptor).setOwner(newOwner).build();
+        null, null, Permission.Action.values());
-      admin.modifyTable(tableDescriptor);
 
       perms = admin.getUserPermissions(GetUserPermissionsRequest.newBuilder(tableName).build());
       UserPermission newOwnerperm = new UserPermission(newOwner.getName(),
@@ -1758,7 +1756,7 @@ public class TestAccessController extends SecureTestUtil {
         new UserPermission(user, Permission.newBuilder().withActions(Action.values()).build()));
     }
     assertTrue("Only super users, global users and user admin has permission on table hbase:acl " +
-        "per setup", perms.size() == 5 + superUsers.size() &&
+        "per setup", perms.size() == 6 + superUsers.size() &&
         hasFoundUserPermission(adminPerms, perms));
   }
 
@@ -2278,8 +2276,8 @@ public class TestAccessController extends SecureTestUtil {
   private void createTestTable(TableName tname, byte[] cf) throws Exception {
     TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(tname)
       .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(cf).setMaxVersions(100).build())
-      .setOwner(USER_OWNER).build();
+      .build();
-    createTable(TEST_UTIL, tableDescriptor, new byte[][] { Bytes.toBytes("s") });
+    createTable(TEST_UTIL, USER_OWNER, tableDescriptor, new byte[][] { Bytes.toBytes("s") });
   }
 
   @Test
@@ -2858,7 +2856,7 @@ public class TestAccessController extends SecureTestUtil {
 
     // Verify that we can read sys-tables
     String aclTableName = PermissionStorage.ACL_TABLE_NAME.getNameAsString();
-    assertEquals(5, SUPERUSER.runAs(getPrivilegedAction(aclTableName)).size());
+    assertEquals(6, SUPERUSER.runAs(getPrivilegedAction(aclTableName)).size());
     assertEquals(0, testRegexHandler.runAs(getPrivilegedAction(aclTableName)).size());
 
     // Grant TABLE ADMIN privs to testUserPerms
@@ -3517,10 +3515,10 @@ public class TestAccessController extends SecureTestUtil {
 
     // Validate global user permission
     List<UserPermission> userPermissions;
-    assertEquals(5 + superUserCount, AccessControlClient.getUserPermissions(conn, null).size());
+    assertEquals(6 + superUserCount, AccessControlClient.getUserPermissions(conn, null).size());
-    assertEquals(5 + superUserCount,
+    assertEquals(6 + superUserCount,
       AccessControlClient.getUserPermissions(conn, HConstants.EMPTY_STRING).size());
-    assertEquals(5 + superUserCount,
+    assertEquals(6 + superUserCount,
       AccessControlClient.getUserPermissions(conn, null, HConstants.EMPTY_STRING).size());
     userPermissions = AccessControlClient.getUserPermissions(conn, null, USER_ADMIN.getName());
     verifyGetUserPermissionResult(userPermissions, 1, null, null, USER_ADMIN.getName(), superUsers);

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestAccessController3.java

@@ -189,6 +189,9 @@ public class TestAccessController3 extends SecureTestUtil {
     USER_GROUP_WRITE =
         User.createUserForTesting(conf, "user_group_write", new String[] { GROUP_WRITE });
 
+    // Grant table creation permission to USER_OWNER
+    grantGlobal(TEST_UTIL, USER_OWNER.getShortName(), Permission.Action.CREATE);
+
     systemUserConnection = TEST_UTIL.getConnection();
     setUpTableAndUserPermissions();
   }
@@ -207,9 +210,8 @@ public class TestAccessController3 extends SecureTestUtil {
   private static void setUpTableAndUserPermissions() throws Exception {
     TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(TEST_TABLE)
       .setColumnFamily(
-        ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY).setMaxVersions(100).build())
+        ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY).setMaxVersions(100).build()).build();
-      .setOwner(USER_OWNER).build();
+    createTable(TEST_UTIL, USER_OWNER, tableDescriptor, new byte[][] { Bytes.toBytes("s") });
-    createTable(TEST_UTIL, tableDescriptor, new byte[][] { Bytes.toBytes("s") });
 
     HRegion region = TEST_UTIL.getHBaseCluster().getRegions(TEST_TABLE).get(0);
     RegionCoprocessorHost rcpHost = region.getCoprocessorHost();

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestCellACLWithMultipleVersions.java

@@ -31,7 +31,6 @@ import org.apache.hadoop.hbase.HBaseClassTestRule;
 import org.apache.hadoop.hbase.HBaseTestingUtility;
 import org.apache.hadoop.hbase.TableNameTestRule;
 import org.apache.hadoop.hbase.TableNotFoundException;
-import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.client.ConnectionFactory;
@@ -125,6 +124,9 @@ public class TestCellACLWithMultipleVersions extends SecureTestUtil {
     GROUP_USER = User.createUserForTesting(conf, "group_user", new String[] { GROUP });
 
     usersAndGroups = new String[] { USER_OTHER.getShortName(), AuthUtil.toGroupEntry(GROUP) };
+
+    // Grant table creation permission to USER_OWNER
+    grantGlobal(TEST_UTIL, USER_OWNER.getShortName(), Action.CREATE);
   }
 
   @AfterClass
@@ -138,14 +140,9 @@ public class TestCellACLWithMultipleVersions extends SecureTestUtil {
       .setColumnFamily(
         ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY1).setMaxVersions(4).build())
       .setColumnFamily(
-        ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY2).setMaxVersions(4).build())
+        ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY2).setMaxVersions(4).build()).build();
-      .setOwner(USER_OWNER).build();
     // Create the test table (owner added to the _acl_ table)
-    try (Connection connection = ConnectionFactory.createConnection(TEST_UTIL.getConfiguration())) {
+    createTable(TEST_UTIL, USER_OWNER, tableDescriptor, new byte[][] { Bytes.toBytes("s") });
-      try (Admin admin = connection.getAdmin()) {
-        admin.createTable(tableDescriptor, new byte[][] { Bytes.toBytes("s") });
-      }
-    }
     TEST_UTIL.waitTableEnabled(testTable.getTableName());
     LOG.info("Sleeping a second because of HBASE-12581");
     Threads.sleep(1000);

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestCellACLs.java

@@ -31,7 +31,6 @@ import org.apache.hadoop.hbase.HBaseTestingUtility;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.TableNameTestRule;
 import org.apache.hadoop.hbase.TableNotFoundException;
-import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.client.ConnectionFactory;
@@ -127,6 +126,9 @@ public class TestCellACLs extends SecureTestUtil {
     GROUP_USER = User.createUserForTesting(conf, "group_user", new String[] { GROUP });
 
     usersAndGroups = new String[] { USER_OTHER.getShortName(), AuthUtil.toGroupEntry(GROUP) };
+
+    // Grant table creation permission to USER_OWNER
+    grantGlobal(TEST_UTIL, USER_OWNER.getShortName(), Action.CREATE);
   }
 
   @AfterClass
@@ -137,12 +139,10 @@ public class TestCellACLs extends SecureTestUtil {
   @Before
   public void setUp() throws Exception {
     // Create the test table (owner added to the _acl_ table)
-    Admin admin = TEST_UTIL.getAdmin();
     TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(testTable.getTableName())
       .setColumnFamily(
-        ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY).setMaxVersions(4).build())
+        ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY).setMaxVersions(4).build()).build();
-      .setOwner(USER_OWNER).build();
+    createTable(TEST_UTIL, USER_OWNER, tableDescriptor, new byte[][] { Bytes.toBytes("s") });
-    admin.createTable(tableDescriptor, new byte[][] { Bytes.toBytes("s") });
     TEST_UTIL.waitTableEnabled(testTable.getTableName());
     LOG.info("Sleeping a second because of HBASE-12581");
     Threads.sleep(1000);

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestHDFSAclHelper.java

@@ -39,6 +39,8 @@ import static org.junit.Assert.assertEquals;
 final class TestHDFSAclHelper {
   private static final Logger LOG = LoggerFactory.getLogger(TestHDFSAclHelper.class);
 
+  private static final String USER_OWNER = "owner";
+
   private TestHDFSAclHelper() {
   }
 
@@ -55,33 +57,41 @@ final class TestHDFSAclHelper {
     }
   }
 
-  static Table createTable(HBaseTestingUtility util, TableName tableName) throws IOException {
+  static Table createTable(HBaseTestingUtility util, TableName tableName) throws Exception {
     createNamespace(util, tableName.getNamespaceAsString());
     TableDescriptor td = getTableDescriptorBuilder(util, tableName)
         .setValue(SnapshotScannerHDFSAclHelper.ACL_SYNC_TO_HDFS_ENABLE, "true").build();
     byte[][] splits = new byte[][] { Bytes.toBytes("2"), Bytes.toBytes("4") };
-    return util.createTable(td, splits);
+    User user = User.createUserForTesting(util.getConfiguration(), USER_OWNER, new String[] {});
+    SecureTestUtil.grantGlobal(util, user.getShortName(), Permission.Action.CREATE);
+    SecureTestUtil.createTable(util, user, td, splits);
+    return util.getConnection().getTable(tableName);
   }
 
-  static Table createMobTable(HBaseTestingUtility util, TableName tableName) throws IOException {
+  static Table createMobTable(HBaseTestingUtility util, TableName tableName) throws Exception {
     createNamespace(util, tableName.getNamespaceAsString());
     TableDescriptor td = TableDescriptorBuilder.newBuilder(tableName)
         .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(COLUMN1).setMobEnabled(true)
             .setMobThreshold(0).build())
         .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(COLUMN2).setMobEnabled(true)
             .setMobThreshold(0).build())
-        .setOwner(User.createUserForTesting(util.getConfiguration(), "owner", new String[] {}))
         .setValue(SnapshotScannerHDFSAclHelper.ACL_SYNC_TO_HDFS_ENABLE, "true").build();
     byte[][] splits = new byte[][] { Bytes.toBytes("2"), Bytes.toBytes("4") };
-    return util.createTable(td, splits);
+    User user = User.createUserForTesting(util.getConfiguration(), USER_OWNER, new String[] {});
+    SecureTestUtil.grantGlobal(util, user.getShortName(), Permission.Action.CREATE);
+    SecureTestUtil.createTable(util, user, td, splits);
+    return util.getConnection().getTable(tableName);
   }
 
   static TableDescriptor createUserScanSnapshotDisabledTable(HBaseTestingUtility util,
-      TableName tableName) throws IOException {
+      TableName tableName) throws Exception {
     createNamespace(util, tableName.getNamespaceAsString());
     TableDescriptor td = getTableDescriptorBuilder(util, tableName).build();
     byte[][] splits = new byte[][] { Bytes.toBytes("2"), Bytes.toBytes("4") };
-    try (Table t = util.createTable(td, splits)) {
+    User user = User.createUserForTesting(util.getConfiguration(), USER_OWNER, new String[] {});
+    SecureTestUtil.grantGlobal(util, user.getShortName(), Permission.Action.CREATE);
+    SecureTestUtil.createTable(util, user, td, splits);
+    try (Table t = util.getConnection().getTable(tableName)) {
       put(t);
     }
     return td;
@@ -91,11 +101,10 @@ final class TestHDFSAclHelper {
       TableName tableName) {
     return TableDescriptorBuilder.newBuilder(tableName)
         .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(COLUMN1).build())
-        .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(COLUMN2).build())
+        .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(COLUMN2).build());
-        .setOwner(User.createUserForTesting(util.getConfiguration(), "owner", new String[] {}));
   }
 
-  static void createTableAndPut(HBaseTestingUtility util, TableName tableNam) throws IOException {
+  static void createTableAndPut(HBaseTestingUtility util, TableName tableNam) throws Exception {
     try (Table t = createTable(util, tableNam)) {
       put(t);
     }

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestScanEarlyTermination.java

@@ -28,7 +28,6 @@ import org.apache.hadoop.hbase.HBaseTestingUtility;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.TableNameTestRule;
 import org.apache.hadoop.hbase.TableNotFoundException;
-import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.client.ConnectionFactory;
@@ -107,6 +106,9 @@ public class TestScanEarlyTermination extends SecureTestUtil {
     // create a set of test users
     USER_OWNER = User.createUserForTesting(conf, "owner", new String[0]);
     USER_OTHER = User.createUserForTesting(conf, "other", new String[0]);
+
+    // Grant table creation permission to USER_OWNER
+    grantGlobal(TEST_UTIL, USER_OWNER.getShortName(), Action.CREATE);
   }
 
   @AfterClass
@@ -116,9 +118,8 @@ public class TestScanEarlyTermination extends SecureTestUtil {
 
   @Before
   public void setUp() throws Exception {
-    Admin admin = TEST_UTIL.getAdmin();
     TableDescriptor tableDescriptor =
-      TableDescriptorBuilder.newBuilder(testTable.getTableName()).setOwner(USER_OWNER)
+      TableDescriptorBuilder.newBuilder(testTable.getTableName())
         .setColumnFamily(
           ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY1).setMaxVersions(10).build())
         .setColumnFamily(
@@ -127,7 +128,7 @@ public class TestScanEarlyTermination extends SecureTestUtil {
         // want to confirm that the per-table configuration is properly picked up.
         .setValue(AccessControlConstants.CF_ATTRIBUTE_EARLY_OUT, "true").build();
 
-    admin.createTable(tableDescriptor);
+    createTable(TEST_UTIL, USER_OWNER, tableDescriptor);
     TEST_UTIL.waitUntilAllRegionsAssigned(testTable.getTableName());
   }
 

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestWithDisabledAuthorization.java

@@ -32,7 +32,6 @@ import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.TableNameTestRule;
 import org.apache.hadoop.hbase.TableNotFoundException;
-import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.Append;
 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
 import org.apache.hadoop.hbase.client.Connection;
@@ -167,6 +166,9 @@ public class TestWithDisabledAuthorization extends SecureTestUtil {
     USER_RO = User.createUserForTesting(conf, "rouser", new String[0]);
     USER_QUAL = User.createUserForTesting(conf, "rwpartial", new String[0]);
     USER_NONE = User.createUserForTesting(conf, "nouser", new String[0]);
+
+    // Grant table creation permission to USER_OWNER
+    grantGlobal(TEST_UTIL, USER_OWNER.getShortName(), Action.CREATE);
   }
 
   @AfterClass
@@ -177,12 +179,10 @@ public class TestWithDisabledAuthorization extends SecureTestUtil {
   @Before
   public void setUp() throws Exception {
     // Create the test table (owner added to the _acl_ table)
-    Admin admin = TEST_UTIL.getAdmin();
     TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(testTable.getTableName())
       .setColumnFamily(
-        ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY).setMaxVersions(100).build())
+        ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY).setMaxVersions(100).build()).build();
-      .setOwner(USER_OWNER).build();
+    createTable(TEST_UTIL, USER_OWNER, tableDescriptor, new byte[][] { Bytes.toBytes("s") });
-    admin.createTable(tableDescriptor, new byte[][] { Bytes.toBytes("s") });
     TEST_UTIL.waitUntilAllRegionsAssigned(testTable.getTableName());
 
     HRegion region = TEST_UTIL.getHBaseCluster().getRegions(testTable.getTableName()).get(0);

hbase-shell/src/main/ruby/hbase/admin.rb

@@ -1470,8 +1470,8 @@ module Hbase
     end
 
     # Parse arguments and update TableDescriptorBuilder accordingly
+    # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
     def update_tdb_from_arg(tdb, arg)
-      tdb.setOwnerString(arg.delete(TableDescriptorBuilder::OWNER)) if arg.include?(TableDescriptorBuilder::OWNER)
       tdb.setMaxFileSize(JLong.valueOf(arg.delete(TableDescriptorBuilder::MAX_FILESIZE))) if arg.include?(TableDescriptorBuilder::MAX_FILESIZE)
       tdb.setReadOnly(JBoolean.valueOf(arg.delete(TableDescriptorBuilder::READONLY))) if arg.include?(TableDescriptorBuilder::READONLY)
       tdb.setCompactionEnabled(JBoolean.valueOf(arg.delete(TableDescriptorBuilder::COMPACTION_ENABLED))) if arg.include?(TableDescriptorBuilder::COMPACTION_ENABLED)
@@ -1490,6 +1490,7 @@ module Hbase
       set_user_metadata(tdb, arg.delete(METADATA)) if arg[METADATA]
       set_descriptor_config(tdb, arg.delete(CONFIGURATION)) if arg[CONFIGURATION]
     end
+    # rubocop:enable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
 
     #----------------------------------------------------------------------------------------------
     # clear compaction queues

hbase-shell/src/main/ruby/shell/commands/alter.rb

@@ -95,7 +95,7 @@ There could be more than one alteration in one command:
 
   hbase> alter 't1', { NAME => 'f1', VERSIONS => 3 },
    { MAX_FILESIZE => '134217728' }, { METHOD => 'delete', NAME => 'f2' },
-   OWNER => 'johndoe', METADATA => { 'mykey' => 'myvalue' }
+   METADATA => { 'mykey' => 'myvalue' }
 EOF
       end
 

hbase-shell/src/main/ruby/shell/commands/create.rb

@@ -45,7 +45,7 @@ Examples:
 
   hbase> create 'ns1:t1', 'f1', SPLITS => ['10', '20', '30', '40']
   hbase> create 't1', 'f1', SPLITS => ['10', '20', '30', '40']
-  hbase> create 't1', 'f1', SPLITS_FILE => 'splits.txt', OWNER => 'johndoe'
+  hbase> create 't1', 'f1', SPLITS_FILE => 'splits.txt'
   hbase> create 't1', {NAME => 'f1', VERSIONS => 5}, METADATA => { 'mykey' => 'myvalue' }
   hbase> # Optionally pre-split the table into NUMREGIONS, using
   hbase> # SPLITALGO ("HexStringSplit", "UniformSplit" or classname)

hbase-shell/src/test/ruby/hbase/admin_test.rb

@@ -426,7 +426,7 @@ module Hbase
     define_test "create should fail without columns when called with options" do
       drop_test_table(@create_test_name)
       assert_raise(ArgumentError) do
-        command(:create, @create_test_name, { OWNER => 'a' })
+        command(:create, @create_test_name, { VERSIONS => '1' })
       end
     end
 
@@ -460,7 +460,6 @@ module Hbase
     define_test "create should be able to set table options" do
       drop_test_table(@create_test_name)
       command(:create, @create_test_name, 'a', 'b', 'MAX_FILESIZE' => 12345678,
-              OWNER => '987654321',
               PRIORITY => '77',
               FLUSH_POLICY => 'org.apache.hadoop.hbase.regionserver.FlushAllLargeStoresPolicy',
               REGION_MEMSTORE_REPLICATION => 'TRUE',
@@ -470,7 +469,6 @@ module Hbase
               MERGE_ENABLED => 'false')
       assert_equal(['a:', 'b:'], table(@create_test_name).get_all_columns.sort)
       assert_match(/12345678/, admin.describe(@create_test_name))
-      assert_match(/987654321/, admin.describe(@create_test_name))
       assert_match(/77/, admin.describe(@create_test_name))
       assert_match(/'COMPACTION_ENABLED' => 'false'/, admin.describe(@create_test_name))
       assert_match(/'SPLIT_ENABLED' => 'false'/, admin.describe(@create_test_name))
@@ -484,9 +482,8 @@ module Hbase
 
     define_test "create should ignore table_att" do
       drop_test_table(@create_test_name)
-      command(:create, @create_test_name, 'a', 'b', METHOD => 'table_att', OWNER => '987654321')
+      command(:create, @create_test_name, 'a', 'b', METHOD => 'table_att')
       assert_equal(['a:', 'b:'], table(@create_test_name).get_all_columns.sort)
-      assert_match(/987654321/, admin.describe(@create_test_name))
     end
 
     define_test "create should work with SPLITALGO" do