nodece commented on code in PR #24658:
URL: https://github.com/apache/pulsar/pull/24658#discussion_r2292631548
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java:
##########
@@ -526,27 +526,30 @@ 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 (cleanReaderCache) {
+ CompletableFuture<SystemTopicClient.Reader<PulsarEvent>>
readerFuture = readerCaches.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 (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;
- });
- }
policyCacheInitMap.compute(namespace, (k, v) -> {
Review Comment:
@dao-jun Please move this logic inside the `cleanReaderCache` condition.
Otherwise, there's a risk of partial policy loss: You're always cleaning up the
`policiesCache`, and when the reader is not closed, the cache state becomes
inconsistent.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
