Remove readerCaches and close reader when exception occurs in SystemTopicBasedTopicPoliciesService.
@@ -228,6 +228,8 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
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));
