[fix][broker] fix broker may lost rack information (apache#23331)

Co-authored-by: fanjianye <[email protected]>
(cherry picked from commit a7d9d8f)

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java

@@ -181,8 +181,13 @@ private CompletableFuture<Versioned<Set<BookieId>>> getBookiesThenFreshCache(Str
     @Override
     public CompletableFuture<Void> watchWritableBookies(RegistrationListener registrationListener) {
         writableBookiesWatchers.add(registrationListener);
+        // trigger all listeners in writableBookiesWatchers one by one. It aims to keep a sync way
+        // to make sure the previous listener has finished when a new listener is register.
+        // Though it would bring duplicate trigger listener problem, but since watchWritableBookies
+        // is only executed when bookieClient construct, the duplicate problem is acceptable.
         return getWritableBookies()
-                .thenAcceptAsync(registrationListener::onBookiesChanged, executor);
+                .thenAcceptAsync(bookies ->
+                        writableBookiesWatchers.forEach(w -> w.onBookiesChanged(bookies)), executor);
     }
 
     @Override
@@ -193,8 +198,13 @@ public void unwatchWritableBookies(RegistrationListener registrationListener) {
     @Override
     public CompletableFuture<Void> watchReadOnlyBookies(RegistrationListener registrationListener) {
         readOnlyBookiesWatchers.add(registrationListener);
+        // trigger all listeners in readOnlyBookiesWatchers one by one. It aims to keep a sync way
+        // to make sure the previous listener has finished when a new listener is register.
+        // Though it would bring duplicate trigger listener problem, but since watchReadOnlyBookies
+        // is only executed when bookieClient construct, the duplicate problem is acceptable.
         return getReadOnlyBookies()
-                .thenAcceptAsync(registrationListener::onBookiesChanged, executor);
+                .thenAcceptAsync(bookies ->
+                        readOnlyBookiesWatchers.forEach(w -> w.onBookiesChanged(bookies)), executor);
     }
 
     @Override