nodece commented on code in PR #24658:
URL: https://github.com/apache/pulsar/pull/24658#discussion_r2296972008


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java:
##########
@@ -526,32 +533,39 @@ private void 
initPolicesCache(SystemTopicClient.Reader<PulsarEvent> reader, Comp
         });
     }
 
-    private void cleanCacheAndCloseReader(@NonNull NamespaceName namespace, 
boolean cleanOwnedBundlesCount) {
-        cleanCacheAndCloseReader(namespace, cleanOwnedBundlesCount, false);
+    private void cleanCache(@NonNull NamespaceName namespace, boolean 
cleanOwnedBundlesCount,
+                            boolean closeReaderCache) {
+        cleanCache(namespace, cleanOwnedBundlesCount, false, closeReaderCache);
     }
 
-    private void cleanCacheAndCloseReader(@NonNull NamespaceName namespace, 
boolean cleanOwnedBundlesCount,
-                                          boolean cleanWriterCache) {
+    private void cleanCache(@NonNull NamespaceName namespace, boolean 
cleanOwnedBundlesCount,
+                            boolean cleanWriterCache, boolean 
cleanReaderCache) {
         if (cleanWriterCache) {
             writerCaches.synchronous().invalidate(namespace);
         }
-        CompletableFuture<SystemTopicClient.Reader<PulsarEvent>> readerFuture 
= readerCaches.remove(namespace);
 
         if (cleanOwnedBundlesCount) {
             ownedBundlesCountPerNamespace.remove(namespace);
         }
-        if (readerFuture != null && !readerFuture.isCompletedExceptionally()) {
-            readerFuture.thenCompose(SystemTopicClient.Reader::closeAsync)
-                    .exceptionally(ex -> {
-                        log.warn("[{}] Close change_event reader fail.", 
namespace, ex);
-                        return null;
-                    });
+
+        if (!cleanReaderCache) {
+            policyCacheInitMap.remove(namespace);

Review Comment:
   We can remove cache from `policyCacheInitMap`, and improve the reader 
creation in the `prepareInitPoliciesCacheAsync`:
   ```
   return policyCacheInitMap.computeIfAbsent(namespace, (k) -> {
       var readerCompletableFuture = readerCaches.compute(namespace, (__, v) -> 
{
           if (v == null || v.isCompletedExceptionally()) {
               CompletableFuture<SystemTopicClient.Reader<PulsarEvent>>
                       future = createSystemTopicClient(namespace);
               future.exceptionally(ex -> {
                   log.error("[{}] Failed to create reader on __change_events 
topic",
                           namespace, ex);
                   return null;
               });
               return future;
           }
           return v;
       });
       // todo
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Reply via email to