AnonHxy commented on code in PR #21030:
URL: https://github.com/apache/pulsar/pull/21030#discussion_r1299696101
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java:
##########
@@ -420,11 +427,26 @@ private void cleanCacheAndCloseReader(@Nonnull
NamespaceName namespace, boolean
});
}
+ private void completeResults(TopicName topicName) {
+ if (results.get(topicName) != null) {
+ BlockingDeque<CompletableFuture> futures = results.get(topicName);
+ CompletableFuture<Void> future;
+ while (true) {
+ future = futures.poll();
+ if (future == null) {
+ break;
+ }
+ future.complete(null);
+ }
+ }
+ }
+
private void readMorePolicies(SystemTopicClient.Reader<PulsarEvent>
reader) {
reader.readNextAsync()
.thenAccept(msg -> {
refreshTopicPoliciesCache(msg);
notifyListener(msg);
+
completeResults(TopicName.get(TopicName.get(msg.getKey()).getPartitionedTopicName()));
Review Comment:
> This solution fix the synchronize case.
Maybe the synchronized case is not fixed yet :).
For example, the result could still be unexpected if we call the following
three synchronized method. The `messageTTL` could be the old value after we
updated `XXX`:
```
admin.topics().setRetention
admin.topics().setMessageTTL
admin.topics().setXXX
```
--
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]
