lhotari commented on code in PR #25070:
URL: https://github.com/apache/pulsar/pull/25070#discussion_r2682090119
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java:
##########
@@ -287,9 +422,70 @@ public void deleteTopicListWatcher(Long watcherId) {
* @param newTopics topics names added(contains the partition suffix).
*/
public void sendTopicListUpdate(long watcherId, String topicsHash,
List<String> deletedTopics,
- List<String> newTopics) {
- connection.getCommandSender().sendWatchTopicListUpdate(watcherId,
newTopics, deletedTopics, topicsHash);
+ List<String> newTopics, Runnable
completionCallback) {
+ performOperationWithPermitAcquiringRetries(watcherId, "topic list
update", permitAcquireErrorHandler ->
+ () -> connection.getCommandSender()
+ .sendWatchTopicListUpdate(watcherId, newTopics,
deletedTopics, topicsHash,
+ permitAcquireErrorHandler)
+ .whenComplete((__, t) -> {
+ if (t != null) {
+ // this is an unexpected case
+ log.warn("[{}] Failed to send topic list
update for watcherId={}. Watcher will be in "
+ + "inconsistent state.", connection,
watcherId, t);
+ }
+ completionCallback.run();
Review Comment:
Yes there are gaps. I haven't yet created a test case to clarify and
validate the behavior. The `permitAcquireErrorHandler` breaks the
completablefuture chain and I'll have to revisit that part. Instead of
`Consumer<Throwable> permitAcquireErrorHandler`, I think it should be
`Function<Throwable, CompletableFuture<Void>> permitAcquireErrorHandler` so
that the chain doesn't break.
Regarding the update events, there is no guarantee that the MetadataStore
implementation will send all updates. For example, with ZooKeeper, this can
happen. There's issue #25020 about this. I've discussed this with Matteo and
the resolution would be to keep on polling even when topic list watchers are
used. Another improvement would be to handle the session events immediately
when they occur.
--
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]
