Merge branch 'master' of github.com:apache/pulsar into issue17354

apache · Sep 4, 2022 · 74f23b4 · 74f23b4
2 parents cf12237 + d139d88
commit 74f23b4
Show file tree

Hide file tree

Showing 4 changed files with 56 additions and 21 deletions.
diff --git a/README.md b/README.md
@@ -97,7 +97,7 @@ components in the Pulsar ecosystem, including connectors, adapters, and other la
 
 ## Pulsar Runtime Java Version Recommendation
 
-- pulsar master branch
+- pulsar ver > 2.10 and master branch
 
 | Pulsar Components | Java Version  |
 | ----------------- | :-----------: |
@@ -115,7 +115,7 @@ components in the Pulsar ecosystem, including connectors, adapters, and other la
 | CLI               |   8 or 11    |
 | Java Client       |   8 or 11    |
 
-- pulsar ver 2.8 <
+- pulsar ver < 2.8
 
 | Pulsar Components | Java Version |
 | ----------------- | :----------: |

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/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java b/...broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
@@ -198,9 +198,7 @@ private CompletableFuture<StoredSchema> getSchema(String schemaId) {
             if (log.isDebugEnabled()) {
                 log.debug("[{}] Fetching schema from store", schemaId);
             }
-            CompletableFuture<StoredSchema> future = new CompletableFuture<>();
-
-            getSchemaLocator(getSchemaPath(schemaId)).thenCompose(locator -> {
+            return getSchemaLocator(getSchemaPath(schemaId)).thenCompose(locator -> {
                 if (log.isDebugEnabled()) {
                     log.debug("[{}] Got schema locator {}", schemaId, locator);
                 }
@@ -213,22 +211,12 @@ private CompletableFuture<StoredSchema> getSchema(String schemaId) {
                 return readSchemaEntry(schemaLocator.getInfo().getPosition())
                         .thenApply(entry -> new StoredSchema(entry.getSchemaData().toByteArray(),
                                 new LongSchemaVersion(schemaLocator.getInfo().getVersion())));
-            }).handleAsync((res, ex) -> {
-                if (log.isDebugEnabled()) {
-                    log.debug("[{}] Get operation completed. res={} -- ex={}", schemaId, res, ex);
-                }
-
-                // Cleanup the pending ops from the map
-                readSchemaOperations.remove(schemaId, future);
-                if (ex != null) {
-                    future.completeExceptionally(ex);
-                } else {
-                    future.complete(res);
-                }
-                return null;
             });
-
-            return future;
+        }).whenComplete((res, ex) -> {
+            if (log.isDebugEnabled()) {
+                log.debug("[{}] Get operation completed. res={} -- ex={}", schemaId, res, ex);
+            }
+            readSchemaOperations.remove(schemaId);
         });
     }
 

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));
+        });
+    }
 }