Remove readerCaches and close reader when exception occurs in SystemT…

…opicBasedTopicPoliciesService. (apache#12873)
fxbing · Dec 19, 2021 · d0b0be3 · d0b0be3
1 parent 6d9bb3a
commit d0b0be3
Showing 1 changed file with 2 additions and 0 deletions.
diff --git a/.../src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/.../src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
@@ -228,6 +228,8 @@ private void prepareInitPoliciesCache(NamespaceName namespace, CompletableFuture
                 if (ex != null) {
                     log.error("[{}] Failed to create reader on __change_events topic", namespace, ex);
                     result.completeExceptionally(ex);
+                    readerCaches.remove(namespace);
+                    reader.closeAsync();
                 } else {
                     initPolicesCache(reader, result);
                     result.thenRun(() -> readMorePolicies(reader));