[improve] clean the empty topicAuthenticationMap in zk when revoke pe…

…rmission (#16815) Co-authored-by: fanjianye <fanjianye@bigo.sg>
apache · Sep 4, 2022 · d139d88 · d139d88
1 parent 0cbf7c3
commit d139d88
Show file tree

Hide file tree

Showing 2 changed files with 48 additions and 1 deletion.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -361,7 +361,13 @@ private CompletableFuture<Void> revokePermissionsAsync(String topicUri, String r
                     }
                     // Write the new policies to metadata store
                     return namespaceResources().setPoliciesAsync(namespaceName, p -> {
-                        p.auth_policies.getTopicAuthentication().get(topicUri).remove(role);
+                        p.auth_policies.getTopicAuthentication().computeIfPresent(topicUri, (k, roles) -> {
+                            roles.remove(role);
+                            if (roles.isEmpty()) {
+                                return null;
+                            }
+                            return roles;
+                        });
                         return p;
                     }).thenAccept(__ ->
                             log.info("[{}] Successfully revoke access for role {} - topic {}", clientAppId(), role,

diff --git a/...-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java b/...-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java
@@ -466,4 +466,45 @@ public void testTlsTransportWithAnyAuth(Supplier<String> url, Authentication aut
         @Cleanup
         Producer<byte[]> ignored = client.newProducer().topic(topicName).create();
     }
+
+    @Test
+    public void testCleanupEmptyTopicAuthenticationMap() throws Exception {
+        Map<String, String> authParams = new HashMap<>();
+        authParams.put("tlsCertFile", TLS_CLIENT_CERT_FILE_PATH);
+        authParams.put("tlsKeyFile", TLS_CLIENT_KEY_FILE_PATH);
+        Authentication authTls = new AuthenticationTls();
+        authTls.configure(authParams);
+        internalSetup(authTls);
+
+        admin.clusters().createCluster("test", ClusterData.builder().build());
+        admin.tenants().createTenant("p1",
+                new TenantInfoImpl(Collections.emptySet(), new HashSet<>(admin.clusters().getClusters())));
+        admin.namespaces().createNamespace("p1/ns1");
+
+        String topic = "persistent://p1/ns1/topic";
+        admin.topics().createNonPartitionedTopic(topic);
+        assertFalse(pulsar.getPulsarResources().getNamespaceResources().getPolicies(NamespaceName.get("p1/ns1"))
+                .get().auth_policies.getTopicAuthentication().containsKey(topic));
+
+        // grant permission
+        admin.topics().grantPermission(topic, "test-user-1", EnumSet.of(AuthAction.consume));
+        Awaitility.await().untilAsserted(() -> {
+            assertTrue(pulsar.getPulsarResources().getNamespaceResources().getPolicies(NamespaceName.get("p1/ns1"))
+                    .get().auth_policies.getTopicAuthentication().containsKey(topic));
+        });
+
+        // revoke permission
+        admin.topics().revokePermissions(topic, "test-user-1");
+        Awaitility.await().untilAsserted(() -> {
+            assertFalse(pulsar.getPulsarResources().getNamespaceResources().getPolicies(NamespaceName.get("p1/ns1"))
+                    .get().auth_policies.getTopicAuthentication().containsKey(topic));
+        });
+
+        // grant permission again
+        admin.topics().grantPermission(topic, "test-user-1", EnumSet.of(AuthAction.consume));
+        Awaitility.await().untilAsserted(() -> {
+            assertTrue(pulsar.getPulsarResources().getNamespaceResources().getPolicies(NamespaceName.get("p1/ns1"))
+                    .get().auth_policies.getTopicAuthentication().containsKey(topic));
+        });
+    }
 }