Avoid CN OOM by Pulling User/Roles to DN When Cache Misses. (apache#16888)

* Fix OOM problem for PIPE.

* Fix IT.

* Fix potential NPE.

---------

Co-authored-by: shiwenyan <[email protected]>

Author: wenyanshi-123

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java

@@ -167,37 +167,37 @@ public TSStatus checkUserSysPrivilegesGrantOpt(String username, PrivilegeType pe
   @Override
   public List<Integer> checkUserPathPrivileges(
       String username, List<? extends PartialPath> allPath, PrivilegeType permission) {
-    checkCacheAvailable();
     List<Integer> posList = new ArrayList<>();
-    User user = iAuthorCache.getUserCache(username);
-    if (user != null) {
-      if (user.isOpenIdUser()) {
-        return posList;
-      }
-      int pos = 0;
-      for (PartialPath path : allPath) {
-        if (!user.checkPathPrivilege(path, permission)) {
-          boolean checkFromRole = false;
-          for (String rolename : user.getRoleSet()) {
-            Role cachedRole = iAuthorCache.getRoleCache(rolename);
-            if (cachedRole == null) {
-              return checkPathFromConfigNode(username, allPath, permission);
-            }
-            if (cachedRole.checkPathPrivilege(path, permission)) {
-              checkFromRole = true;
-              break;
-            }
+    if (username.equals(AuthorityChecker.INTERNAL_AUDIT_USER)) {
+      return posList;
+    }
+    checkCacheAvailable();
+    User user = getUser(username);
+    if (user.isOpenIdUser()) {
+      return posList;
+    }
+    int pos = 0;
+    for (PartialPath path : allPath) {
+      if (!user.checkPathPrivilege(path, permission)) {
+        boolean checkFromRole = false;
+        for (String rolename : user.getRoleSet()) {
+          Role cachedRole = iAuthorCache.getRoleCache(rolename);
+          if (cachedRole == null) {
+            checkRoleFromConfigNode(username, rolename);
+            cachedRole = iAuthorCache.getRoleCache(rolename);
           }
-          if (!checkFromRole) {
-            posList.add(pos);
+          if (cachedRole != null && cachedRole.checkPathPrivilege(path, permission)) {
+            checkFromRole = true;
+            break;
           }
         }
-        pos++;
+        if (!checkFromRole) {
+          posList.add(pos);
+        }
       }
-      return posList;
-    } else {
-      return checkPathFromConfigNode(username, allPath, permission);
+      pos++;
     }
+    return posList;
   }
 
   @Override
@@ -642,15 +642,6 @@ private TPermissionInfoResp checkPrivilegeFromConfigNode(TCheckUserPrivilegesReq
     return permissionInfoResp;
   }
 
-  private List<Integer> checkPathFromConfigNode(
-      String username, List<? extends PartialPath> allPath, PrivilegeType permission) {
-    TCheckUserPrivilegesReq req =
-        new TCheckUserPrivilegesReq(
-            username, PrivilegeModelType.TREE.ordinal(), permission.ordinal(), false);
-    req.setPaths(AuthUtils.serializePartialPathList(allPath));
-    return checkPrivilegeFromConfigNode(req).getFailPos();
-  }
-
   private boolean checkRoleFromConfigNode(String username, String rolename) {
     TAuthorizerReq req = new TAuthorizerReq();
     // just reuse authorizer request. only need username and rolename field.