This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 5d163835737 Support high availability for the authority module (#18425)
5d163835737 is described below
commit 5d163835737190c94aaa76c51f4d5b04c42aa6d7
Author: CYB <[email protected]>
AuthorDate: Tue Aug 11 09:39:57 2026 +0800
Support high availability for the authority module (#18425)
---
.../relational/it/schema/IoTDBTableAuthHAIT.java | 670 +++++++++++++++++++++
.../iotdb/confignode/i18n/ProcedureMessages.java | 9 +-
.../iotdb/confignode/i18n/ProcedureMessages.java | 7 +-
.../client/async/CnToDnAsyncRequestType.java | 3 +
.../CnToDnInternalServiceAsyncRequestManager.java | 6 +
.../impl/sync/AuthOperationProcedure.java | 86 ++-
.../apache/iotdb/db/i18n/DataNodeMiscMessages.java | 6 -
.../apache/iotdb/db/i18n/DataNodeMiscMessages.java | 5 -
.../org/apache/iotdb/db/auth/AuthorityChecker.java | 1 +
.../apache/iotdb/db/auth/BasicAuthorityCache.java | 9 -
.../iotdb/db/auth/ClusterAuthorityFetcher.java | 68 +--
.../apache/iotdb/db/auth/IAuthorityFetcher.java | 2 -
.../impl/DataNodeInternalRPCServiceImpl.java | 1 -
.../schemaengine/lease/MetadataLeaseManager.java | 4 +-
.../db/auth/ClusterAuthorityFetcherLeaseTest.java | 24 +-
15 files changed, 766 insertions(+), 135 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableAuthHAIT.java
b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableAuthHAIT.java
new file mode 100644
index 00000000000..6422ca01826
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableAuthHAIT.java
@@ -0,0 +1,670 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.relational.it.schema;
+
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.isession.SessionConfig;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.TableClusterIT;
+import org.apache.iotdb.itbase.env.BaseEnv;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.concurrent.Callable;
+
+import static org.junit.Assert.assertTrue;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({TableClusterIT.class})
+public class IoTDBTableAuthHAIT {
+
+ private final Logger LOGGER =
LoggerFactory.getLogger(IoTDBTableAuthHAIT.class);
+
+ private static final String DATABASE_NAME = "test_auth_db";
+ private static final String TABLE_NAME = "test_tb";
+ private static final String TREE_DB_NAME = "root.test_auth_tree";
+
+ private static final String TEST_USER = "test_user";
+ private static final String TEST_USER_INITIAL_PWD = "Test_user@336699!";
+ private static final String TEST_USER_NEW_PWD = "New_pass@2024Pwd!";
+
+ private static final String HA_USER = "ha_user";
+ private static final String HA_USER_PWD = "Ha_user@123456!";
+
+ private static final String RENAMED_USER = "renamed_user";
+ private static final String TEST_ROLE = "test_role";
+ private static final String HA_ROLE = "ha_role";
+
+ private static void initCluster() {
+ EnvFactory.getEnv()
+ .getConfig()
+ .getCommonConfig()
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+
.setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
+ .setSchemaReplicationFactor(3)
+ .setDataReplicationFactor(2);
+
+
EnvFactory.getEnv().getConfig().getConfigNodeConfig().setMetadataLeaseFenceMs(20000);
+ EnvFactory.getEnv().initClusterEnvironment(1, 3);
+ }
+
+ private static void cleanCluster() {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ private void preTestData(
+ final Statement rootStmt, final String databaseName, final String
tableName)
+ throws SQLException {
+ // Table model setup
+ rootStmt.execute("CREATE DATABASE " + databaseName);
+ rootStmt.execute("USE " + databaseName);
+ rootStmt.execute("CREATE TABLE " + tableName + " (dev STRING TAG, s1 INT32
FIELD)");
+ rootStmt.execute("INSERT INTO " + tableName + "(time, dev, s1) VALUES(1,
'dev1', 100)");
+
+ // Tree model setup
+ rootStmt.execute("SET SQL_DIALECT=tree");
+ rootStmt.execute("CREATE TIMESERIES " + TREE_DB_NAME + ".dev1.s1 WITH
DATATYPE=INT32");
+ rootStmt.execute("INSERT INTO " + TREE_DB_NAME + ".dev1(time, s1)
VALUES(1, 100)");
+ rootStmt.execute("SET SQL_DIALECT=table");
+
+ // Create user and role
+ rootStmt.execute("CREATE USER " + TEST_USER + " '" + TEST_USER_INITIAL_PWD
+ "'");
+ rootStmt.execute("CREATE ROLE " + TEST_ROLE);
+ }
+
+ @Test
+ public void testAuthHAWithOneDataNodeDown() throws Exception {
+ initCluster();
+ try {
+ final DataNodeWrapper liveDN0 =
EnvFactory.getEnv().getDataNodeWrapper(0);
+ final DataNodeWrapper liveDN1 =
EnvFactory.getEnv().getDataNodeWrapper(1);
+ final DataNodeWrapper victimDN2 =
EnvFactory.getEnv().getDataNodeWrapper(2);
+
+ // Prepare data (all 3 DNs alive)
+ try (final Connection rootConn =
+ EnvFactory.getEnv()
+ .getConnection(
+ liveDN0,
+ SessionConfig.DEFAULT_USER,
+ SessionConfig.DEFAULT_PASSWORD,
+ BaseEnv.TABLE_SQL_DIALECT);
+ final Statement rootStmt = rootConn.createStatement()) {
+ preTestData(rootStmt, DATABASE_NAME, TABLE_NAME);
+ }
+
+ // Take one DataNode down
+ victimDN2.stop();
+ Assert.assertFalse("victim DataNode should be stopped",
victimDN2.isAlive());
+
+ // Execute all HA tests via live DN-0, verify effects via DN-1
+ try (final Connection rootConn =
+ EnvFactory.getEnv()
+ .getConnection(
+ liveDN0,
+ SessionConfig.DEFAULT_USER,
+ SessionConfig.DEFAULT_PASSWORD,
+ BaseEnv.TABLE_SQL_DIALECT);
+ final Statement rootStmt = rootConn.createStatement()) {
+
+ executeUserManagementHA(rootStmt, liveDN0, liveDN1);
+ executeRoleManagementHA(rootStmt);
+ executeTablePermissionHA(rootStmt, liveDN0, liveDN1);
+ executeTableRoleBasedPermissionHA(rootStmt, liveDN1);
+ executeTreePermissionHA(rootStmt, liveDN0, liveDN1);
+ executeTreeRoleBasedPermissionHA(rootStmt, liveDN1);
+ executeCleanup(rootStmt, liveDN1);
+ }
+ } finally {
+ cleanCluster();
+ }
+ }
+
+ // ==================== User Management ====================
+
+ private void executeUserManagementHA(
+ final Statement rootStmt, final DataNodeWrapper liveDN0, final
DataNodeWrapper liveDN1)
+ throws Exception {
+
+ // Step 1: CREATE USER
+ LOGGER.info("1. start to test high availability of CREATE USER");
+ assertStatementEffect(
+ rootStmt,
+ "CREATE USER " + HA_USER + " '" + HA_USER_PWD + "'",
+ () -> userExists(rootStmt, HA_USER),
+ "CREATE USER must succeed");
+
+ // Step 2: ALTER USER SET PASSWORD
+ LOGGER.info("2. start to test high availability of ALTER USER SET
PASSWORD");
+ rootStmt.execute("ALTER USER " + TEST_USER + " SET PASSWORD '" +
TEST_USER_NEW_PWD + "'");
+ // Verify: old password fails, new password succeeds on DN-1
+ assertConnectionFails(
+ liveDN1,
+ TEST_USER,
+ TEST_USER_INITIAL_PWD,
+ BaseEnv.TABLE_SQL_DIALECT,
+ "old password should fail after password change");
+ try (Connection newConn =
+ EnvFactory.getEnv()
+ .getConnection(liveDN1, TEST_USER, TEST_USER_NEW_PWD,
BaseEnv.TABLE_SQL_DIALECT);
+ Statement s = newConn.createStatement()) {
+ s.executeQuery("LIST USER");
+ }
+
+ // Step 3: ALTER USER RENAME TO
+ LOGGER.info("3. start to test high availability of ALTER USER RENAME TO");
+ rootStmt.execute("ALTER USER " + TEST_USER + " RENAME TO " + RENAMED_USER);
+ assertTrue("old user should not exist", !userExists(rootStmt, TEST_USER));
+ assertTrue("new user should exist", userExists(rootStmt, RENAMED_USER));
+ // Verify: old name fails, new name succeeds on DN-1
+ assertConnectionFails(
+ liveDN1,
+ TEST_USER,
+ TEST_USER_NEW_PWD,
+ BaseEnv.TABLE_SQL_DIALECT,
+ "old user name should fail after rename");
+ try (Connection renamedConn =
+ EnvFactory.getEnv()
+ .getConnection(
+ liveDN1, RENAMED_USER, TEST_USER_NEW_PWD,
BaseEnv.TABLE_SQL_DIALECT);
+ Statement s = renamedConn.createStatement()) {
+ s.executeQuery("LIST USER");
+ }
+
+ // Step 4: LIST USER
+ LOGGER.info("4. start to test high availability of LIST USER");
+ assertTrue("LIST USER should return results", listHasRows(rootStmt, "LIST
USER"));
+ }
+
+ // ==================== Role Management ====================
+
+ private void executeRoleManagementHA(final Statement rootStmt) throws
Exception {
+
+ // Step 6: CREATE ROLE
+ LOGGER.info("5. start to test high availability of CREATE ROLE");
+ assertStatementEffect(
+ rootStmt,
+ "CREATE ROLE " + HA_ROLE,
+ () -> roleExists(rootStmt, HA_ROLE),
+ "CREATE ROLE must succeed");
+
+ // Grant ha_role to renamed_user, then verify
+ rootStmt.execute("GRANT ROLE " + HA_ROLE + " TO " + RENAMED_USER);
+
+ // Step 6: LIST ROLE OF USER renamed_user
+ LOGGER.info("6. start to test high availability of LIST ROLE OF USER");
+ assertTrue(
+ "LIST ROLE OF USER renamed_user should contain ha_role",
+ userHasRole(rootStmt, RENAMED_USER, HA_ROLE));
+ }
+
+ // ==================== Table Model Permission Management
====================
+
+ private void executeTablePermissionHA(
+ final Statement rootStmt, final DataNodeWrapper liveDN0, final
DataNodeWrapper liveDN1)
+ throws Exception {
+
+ final String testTable = DATABASE_NAME + "." + TABLE_NAME;
+
+ // Step 8: GRANT ROLE to user
+ LOGGER.info("7. start to test high availability of GRANT ROLE (table
model)");
+ assertStatementEffect(
+ rootStmt,
+ "GRANT ROLE " + TEST_ROLE + " TO " + RENAMED_USER,
+ () -> userHasRole(rootStmt, RENAMED_USER, TEST_ROLE),
+ "GRANT ROLE must succeed");
+
+ // Step 9: GRANT SELECT ON TABLE
+ LOGGER.info("8. start to test high availability of GRANT SELECT ON TABLE");
+ assertStatementEffect(
+ rootStmt,
+ "GRANT SELECT ON TABLE " + testTable + " TO USER " + RENAMED_USER,
+ () -> userHasPrivilege(rootStmt, RENAMED_USER, "SELECT"),
+ "GRANT SELECT ON TABLE must succeed");
+ // Verify: SELECT succeeds on DN-1
+ tableUserSelect(liveDN1, RENAMED_USER, TEST_USER_NEW_PWD, DATABASE_NAME,
TABLE_NAME, true);
+
+ // Step 10: Verify NO INSERT permission (never granted)
+ LOGGER.info("9. start to test NO INSERT permission enforcement (table
model)");
+ tableUserInsert(liveDN1, RENAMED_USER, TEST_USER_NEW_PWD, DATABASE_NAME,
TABLE_NAME, false);
+
+ // Step 11: GRANT INSERT ON TABLE
+ LOGGER.info("10. start to test high availability of GRANT INSERT ON
TABLE");
+ assertStatementEffect(
+ rootStmt,
+ "GRANT INSERT ON TABLE " + testTable + " TO USER " + RENAMED_USER,
+ () -> userHasPrivilege(rootStmt, RENAMED_USER, "INSERT"),
+ "GRANT INSERT ON TABLE must succeed");
+ tableUserInsert(liveDN1, RENAMED_USER, TEST_USER_NEW_PWD, DATABASE_NAME,
TABLE_NAME, true);
+
+ // Step 12: REVOKE INSERT ON TABLE
+ LOGGER.info("11. start to test high availability of REVOKE INSERT ON
TABLE");
+ assertStatementEffect(
+ rootStmt,
+ "REVOKE INSERT ON TABLE " + testTable + " FROM USER " + RENAMED_USER,
+ () -> !userHasPrivilege(rootStmt, RENAMED_USER, "INSERT"),
+ "REVOKE INSERT ON TABLE must succeed");
+ tableUserInsert(liveDN1, RENAMED_USER, TEST_USER_NEW_PWD, DATABASE_NAME,
TABLE_NAME, false);
+
+ // Step 13: REVOKE SELECT ON TABLE
+ LOGGER.info("12. start to test high availability of REVOKE SELECT ON
TABLE");
+ assertStatementEffect(
+ rootStmt,
+ "REVOKE SELECT ON TABLE " + testTable + " FROM USER " + RENAMED_USER,
+ () -> !userHasPrivilege(rootStmt, RENAMED_USER, "SELECT"),
+ "REVOKE SELECT ON TABLE must succeed");
+ tableUserSelect(liveDN1, RENAMED_USER, TEST_USER_NEW_PWD, DATABASE_NAME,
TABLE_NAME, false);
+
+ // Step 14: GRANT SYSTEM
+ LOGGER.info("13. start to test high availability of GRANT SYSTEM");
+ assertStatementEffect(
+ rootStmt,
+ "GRANT SYSTEM TO USER " + RENAMED_USER,
+ () -> userHasPrivilege(rootStmt, RENAMED_USER, "SYSTEM"),
+ "GRANT SYSTEM must succeed");
+
+ // Step 15: REVOKE SYSTEM
+ LOGGER.info("14. start to test high availability of REVOKE SYSTEM");
+ assertStatementEffect(
+ rootStmt,
+ "REVOKE SYSTEM FROM USER " + RENAMED_USER,
+ () -> !userHasPrivilege(rootStmt, RENAMED_USER, "SYSTEM"),
+ "REVOKE SYSTEM must succeed");
+ }
+
+ // ==================== Table Model Role-Based Permission
====================
+
+ private void executeTableRoleBasedPermissionHA(
+ final Statement rootStmt, final DataNodeWrapper liveDN1) throws
Exception {
+
+ final String testTable = DATABASE_NAME + "." + TABLE_NAME;
+
+ // Step: GRANT SELECT ON TABLE TO ROLE → user inherits via role
+ LOGGER.info("start to test high availability of GRANT SELECT ON TABLE TO
ROLE (table model)");
+ assertStatementEffect(
+ rootStmt,
+ "GRANT SELECT ON TABLE " + testTable + " TO ROLE " + TEST_ROLE,
+ () -> userHasPrivilege(rootStmt, RENAMED_USER, "SELECT"),
+ "GRANT SELECT ON TABLE TO ROLE must succeed");
+ // renamed_user inherits SELECT via test_role
+ tableUserSelect(liveDN1, RENAMED_USER, TEST_USER_NEW_PWD, DATABASE_NAME,
TABLE_NAME, true);
+
+ // Step: REVOKE SELECT ON TABLE FROM ROLE → user loses inherited privilege
+ LOGGER.info(
+ "start to test high availability of REVOKE SELECT ON TABLE FROM ROLE
(table model)");
+ assertStatementEffect(
+ rootStmt,
+ "REVOKE SELECT ON TABLE " + testTable + " FROM ROLE " + TEST_ROLE,
+ () -> !userHasPrivilege(rootStmt, RENAMED_USER, "SELECT"),
+ "REVOKE SELECT ON TABLE FROM ROLE must succeed");
+ tableUserSelect(liveDN1, RENAMED_USER, TEST_USER_NEW_PWD, DATABASE_NAME,
TABLE_NAME, false);
+ }
+
+ // ==================== Tree Model Permission Management ====================
+
+ private void executeTreePermissionHA(
+ final Statement rootStmt, final DataNodeWrapper liveDN0, final
DataNodeWrapper liveDN1)
+ throws Exception {
+
+ final String treePath = "root.test_auth_tree.**";
+
+ // Step 15: GRANT READ_DATA ON tree path
+ LOGGER.info("15. start to test high availability of GRANT READ_DATA (tree
model)");
+ rootStmt.execute("SET SQL_DIALECT=tree");
+ assertStatementEffect(
+ rootStmt,
+ "GRANT READ_DATA ON " + treePath + " TO USER " + RENAMED_USER,
+ () -> userHasPrivilege(rootStmt, RENAMED_USER, "READ_DATA"),
+ "GRANT READ_DATA must succeed");
+ // Verify: SELECT succeeds on DN-1
+ treeUserSelect(liveDN1, RENAMED_USER, TEST_USER_NEW_PWD, true);
+
+ // Step 16: Verify NO WRITE permission
+ LOGGER.info("16. start to test NO WRITE permission enforcement (tree
model)");
+ treeUserInsert(liveDN1, RENAMED_USER, TEST_USER_NEW_PWD, false);
+
+ // Step 17: GRANT WRITE_DATA ON tree path
+ LOGGER.info("17. start to test high availability of GRANT WRITE_DATA (tree
model)");
+ assertStatementEffect(
+ rootStmt,
+ "GRANT WRITE_DATA ON " + treePath + " TO USER " + RENAMED_USER,
+ () -> userHasPrivilege(rootStmt, RENAMED_USER, "WRITE_DATA"),
+ "GRANT WRITE_DATA must succeed");
+ treeUserInsert(liveDN1, RENAMED_USER, TEST_USER_NEW_PWD, true);
+
+ // Step 18: REVOKE WRITE_DATA ON tree path
+ LOGGER.info("18. start to test high availability of REVOKE WRITE_DATA
(tree model)");
+ assertStatementEffect(
+ rootStmt,
+ "REVOKE WRITE_DATA ON " + treePath + " FROM USER " + RENAMED_USER,
+ () -> !userHasPrivilege(rootStmt, RENAMED_USER, "WRITE_DATA"),
+ "REVOKE WRITE_DATA must succeed");
+ treeUserInsert(liveDN1, RENAMED_USER, TEST_USER_NEW_PWD, false);
+
+ // Step 19: REVOKE READ_DATA ON tree path
+ LOGGER.info("19. start to test high availability of REVOKE READ_DATA (tree
model)");
+ assertStatementEffect(
+ rootStmt,
+ "REVOKE READ_DATA ON " + treePath + " FROM USER " + RENAMED_USER,
+ () -> !userHasPrivilege(rootStmt, RENAMED_USER, "READ_DATA"),
+ "REVOKE READ_DATA must succeed");
+ treeUserSelect(liveDN1, RENAMED_USER, TEST_USER_NEW_PWD, false);
+
+ rootStmt.execute("SET SQL_DIALECT=table");
+ }
+
+ // ==================== Tree Model Role-Based Permission ====================
+
+ private void executeTreeRoleBasedPermissionHA(
+ final Statement rootStmt, final DataNodeWrapper liveDN1) throws
Exception {
+
+ final String treePath = "root.test_auth_tree.**";
+
+ // Grant READ_DATA to test_role, renamed_user inherits via role
+ LOGGER.info("start to test high availability of GRANT READ_DATA TO ROLE
(tree model)");
+ rootStmt.execute("SET SQL_DIALECT=tree");
+ assertStatementEffect(
+ rootStmt,
+ "GRANT READ_DATA ON " + treePath + " TO ROLE " + TEST_ROLE,
+ () -> roleHasPrivilege(rootStmt, TEST_ROLE, "READ_DATA"),
+ "GRANT READ_DATA ON TO ROLE must succeed");
+ // renamed_user inherits READ_DATA via test_role
+ assertTrue(
+ "renamed_user should inherit READ_DATA via test_role",
+ userHasPrivilege(rootStmt, RENAMED_USER, "READ_DATA"));
+ treeUserSelect(liveDN1, RENAMED_USER, TEST_USER_NEW_PWD, true);
+
+ // Revoke READ_DATA from test_role → renamed_user loses inherited privilege
+ LOGGER.info("start to test high availability of REVOKE READ_DATA FROM ROLE
(tree model)");
+ assertStatementEffect(
+ rootStmt,
+ "REVOKE READ_DATA ON " + treePath + " FROM ROLE " + TEST_ROLE,
+ () -> !roleHasPrivilege(rootStmt, TEST_ROLE, "READ_DATA"),
+ "REVOKE READ_DATA ON FROM ROLE must succeed");
+ assertTrue(
+ "renamed_user should lose READ_DATA after revoke from test_role",
+ !userHasPrivilege(rootStmt, RENAMED_USER, "READ_DATA"));
+ treeUserSelect(liveDN1, RENAMED_USER, TEST_USER_NEW_PWD, false);
+
+ rootStmt.execute("SET SQL_DIALECT=table");
+ }
+
+ // ==================== Cleanup ====================
+
+ private void executeCleanup(final Statement rootStmt, final DataNodeWrapper
liveDN1)
+ throws Exception {
+
+ // Step 21: REVOKE ROLE
+ LOGGER.info("20. start to test high availability of REVOKE ROLE");
+ assertStatementEffect(
+ rootStmt,
+ "REVOKE ROLE " + TEST_ROLE + " FROM " + RENAMED_USER,
+ () -> !userHasRole(rootStmt, RENAMED_USER, TEST_ROLE),
+ "REVOKE ROLE must succeed");
+
+ // Step 22: DROP ROLE
+ LOGGER.info("21. start to test high availability of DROP ROLE");
+ assertStatementEffect(
+ rootStmt,
+ "DROP ROLE " + HA_ROLE,
+ () -> !roleExists(rootStmt, HA_ROLE),
+ "DROP ROLE must succeed");
+
+ // Step 23: DROP USER renamed_user
+ LOGGER.info("22. start to test high availability of DROP USER
renamed_user");
+ assertStatementEffect(
+ rootStmt,
+ "DROP USER " + RENAMED_USER,
+ () -> !userExists(rootStmt, RENAMED_USER),
+ "DROP USER renamed_user must succeed");
+
+ // Step 24: DROP USER ha_user
+ LOGGER.info("23. start to test high availability of DROP USER ha_user");
+ assertStatementEffect(
+ rootStmt,
+ "DROP USER " + HA_USER,
+ () -> !userExists(rootStmt, HA_USER),
+ "DROP USER ha_user must succeed");
+ }
+
+ // ==================== Verification Helpers ====================
+
+ private void assertStatementEffect(
+ final Statement statement,
+ final String sql,
+ final Callable<Boolean> effect,
+ final String message)
+ throws Exception {
+ statement.execute(sql);
+ assertTrue(message, effect.call());
+ }
+
+ private boolean userExists(final Statement stmt, final String userName)
throws SQLException {
+ try (final ResultSet rs = stmt.executeQuery("LIST USER")) {
+ while (rs.next()) {
+ if (userName.equalsIgnoreCase(rs.getString(2))) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ private boolean roleExists(final Statement stmt, final String roleName)
throws SQLException {
+ try (final ResultSet rs = stmt.executeQuery("LIST ROLE")) {
+ while (rs.next()) {
+ if (roleName.equalsIgnoreCase(rs.getString(1))) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ private boolean listHasRows(final Statement stmt, final String sql) throws
SQLException {
+ try (final ResultSet rs = stmt.executeQuery(sql)) {
+ return rs.next();
+ }
+ }
+
+ private boolean userHasRole(final Statement stmt, final String userName,
final String roleName)
+ throws SQLException {
+ try (final ResultSet rs = stmt.executeQuery("LIST ROLE OF USER " +
userName)) {
+ while (rs.next()) {
+ if (roleName.equalsIgnoreCase(rs.getString(1))) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ private boolean userHasPrivilege(
+ final Statement stmt, final String userName, final String privilege)
throws SQLException {
+ try (final ResultSet rs = stmt.executeQuery("LIST PRIVILEGES OF USER " +
userName)) {
+ while (rs.next()) {
+ // LIST PRIVILEGES columns: Role, Scope, Privileges, GrantOption
+ if (privilege.equalsIgnoreCase(rs.getString(3))) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ private boolean roleHasPrivilege(
+ final Statement stmt, final String roleName, final String privilege)
throws SQLException {
+ try (final ResultSet rs = stmt.executeQuery("LIST PRIVILEGES OF ROLE " +
roleName)) {
+ while (rs.next()) {
+ // LIST PRIVILEGES columns: Role, Scope, Privileges, GrantOption
+ if (privilege.equalsIgnoreCase(rs.getString(3))) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ private void assertConnectionFails(
+ final DataNodeWrapper dn,
+ final String user,
+ final String password,
+ final String sqlDialect,
+ final String message) {
+ try {
+ final Connection conn = EnvFactory.getEnv().getConnection(dn, user,
password, sqlDialect);
+ conn.close();
+ Assert.fail(message + " — expected connection failure but succeeded");
+ } catch (final SQLException e) {
+ // Expected
+ }
+ }
+
+ // ==================== Table Model User Operation Helpers
====================
+
+ private void tableUserSelect(
+ final DataNodeWrapper dn,
+ final String user,
+ final String password,
+ final String databaseName,
+ final String tableName,
+ final boolean expectSuccess)
+ throws SQLException {
+ final String sql = "SELECT * FROM " + tableName;
+ if (expectSuccess) {
+ try (final Connection conn =
+ EnvFactory.getEnv().getConnection(dn, user, password,
BaseEnv.TABLE_SQL_DIALECT);
+ final Statement stmt = conn.createStatement()) {
+ stmt.execute("USE " + databaseName);
+ try (final ResultSet rs = stmt.executeQuery(sql)) {
+ assertTrue("SELECT should succeed", rs.next());
+ }
+ }
+ } else {
+ try (final Connection conn =
+ EnvFactory.getEnv().getConnection(dn, user, password,
BaseEnv.TABLE_SQL_DIALECT);
+ final Statement stmt = conn.createStatement()) {
+ stmt.execute("USE " + databaseName);
+ stmt.executeQuery(sql);
+ Assert.fail("SELECT should fail");
+ } catch (final SQLException e) {
+ assertTrue(
+ e.getMessage(),
+ e.getMessage().contains("No permissions") ||
e.getMessage().contains("Access Denied"));
+ }
+ }
+ }
+
+ private void tableUserInsert(
+ final DataNodeWrapper dn,
+ final String user,
+ final String password,
+ final String databaseName,
+ final String tableName,
+ final boolean expectSuccess)
+ throws SQLException {
+ final String sql = "INSERT INTO " + tableName + "(time, dev, s1) VALUES(2,
'dev2', 200)";
+ if (expectSuccess) {
+ try (final Connection conn =
+ EnvFactory.getEnv().getConnection(dn, user, password,
BaseEnv.TABLE_SQL_DIALECT);
+ final Statement stmt = conn.createStatement()) {
+ stmt.execute("USE " + databaseName);
+ stmt.execute(sql);
+ }
+ } else {
+ try (final Connection conn =
+ EnvFactory.getEnv().getConnection(dn, user, password,
BaseEnv.TABLE_SQL_DIALECT);
+ final Statement stmt = conn.createStatement()) {
+ stmt.execute("USE " + databaseName);
+ stmt.execute(sql);
+ Assert.fail("INSERT should fail");
+ } catch (final SQLException e) {
+ assertTrue(
+ e.getMessage(),
+ e.getMessage().contains("No permissions") ||
e.getMessage().contains("Access Denied"));
+ }
+ }
+ }
+
+ // ==================== Tree Model User Operation Helpers
====================
+
+ private void treeUserSelect(
+ final DataNodeWrapper dn,
+ final String user,
+ final String password,
+ final boolean expectSuccess)
+ throws SQLException {
+ final String sql = "SELECT * FROM " + TREE_DB_NAME + ".dev1";
+ if (expectSuccess) {
+ try (final Connection conn =
+ EnvFactory.getEnv().getConnection(dn, user, password,
BaseEnv.TREE_SQL_DIALECT);
+ final Statement stmt = conn.createStatement();
+ final ResultSet rs = stmt.executeQuery(sql)) {
+ assertTrue("Tree SELECT should succeed", rs.next());
+ }
+ } else {
+ // Tree model returns empty result set on permission denial, no
exception thrown
+ try (final Connection conn =
+ EnvFactory.getEnv().getConnection(dn, user, password,
BaseEnv.TREE_SQL_DIALECT);
+ final Statement stmt = conn.createStatement();
+ final ResultSet rs = stmt.executeQuery(sql)) {
+ Assert.assertFalse("Tree SELECT should return empty result set",
rs.next());
+ }
+ }
+ }
+
+ private void treeUserInsert(
+ final DataNodeWrapper dn,
+ final String user,
+ final String password,
+ final boolean expectSuccess)
+ throws SQLException {
+ final String sql = "INSERT INTO " + TREE_DB_NAME + ".dev1(time, s1)
VALUES(2, 200)";
+ if (expectSuccess) {
+ try (final Connection conn =
+ EnvFactory.getEnv().getConnection(dn, user, password,
BaseEnv.TREE_SQL_DIALECT);
+ final Statement stmt = conn.createStatement()) {
+ stmt.execute(sql);
+ }
+ } else {
+ try (final Connection conn =
+ EnvFactory.getEnv().getConnection(dn, user, password,
BaseEnv.TREE_SQL_DIALECT);
+ final Statement stmt = conn.createStatement()) {
+ stmt.execute(sql);
+ Assert.fail("Tree INSERT should fail");
+ } catch (final SQLException e) {
+ assertTrue(
+ e.getMessage(),
+ e.getMessage().contains("No permissions") ||
e.getMessage().contains("Access Denied"));
+ }
+ }
+ }
+}
diff --git
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
index b29244d4117..e04460375f9 100644
---
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
@@ -93,6 +93,9 @@ public final class ProcedureMessages {
public static final String AUTHENTICATION_FAILED = "Authentication failed.";
public static final String AUTH_PROCEDURE_CLEAN_DATANODE_CACHE_SUCCESSFULLY =
"Auth procedure: clean datanode cache successfully";
+ public static final String AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED =
+ "Auth plan has been committed, but DataNode permission cache
invalidation failed. "
+ + "Some DataNodes may have stale permissions; please clear their
permission cache manually.";
public static final String BEGIN_TO_CHANGE_DATANODE_STATUS_NODESTATUSMAP =
"{}, Begin to change DataNode status, nodeStatusMap: {}";
public static final String
BEGIN_TO_STOP_DATANODES_AND_KILL_THE_DATANODE_PROCESS =
@@ -356,7 +359,7 @@ public final class ProcedureMessages {
public static final String
ERROR_IN_DESERIALIZE_PROCID_THIS_PROCEDURE_WILL_BE_IGNORED_IT =
"Error in deserialize {} (procID {}). This procedure will be ignored. It
may belong to old version and cannot be used now.";
public static final String EXECUTE_AUTH_PLAN_SUCCESS_TO_INVALIDATE_DATANODES
=
- "Execute auth plan {} success. To invalidate datanodes: {}";
+ "Execute auth plan {} success.";
public static final String
EXECUTING_ON_REGION_FOR_COLUMN_IN_WHEN_DROPPING_COLUMN =
"Executing on region for column {} in {}.{} when dropping column";
public static final String FAILED_TO_ACTIVE_CQ_BECAUSE_OF_NO_SUCH_CQ =
@@ -551,8 +554,6 @@ public final class ProcedureMessages {
"Fail to drop trigger [%s] at STATE [%s]";
public static final String FAIL_TO_DROP_TRIGGER_ON_DATA_NODES =
"Fail to drop trigger [%s] on Data Nodes";
- public static final String FAIL_TO_EXECUTE_PLAN_AT_STATE =
- "Fail to execute plan [%s] at state[%s]";
public static final String FAIL_TO_REMOVE_AINODE_AT_STATE =
"Fail to remove AINode [%s] at STATE [%s], %s";
public static final String FAIL_TO_REMOVE_AINODE_ON_CONFIG_NODES =
@@ -860,8 +861,6 @@ public final class ProcedureMessages {
"Retrievable error trying to create pipe plugin [{}], state: {}";
public static final String
RETRIEVABLE_ERROR_TRYING_TO_DROP_PIPE_PLUGIN_STATE =
"Retrievable error trying to drop pipe plugin [{}], state: {}";
- public static final String RETRIEVABLE_ERROR_TRYING_TO_EXECUTE_PLAN_STATE =
- "Retrievable error trying to execute plan {}, state: {}";
public static final String RETRIEVABLE_ERROR_TRYING_TO_REMOVE_AINODE_STATE =
"Retrievable error trying to remove AINode [{}], state [{}]";
public static final String ROLLBACK_CREATETABLE_COSTS_MS = "Rollback
CreateTable-{} costs {}ms.";
diff --git
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
index 0569e1b6261..3014e1c0076 100644
---
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
@@ -91,6 +91,8 @@ public final class ProcedureMessages {
public static final String AUTHENTICATION_FAILED = "认证失败。";
public static final String AUTH_PROCEDURE_CLEAN_DATANODE_CACHE_SUCCESSFULLY =
"Auth procedure：成功清理 datanode 缓存";
+ public static final String AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED =
+ "权限计划已提交，但 DataNode 权限缓存失效失败。部分 DataNode 可能仍持有过期权限，请手动清理权限缓存。";
public static final String BEGIN_TO_CHANGE_DATANODE_STATUS_NODESTATUSMAP =
"{}, 开始修改 DataNode 状态，nodeStatusMap：{}";
public static final String
BEGIN_TO_STOP_DATANODES_AND_KILL_THE_DATANODE_PROCESS =
@@ -344,7 +346,7 @@ public final class ProcedureMessages {
public static final String
ERROR_IN_DESERIALIZE_PROCID_THIS_PROCEDURE_WILL_BE_IGNORED_IT =
"反序列化 {}（procID {}）出错。该 procedure 将被忽略。它可能属于旧版本，目前无法使用。";
public static final String EXECUTE_AUTH_PLAN_SUCCESS_TO_INVALIDATE_DATANODES
=
- "执行 auth plan {} 成功。使 datanode 缓存失效：{}";
+ "执行 auth plan {} 成功。";
public static final String
EXECUTING_ON_REGION_FOR_COLUMN_IN_WHEN_DROPPING_COLUMN =
"删除列时在表 {}.{} 中列 {} 对应的 region 上执行";
public static final String FAILED_TO_ACTIVE_CQ_BECAUSE_OF_NO_SUCH_CQ =
@@ -531,7 +533,6 @@ public final class ProcedureMessages {
"重试 {} 次后删除 pipe plugin [{}] 仍失败";
public static final String FAIL_TO_DROP_TRIGGER_AT_STATE = "在 STATE [%s] 处删除
trigger [%s] 失败";
public static final String FAIL_TO_DROP_TRIGGER_ON_DATA_NODES = "在 DataNode
上删除 trigger [%s] 失败";
- public static final String FAIL_TO_EXECUTE_PLAN_AT_STATE = "在 state[%s] 处执行
plan [%s] 失败";
public static final String FAIL_TO_REMOVE_AINODE_AT_STATE = "在 STATE [%s]
处移除 AINode [%s] 失败，%s";
public static final String FAIL_TO_REMOVE_AINODE_ON_CONFIG_NODES =
"在 ConfigNode [%s] 上移除 [%s] 个 AINode 失败";
@@ -818,8 +819,6 @@ public final class ProcedureMessages {
"尝试创建 pipe plugin [{}] 时发生可重试错误，状态：{}";
public static final String
RETRIEVABLE_ERROR_TRYING_TO_DROP_PIPE_PLUGIN_STATE =
"尝试删除 pipe plugin [{}] 时发生可重试错误，状态：{}";
- public static final String RETRIEVABLE_ERROR_TRYING_TO_EXECUTE_PLAN_STATE =
- "尝试执行 plan {} 时发生可重试错误，状态：{}";
public static final String RETRIEVABLE_ERROR_TRYING_TO_REMOVE_AINODE_STATE =
"尝试移除 AINode [{}] 时发生可重试错误，状态 [{}]";
public static final String ROLLBACK_CREATETABLE_COSTS_MS = "Rollback
CreateTable-{} costs {}ms.";
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java
index 2946884e373..0749ba5f9b0 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java
@@ -138,4 +138,7 @@ public enum CnToDnAsyncRequestType {
// audit log and event write-back
INSERT_RECORD,
ENABLE_SEPARATION_OF_ADMIN_POWERS,
+
+ // authority
+ INVALIDATE_PERMISSION_CACHE,
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java
index 4f66330f6ec..7ca698afa2c 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java
@@ -82,6 +82,7 @@ import
org.apache.iotdb.mpp.rpc.thrift.TInactiveTriggerInstanceReq;
import org.apache.iotdb.mpp.rpc.thrift.TInvalidateCacheReq;
import org.apache.iotdb.mpp.rpc.thrift.TInvalidateColumnCacheReq;
import org.apache.iotdb.mpp.rpc.thrift.TInvalidateMatchedSchemaCacheReq;
+import org.apache.iotdb.mpp.rpc.thrift.TInvalidatePermissionCacheReq;
import org.apache.iotdb.mpp.rpc.thrift.TInvalidateTableCacheReq;
import org.apache.iotdb.mpp.rpc.thrift.TKillQueryInstanceReq;
import org.apache.iotdb.mpp.rpc.thrift.TNotifyRegionMigrationReq;
@@ -527,6 +528,11 @@ public class CnToDnInternalServiceAsyncRequestManager
CnToDnAsyncRequestType.GET_BUILTIN_SERVICE,
(req, client, handler) ->
client.getBuiltInService((GetBuiltInExternalServiceRPCHandler)
handler));
+ actionMapBuilder.put(
+ CnToDnAsyncRequestType.INVALIDATE_PERMISSION_CACHE,
+ (req, client, handler) ->
+ client.invalidatePermissionCache(
+ (TInvalidatePermissionCacheReq) req,
(DataNodeTSStatusRPCHandler) handler));
}
@Override
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedure.java
index 7ae95858ac6..e3ff2c08188 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedure.java
@@ -20,21 +20,25 @@
package org.apache.iotdb.confignode.procedure.impl.sync;
import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.conf.CommonConfig;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
-import org.apache.iotdb.confignode.client.sync.CnToDnSyncRequestType;
-import org.apache.iotdb.confignode.client.sync.SyncDataNodeClientPool;
+import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType;
+import
org.apache.iotdb.confignode.client.async.CnToDnInternalServiceAsyncRequestManager;
+import
org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestContext;
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
import org.apache.iotdb.confignode.consensus.request.write.auth.AuthorPlan;
import
org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan;
import org.apache.iotdb.confignode.i18n.ProcedureMessages;
+import org.apache.iotdb.confignode.manager.lease.ClusterCachePropagator;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.impl.node.AbstractNodeProcedure;
+import org.apache.iotdb.confignode.procedure.impl.schema.SchemaUtils;
import
org.apache.iotdb.confignode.procedure.state.auth.AuthOperationProcedureState;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
import org.apache.iotdb.consensus.exception.ConsensusException;
@@ -50,8 +54,8 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
-import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import static
org.apache.iotdb.confignode.procedure.state.auth.AuthOperationProcedureState.DATANODE_AUTHCACHE_INVALIDING;
@@ -68,7 +72,6 @@ public class AuthOperationProcedure extends
AbstractNodeProcedure<AuthOperationP
private static final String CONSENSUS_WRITE_ERROR =
ProcedureMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE;
- private static final int RETRY_THRESHOLD = 2;
private static final CommonConfig commonConfig =
CommonDescriptor.getInstance().getConfig();
private final List<Pair<TDataNodeConfiguration, Long>> dataNodesToInvalid =
new ArrayList<>();
@@ -97,54 +100,40 @@ public class AuthOperationProcedure extends
AbstractNodeProcedure<AuthOperationP
writePlan(env);
return Flow.HAS_MORE_STATE;
case DATANODE_AUTHCACHE_INVALIDING:
- TInvalidatePermissionCacheReq req = new
TInvalidatePermissionCacheReq();
- TSStatus status;
- req.setUsername(user);
- req.setRoleName(role);
+ TInvalidatePermissionCacheReq req = new
TInvalidatePermissionCacheReq(user, role);
if (plan.getAuthorType() == ConfigPhysicalPlanType.AccountUnlock
|| plan.getAuthorType() ==
ConfigPhysicalPlanType.RAccountUnlock) {
// For account unlock, role carries the optional login address.
req.setNeedDisconnect(true);
}
- Iterator<Pair<TDataNodeConfiguration, Long>> it =
dataNodesToInvalid.iterator();
- while (it.hasNext()) {
- Pair<TDataNodeConfiguration, Long> pair = it.next();
- if (pair.getRight() + this.timeoutMS < System.currentTimeMillis())
{
- it.remove();
- continue;
- }
- status =
- (TSStatus)
- SyncDataNodeClientPool.getInstance()
- .sendSyncRequestToDataNodeWithRetry(
- pair.getLeft().getLocation().getInternalEndPoint(),
- req,
- CnToDnSyncRequestType.INVALIDATE_PERMISSION_CACHE);
- if (status.getCode() ==
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- it.remove();
- }
- }
- if (dataNodesToInvalid.isEmpty()) {
+ final boolean proceeded =
+ new
ClusterCachePropagator(SchemaUtils.filterFencedDataNode(env.getConfigManager()))
+ .propagate(targets -> broadcastAuthorityCache(req, targets));
+
+ if (proceeded) {
LOGGER.info(ProcedureMessages.AUTH_PROCEDURE_CLEAN_DATANODE_CACHE_SUCCESSFULLY);
return Flow.NO_MORE_STATE;
- } else {
-
setNextState(AuthOperationProcedureState.DATANODE_AUTHCACHE_INVALIDING);
}
- break;
+
LOGGER.warn(ProcedureMessages.AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED);
+ setFailure(
+ new ProcedureException(
+ new IoTDBException(
+
ProcedureMessages.AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED,
+ TSStatusCode.AUTH_OPERATE_EXCEPTION.getStatusCode())));
+ return Flow.NO_MORE_STATE;
}
} catch (Exception e) {
if (isRollbackSupported(state)) {
LOGGER.error(ProcedureMessages.FAIL_WHEN_EXECUTE, plan);
setFailure(new ProcedureException(e));
} else {
- LOGGER.error(
- ProcedureMessages.RETRIEVABLE_ERROR_TRYING_TO_EXECUTE_PLAN_STATE,
plan, state, e);
- if (getCycles() > RETRY_THRESHOLD) {
- setFailure(
- new ProcedureException(
- String.format(
- ProcedureMessages.FAIL_TO_EXECUTE_PLAN_AT_STATE,
plan.toString(), state)));
- }
+
LOGGER.warn(ProcedureMessages.AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED, e);
+ setFailure(
+ new ProcedureException(
+ new IoTDBException(
+ ProcedureMessages.AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED,
+ e,
+ TSStatusCode.AUTH_OPERATE_EXCEPTION.getStatusCode())));
}
}
return Flow.HAS_MORE_STATE;
@@ -164,19 +153,26 @@ public class AuthOperationProcedure extends
AbstractNodeProcedure<AuthOperationP
}
if (res.code == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
setNextState(DATANODE_AUTHCACHE_INVALIDING);
- for (TDataNodeConfiguration item : datanodes) {
- this.dataNodesToInvalid.add(new Pair<>(item,
System.currentTimeMillis()));
- }
- LOGGER.info(
- ProcedureMessages.EXECUTE_AUTH_PLAN_SUCCESS_TO_INVALIDATE_DATANODES,
- plan,
- dataNodesToInvalid);
+
LOGGER.info(ProcedureMessages.EXECUTE_AUTH_PLAN_SUCCESS_TO_INVALIDATE_DATANODES,
plan);
} else {
LOGGER.info(ProcedureMessages.FAILED_TO_EXECUTE_PLAN_BECAUSE, plan,
res.message);
setFailure(new ProcedureException(new IoTDBException(res)));
}
}
+ private static Map<Integer, TSStatus> broadcastAuthorityCache(
+ final TInvalidatePermissionCacheReq req, final Map<Integer,
TDataNodeLocation> targets) {
+ final DataNodeAsyncRequestContext<TInvalidatePermissionCacheReq, TSStatus>
clientHandler =
+ new DataNodeAsyncRequestContext<>(
+ CnToDnAsyncRequestType.INVALIDATE_PERMISSION_CACHE, req, targets);
+ CnToDnInternalServiceAsyncRequestManager.getInstance()
+ .sendAsyncRequest(
+ clientHandler,
+ ClusterCachePropagator.BROADCAST_RPC_RETRY,
+ ClusterCachePropagator.BROADCAST_RPC_TIMEOUT_MS);
+ return clientHandler.getResponseMap();
+ }
+
@Override
protected boolean isRollbackSupported(AuthOperationProcedureState state) {
return state == AuthOperationProcedureState.INIT;
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
index 0853e59b9fa..080fabdc546 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
@@ -949,12 +949,6 @@ public final class DataNodeMiscMessages {
public static final String CACHE_ROLE_PATH_PRIVILEGES_ERROR =
"cache role's path privileges error";
- //
---------------------------------------------------------------------------
- // auth – BasicAuthorityCache
- //
---------------------------------------------------------------------------
- public static final String DATANODE_CACHE_INIT_FAILED =
- "datanode cache initialization failed";
-
//
---------------------------------------------------------------------------
// trigger – TriggerExecutor
//
---------------------------------------------------------------------------
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
index 1abc4bb25dc..346dc8dafc1 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
@@ -947,11 +947,6 @@ public final class DataNodeMiscMessages {
public static final String CACHE_ROLE_PATH_PRIVILEGES_ERROR =
"缓存角色路径权限时发生错误";
- //
---------------------------------------------------------------------------
- // auth – BasicAuthorityCache
- //
---------------------------------------------------------------------------
- public static final String DATANODE_CACHE_INIT_FAILED =
- "DataNode 缓存初始化失败";
//
---------------------------------------------------------------------------
// trigger – TriggerExecutor
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
index fe8748ba4c3..0593db10840 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
@@ -136,6 +136,7 @@ public class AuthorityChecker {
}
public static void invalidateAllCache() {
+ PipeInsertionDataNodeListener.getInstance().invalidateAllCache();
authorityFetcher.get().getAuthorCache().invalidAllCache();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/BasicAuthorityCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/BasicAuthorityCache.java
index 1056475f5c0..83965effec0 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/BasicAuthorityCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/BasicAuthorityCache.java
@@ -22,7 +22,6 @@ package org.apache.iotdb.db.auth;
import org.apache.iotdb.commons.auth.entity.Role;
import org.apache.iotdb.commons.auth.entity.User;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
@@ -90,19 +89,11 @@ public class BasicAuthorityCache implements IAuthorCache {
}
userCache.invalidate(userName);
}
- if (userCache.getIfPresent(userName) != null) {
- LOGGER.error(DataNodeMiscMessages.DATANODE_CACHE_INIT_FAILED);
- return false;
- }
}
if (roleName != null) {
if (roleCache.getIfPresent(roleName) != null) {
roleCache.invalidate(roleName);
}
- if (roleCache.getIfPresent(roleName) != null) {
- LOGGER.error(DataNodeMiscMessages.DATANODE_CACHE_INIT_FAILED);
- return false;
- }
}
return true;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
index 5679a7ce6c9..1dd61c5b609 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
@@ -34,6 +34,7 @@ import org.apache.iotdb.commons.consensus.ConfigRegionId;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.exception.MetadataLeaseFencedException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.security.encrypt.AsymmetricEncrypt;
@@ -81,8 +82,6 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
private static final Logger LOGGER =
LoggerFactory.getLogger(ClusterAuthorityFetcher.class);
private static final CommonConfig CONFIG =
CommonDescriptor.getInstance().getConfig();
private final IAuthorCache iAuthorCache;
- private boolean cacheOutDate = false;
- private long heartBeatTimeStamp = 0;
private boolean acceptCache = true;
@@ -127,7 +126,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
@Override
public TSStatus checkUserSysPrivilege(String username, PrivilegeType
permission) {
- checkCacheAvailable();
+ failIfMetadataLeaseFenced();
return checkPrivilege(
username,
new PrivilegeUnion(permission, false),
@@ -139,7 +138,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
@Override
public Collection<PrivilegeType> checkUserSysPrivileges(
String username, Collection<PrivilegeType> permissions) {
- checkCacheAvailable();
+ failIfMetadataLeaseFenced();
Set<PrivilegeType> missingPrivileges = new HashSet<>();
for (PrivilegeType permission : permissions) {
TSStatus status =
@@ -158,7 +157,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
@Override
public TSStatus checkUserSysPrivilegesGrantOpt(String username,
PrivilegeType permission) {
- checkCacheAvailable();
+ failIfMetadataLeaseFenced();
return checkPrivilege(
username,
new PrivilegeUnion(permission, true),
@@ -174,7 +173,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
if (username.equals(AuthorityChecker.INTERNAL_AUDIT_USER)) {
return posList;
}
- checkCacheAvailable();
+ failIfMetadataLeaseFenced();
User user = getUser(username, true);
if (user.isOpenIdUser()) {
return posList;
@@ -206,6 +205,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
@Override
public TSStatus checkUserPathPrivilegesGrantOpt(
String username, List<? extends PartialPath> paths, PrivilegeType
permission) {
+ failIfMetadataLeaseFenced();
User user = iAuthorCache.getUserCache(username);
if (user != null) {
if (user.isOpenIdUser()) {
@@ -249,7 +249,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
@Override
public TSStatus checkUserDBPrivileges(
String username, String database, PrivilegeType permission) {
- checkCacheAvailable();
+ failIfMetadataLeaseFenced();
return checkPrivilege(
username,
new PrivilegeUnion(database, permission),
@@ -262,7 +262,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
@Override
public TSStatus checkUserDBPrivilegesGrantOpt(
String username, String database, PrivilegeType permission) {
- checkCacheAvailable();
+ failIfMetadataLeaseFenced();
return checkPrivilege(
username,
new PrivilegeUnion(database, permission, true),
@@ -276,7 +276,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
@Override
public TSStatus checkUserTBPrivileges(
String username, String database, String table, PrivilegeType
permission) {
- checkCacheAvailable();
+ failIfMetadataLeaseFenced();
return checkPrivilege(
username,
new PrivilegeUnion(database, table, permission),
@@ -292,7 +292,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
@Override
public TSStatus checkUserTBPrivilegesGrantOpt(
String username, String database, String table, PrivilegeType
permission) {
- checkCacheAvailable();
+ failIfMetadataLeaseFenced();
return checkPrivilege(
username,
new PrivilegeUnion(database, table, permission, true),
@@ -307,7 +307,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
@Override
public TSStatus checkUserAnyScopePrivilegeGrantOption(String username,
PrivilegeType permission) {
- checkCacheAvailable();
+ failIfMetadataLeaseFenced();
return checkPrivilege(
username,
new PrivilegeUnion(permission, false, true),
@@ -319,7 +319,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
/** -- check database/table visible -- * */
@Override
public TSStatus checkDBVisible(String username, String database) {
- checkCacheAvailable();
+ failIfMetadataLeaseFenced();
return checkPrivilege(
username,
new PrivilegeUnion(database, null, false),
@@ -330,7 +330,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
@Override
public TSStatus checkTBVisible(String username, String database, String
table) {
- checkCacheAvailable();
+ failIfMetadataLeaseFenced();
return checkPrivilege(
username,
new PrivilegeUnion(database, table, null, false),
@@ -343,6 +343,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
@Override
public PathPatternTree getAuthorizedPatternTree(String username,
PrivilegeType permission)
throws AuthException {
+ failIfMetadataLeaseFenced();
PathPatternTree patternTree = new PathPatternTree();
User user = iAuthorCache.getUserCache(username);
if (user != null) {
@@ -389,6 +390,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
private SettableFuture<ConfigTaskResult> operatePermissionInternal(
Object plan, boolean isRelational) {
+ failIfMetadataLeaseFenced();
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
try (ConfigNodeClient configNodeClient =
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
@@ -463,6 +465,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
private SettableFuture<ConfigTaskResult> queryPermissionInternal(
Object plan, boolean isRelational) {
+ failIfMetadataLeaseFenced();
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
TAuthorizerResp authorizerResp = new TAuthorizerResp();
try (ConfigNodeClient configNodeClient =
@@ -506,35 +509,10 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
return iAuthorCache;
}
- @Override
- public void refreshToken() {
- long currentTime = System.currentTimeMillis();
- if (heartBeatTimeStamp == 0) {
- heartBeatTimeStamp = currentTime;
- return;
- }
- if (currentTime - heartBeatTimeStamp > CONFIG.getDatanodeTokenTimeoutMS())
{
- cacheOutDate = true;
- }
- heartBeatTimeStamp = currentTime;
- }
-
- // Package-private for testing (ClusterAuthorityFetcherLeaseTest).
- void checkCacheAvailable() {
- // cacheOutDate is set by refreshToken() only when a heartbeat finally
arrives after a long gap,
- // so it cannot catch an *ongoing* ConfigNode partition (no heartbeat
arrives, refreshToken() is
- // never called). isFenced() is evaluated on this DataNode's own clock and
fires without any
- // heartbeat: while fenced we drop the permission cache and force a
re-fetch from the
- // ConfigNode,
- // which fails closed while partitioned, so a missed REVOKE cannot keep
authorizing a privilege.
- if (cacheOutDate || isMetadataLeaseFenced()) {
- iAuthorCache.invalidAllCache();
- }
- cacheOutDate = false;
- }
-
- boolean isMetadataLeaseFenced() {
- return MetadataLeaseManager.getInstance().isFenced();
+ void failIfMetadataLeaseFenced() {
+ MetadataLeaseManager.getInstance()
+ .failIfMetadataLeaseFenced(
+
MetadataLeaseFencedException.LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
}
@TestOnly
@@ -545,7 +523,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
@Override
public TSStatus checkUser(
final String username, final String password, final boolean
useEncryptedPassword) {
- checkCacheAvailable();
+ failIfMetadataLeaseFenced();
final User user = iAuthorCache.getUserCache(username);
if (user != null) {
if (user.isOpenIdUser()) {
@@ -595,7 +573,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
@Override
public User getUser(String userName, final boolean force) {
- checkCacheAvailable();
+ failIfMetadataLeaseFenced();
User user = iAuthorCache.getUserCache(userName);
if (user != null) {
return user;
@@ -628,7 +606,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
@Override
public boolean checkRole(String userName, String roleName) {
- checkCacheAvailable();
+ failIfMetadataLeaseFenced();
User user = iAuthorCache.getUserCache(userName);
if (user != null) {
return user.isOpenIdUser() || user.getRoleSet().contains(roleName);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/IAuthorityFetcher.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/IAuthorityFetcher.java
index b14d5c599c1..8193456eb15 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/IAuthorityFetcher.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/IAuthorityFetcher.java
@@ -84,7 +84,5 @@ public interface IAuthorityFetcher {
IAuthorCache getAuthorCache();
- void refreshToken();
-
User getUser(String username, final boolean force);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index a06d10413fd..84ad19df0ad 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -2352,7 +2352,6 @@ public class DataNodeInternalRPCServiceImpl implements
IDataNodeRPCService.Iface
.forEach((key, value) ->
regionRawDataSize.put(Integer.parseInt(key), value.getLeft()));
resp.setDataRegionRawDataSize(regionRawDataSize);
}
- AuthorityChecker.getAuthorityFetcher().refreshToken();
resp.setHeartbeatTimestamp(req.getHeartbeatTimestamp());
resp.setStatus(commonConfig.getNodeStatus().getStatus());
// Advertise that this DataNode supports metadata-lease self-fencing, so
the ConfigNode may
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseManager.java
index da198225243..7f7365a1550 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseManager.java
@@ -24,6 +24,7 @@ import
org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
import org.apache.iotdb.commons.exception.MetadataLeaseFencedException;
import
org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.DataNodeSchemaMessages;
import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
@@ -106,7 +107,8 @@ public class MetadataLeaseManager {
return Arrays.asList(
() -> ClusterPartitionFetcher.getInstance().invalidAllCache(),
() -> DataNodeTableCache.getInstance().invalidateAll(),
- () -> TreeDeviceSchemaCacheManager.getInstance().cleanUp());
+ () -> TreeDeviceSchemaCacheManager.getInstance().cleanUp(),
+ AuthorityChecker::invalidateAllCache);
}
private static List<MetadataAction> defaultPullMetaList() {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcherLeaseTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcherLeaseTest.java
index 6933b11a055..fb0f04f989d 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcherLeaseTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcherLeaseTest.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.auth;
import org.apache.iotdb.commons.auth.entity.User;
+import org.apache.iotdb.commons.exception.MetadataLeaseFencedException;
import org.apache.iotdb.db.schemaengine.lease.MetadataLeaseManager;
import org.apache.iotdb.db.schemaengine.lease.MetadataLeaseTestUtils;
@@ -41,19 +42,17 @@ public class ClusterAuthorityFetcherLeaseTest {
}
@Test
- public void fencedLeaseDropsPermissionCache() {
+ public void fencedLeaseThrowsException() {
final ClusterAuthorityFetcher fetcher =
new TestingClusterAuthorityFetcher(new BasicAuthorityCache(),
leaseManager);
- final User user = new User("user_fenced", "password");
- fetcher.getAuthorCache().putUserCache(user.getName(), user);
-
Assert.assertNotNull(fetcher.getAuthorCache().getUserCache(user.getName()));
clock.addMillis(T_FENCE_MS + 1);
- fetcher.checkCacheAvailable();
-
- Assert.assertNull(
- "a fenced DataNode must drop its permission cache so a missed REVOKE
cannot keep authorizing",
- fetcher.getAuthorCache().getUserCache(user.getName()));
+ try {
+ fetcher.failIfMetadataLeaseFenced();
+ Assert.fail("Expected MetadataLeaseFencedException");
+ } catch (MetadataLeaseFencedException e) {
+ // Expected.
+ }
}
@Test
@@ -66,7 +65,7 @@ public class ClusterAuthorityFetcherLeaseTest {
// An active lease (a ConfigNode heartbeat was just received) must not
needlessly drop the
// cache.
clock.addMillis(1_000L);
- fetcher.checkCacheAvailable();
+ fetcher.failIfMetadataLeaseFenced();
Assert.assertNotNull(
"an active lease must not needlessly drop the permission cache",
@@ -96,8 +95,9 @@ public class ClusterAuthorityFetcherLeaseTest {
}
@Override
- boolean isMetadataLeaseFenced() {
- return MetadataLeaseTestUtils.isFenced(leaseManager);
+ void failIfMetadataLeaseFenced() {
+ MetadataLeaseTestUtils.failIfMetadataLeaseFenced(
+ leaseManager,
MetadataLeaseFencedException.LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
}
}
}
