HBASE-29142 fix permission caches can be empty during ACL update

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

@@ -78,12 +78,17 @@ Set<T> get(String name) {
       }
     }
 
-    void clear() {
+    void cleanUp(ListMultimap<String, ? extends Permission> newPermissions) {
       synchronized (mutex) {
         for (Map.Entry<String, Set<T>> entry : cache.entrySet()) {
-          entry.getValue().clear();
+          String key = entry.getKey();
+          Set<T> value = entry.getValue();
+          if (newPermissions.containsKey(key)) {
+            value.removeIf(t -> !newPermissions.get(key).contains(t));
+          } else {
+            value.clear();
+          }
         }
-        cache.clear();
       }
     }
   }
@@ -163,7 +168,6 @@ public void refreshNamespaceCacheFromWritable(String namespace, byte[] data) thr
    * @param globalPerms new global permissions
    */
   private void updateGlobalCache(ListMultimap<String, Permission> globalPerms) {
-    globalCache.clear();
     for (String name : globalPerms.keySet()) {
       for (Permission permission : globalPerms.get(name)) {
         // Before 2.2, the global permission which storage in zk is not right. It was saved as a
@@ -175,6 +179,12 @@ private void updateGlobalCache(ListMultimap<String, Permission> globalPerms) {
         }
       }
     }
+    for (String name : globalCache.keySet()) {
+      if (!globalPerms.containsKey(name)) {
+        globalCache.remove(name);
+      }
+    }
+
     mtime.incrementAndGet();
   }
 
@@ -186,7 +196,6 @@ private void updateGlobalCache(ListMultimap<String, Permission> globalPerms) {
   private void updateTableCache(TableName table, ListMultimap<String, Permission> tablePerms) {
     PermissionCache<TablePermission> cacheToUpdate =
       tableCache.getOrDefault(table, new PermissionCache<>());
-    clearCache(cacheToUpdate);
     updateCache(tablePerms, cacheToUpdate);
     tableCache.put(table, cacheToUpdate);
     mtime.incrementAndGet();
@@ -200,16 +209,11 @@ private void updateTableCache(TableName table, ListMultimap<String, Permission>
   private void updateNamespaceCache(String namespace, ListMultimap<String, Permission> nsPerms) {
     PermissionCache<NamespacePermission> cacheToUpdate =
       namespaceCache.getOrDefault(namespace, new PermissionCache<>());
-    clearCache(cacheToUpdate);
     updateCache(nsPerms, cacheToUpdate);
     namespaceCache.put(namespace, cacheToUpdate);
     mtime.incrementAndGet();
   }
 
-  private void clearCache(PermissionCache cacheToUpdate) {
-    cacheToUpdate.clear();
-  }
-
   @SuppressWarnings("unchecked")
   private void updateCache(ListMultimap<String, ? extends Permission> newPermissions,
     PermissionCache cacheToUpdate) {
@@ -218,6 +222,7 @@ private void updateCache(ListMultimap<String, ? extends Permission> newPermissio
         cacheToUpdate.put(name, permission);
       }
     }
+    cacheToUpdate.cleanUp(newPermissions);
   }
 
   /**

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

@@ -24,6 +24,7 @@
 import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
@@ -462,4 +463,99 @@ public void testAuthManager() throws Exception {
       }
     }
   }
+
+  @Test
+  public void testPermissionShouldAlwaysSucceedDuringUpdate()
+    throws IOException, InterruptedException {
+    Configuration conf = UTIL.getConfiguration();
+    AuthManager authManager = new AuthManager(conf);
+
+    User user = User.createUserForTesting(conf, "test-user", new String[] {});
+    AtomicBoolean authorizationSucceeded = new AtomicBoolean(true);
+
+    // test from fine-grained to coarse-grained permissions
+    // test table permission
+    byte[] tableAclData = createPermissionsData(conf, user, Permission.newBuilder(TEST_TABLE));
+    authManager.refreshTableCacheFromWritable(TEST_TABLE, tableAclData);
+    verifyAuthorization(() -> {
+      for (int i = 0; i < 1000; i++) {
+        try {
+          authManager.refreshTableCacheFromWritable(TEST_TABLE, tableAclData);
+        } catch (IOException e) {
+          throw new RuntimeException(e);
+        }
+      }
+    }, () -> {
+      for (int i = 0; i < 1000; i++) {
+        if (!authManager.authorizeUserTable(user, TEST_TABLE, Permission.Action.READ)) {
+          authorizationSucceeded.set(false);
+          break;
+        }
+      }
+    }, authorizationSucceeded);
+
+    // test namespace permission
+    byte[] nsAclData = createPermissionsData(conf, user, Permission.newBuilder(TEST_NAMESPACE));
+    authManager.refreshNamespaceCacheFromWritable(TEST_NAMESPACE, nsAclData);
+
+    verifyAuthorization(() -> {
+      for (int i = 0; i < 1000; i++) {
+        try {
+          authManager.refreshNamespaceCacheFromWritable(TEST_NAMESPACE, nsAclData);
+        } catch (IOException e) {
+          throw new RuntimeException(e);
+        }
+      }
+    }, () -> {
+      for (int i = 0; i < 1000; i++) {
+        if (!authManager.authorizeUserNamespace(user, TEST_NAMESPACE, Permission.Action.READ)) {
+          authorizationSucceeded.set(false);
+          break;
+        }
+      }
+    }, authorizationSucceeded);
+
+    // test global permission
+    byte[] globalAclData = createPermissionsData(conf, user, Permission.newBuilder());
+    authManager.refreshTableCacheFromWritable(PermissionStorage.ACL_TABLE_NAME, globalAclData);
+
+    verifyAuthorization(() -> {
+      for (int i = 0; i < 1000; i++) {
+        try {
+          authManager.refreshTableCacheFromWritable(PermissionStorage.ACL_TABLE_NAME,
+            globalAclData);
+        } catch (IOException e) {
+          throw new RuntimeException(e);
+        }
+      }
+    }, () -> {
+      for (int i = 0; i < 1000; i++) {
+        if (!authManager.authorizeUserGlobal(user, Permission.Action.READ)) {
+          authorizationSucceeded.set(false);
+          break;
+        }
+      }
+    }, authorizationSucceeded);
+  }
+
+  private static void verifyAuthorization(Runnable m1, Runnable m2,
+    AtomicBoolean authorizationSucceeded) throws InterruptedException {
+    Thread t1 = new Thread(m1);
+    Thread t2 = new Thread(m2);
+    t1.start();
+    t2.start();
+    t1.join();
+    t2.join();
+    assertTrue("Authorization should always succeed", authorizationSucceeded.get());
+  }
+
+  private static byte[] createPermissionsData(Configuration conf, User user,
+    Permission.Builder newBuilder) {
+    List<UserPermission> acl = new ArrayList<>(1);
+    acl.add(new UserPermission(user.getShortName(),
+      newBuilder.withActions(Permission.Action.READ, Permission.Action.WRITE).build()));
+    ListMultimap<String, UserPermission> permissions = ArrayListMultimap.create();
+    permissions.putAll(user.getShortName(), acl);
+    return PermissionStorage.writePermissionsAsBytes(permissions, conf);
+  }
 }