[fix][broker] Fix the broker registering might be blocked for long time

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java

@@ -83,13 +83,6 @@ public BrokerRegistryImpl(PulsarService pulsar) {
         this.brokerLookupDataMetadataCache = pulsar.getLocalMetadataStore().getMetadataCache(BrokerLookupData.class);
         this.scheduler = pulsar.getLoadManagerExecutor();
         this.listeners = new ArrayList<>();
-        // The registered node is an ephemeral node that could be deleted when the metadata store client's session
-        // is expired. In this case, we should register again.
-        this.listeners.add((broker, notificationType) -> {
-            if (notificationType == NotificationType.Deleted && getBrokerId().equals(broker)) {
-                registerAsync();
-            }
-        });
         this.brokerIdKeyPath = keyPath(pulsar.getBrokerId());
         this.brokerLookupData = new BrokerLookupData(
                 pulsar.getWebServiceAddress(),
@@ -223,11 +216,16 @@ private void handleMetadataStoreNotification(Notification t) {
             if (log.isDebugEnabled()) {
                 log.debug("Handle notification: [{}]", t);
             }
+            // The registered node is an ephemeral node that could be deleted when the metadata store client's session
+            // is expired. In this case, we should register again.
+            final var brokerId = t.getPath().substring(LOADBALANCE_BROKERS_ROOT.length() + 1);
+            if (t.getType() == NotificationType.Deleted && getBrokerId().equals(brokerId)) {
+                registerAsync();
+            }
             if (listeners.isEmpty()) {
                 return;
             }
             this.scheduler.submit(() -> {
-                String brokerId = t.getPath().substring(LOADBALANCE_BROKERS_ROOT.length() + 1);
                 for (BiConsumer<String, NotificationType> listener : listeners) {
                     listener.accept(brokerId, t.getType());
                 }

pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryIntegrationTest.java

@@ -37,7 +37,7 @@
 import org.testng.annotations.Test;
 
 @Slf4j
-@Test(groups = "flaky")
+@Test(groups = "broker")
 public class BrokerRegistryIntegrationTest {
 
     private static final String clusterName = "test";
@@ -63,13 +63,18 @@ protected void setup() throws Exception {
 
     @AfterClass(alwaysRun = true)
     protected void cleanup() throws Exception {
+        final var startMs = System.currentTimeMillis();
         if (pulsar != null) {
             pulsar.close();
         }
+        final var elapsedMs = System.currentTimeMillis() - startMs;
         bk.stop();
+        if (elapsedMs > 5000) {
+            throw new RuntimeException("Broker took " + elapsedMs + "ms to close");
+        }
     }
 
-    @Test(enabled = false)
+    @Test
     public void testRecoverFromNodeDeletion() throws Exception {
         // Simulate the case that the node was somehow deleted (e.g. by session timeout)
         Awaitility.await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> Assert.assertEquals(
@@ -88,7 +93,7 @@ public void testRecoverFromNodeDeletion() throws Exception {
         Assert.assertEquals(brokerRegistry.getAvailableBrokersAsync().get(), List.of(pulsar.getBrokerId()));
     }
 
-    @Test(enabled = false)
+    @Test
     public void testRegisterAgain() throws Exception {
         Awaitility.await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> Assert.assertEquals(
                 brokerRegistry.getAvailableBrokersAsync().join(), List.of(pulsar.getBrokerId())));
@@ -105,7 +110,7 @@ public void testRegisterAgain() throws Exception {
         });
     }
 
-    private ServiceConfiguration brokerConfig() {
+    protected ServiceConfiguration brokerConfig() {
         final var config = new ServiceConfiguration();
         config.setClusterName(clusterName);
         config.setAdvertisedAddress("localhost");

pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryMetadataStoreIntegrationTest.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.loadbalance.extensions;
+
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateMetadataStoreTableViewImpl;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class BrokerRegistryMetadataStoreIntegrationTest extends BrokerRegistryIntegrationTest {
+
+    @Override
+    protected ServiceConfiguration brokerConfig() {
+        final var config = super.brokerConfig();
+        config.setLoadManagerServiceUnitStateTableViewClassName(
+                ServiceUnitStateMetadataStoreTableViewImpl.class.getName());
+        return config;
+    }
+}

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TableViewTest.java

@@ -173,6 +173,9 @@ public void testRefreshAPI(int partition) throws Exception {
         TableView<byte[]> tv = pulsarClient.newTableView(Schema.BYTES)
                 .topic(topic)
                 .create();
+        // Verify refresh can handle the case when the topic is empty
+        tv.refreshAsync().get(3, TimeUnit.SECONDS);
+
         // 2. Add a listen action to provide the test environment.
         // The listen action will be triggered when there are incoming messages every time.
         // This is a sync operation, so sleep in the listen action can slow down the reading rate of messages.

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java

@@ -36,6 +36,7 @@
 import org.apache.pulsar.client.api.CryptoKeyReader;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageIdAdv;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Reader;
 import org.apache.pulsar.client.api.ReaderBuilder;
@@ -259,7 +260,11 @@ private void handleMessage(Message<T> msg) {
     @Override
     public CompletableFuture<Void> refreshAsync() {
         CompletableFuture<Void> completableFuture = new CompletableFuture<>();
-        reader.thenCompose(reader -> getLastMessageIds(reader).thenAccept(lastMessageIds -> {
+        reader.thenCompose(reader -> getLastMessageIdOfNonEmptyTopics(reader).thenAccept(lastMessageIds -> {
+            if (lastMessageIds.isEmpty()) {
+                completableFuture.complete(null);
+                return;
+            }
             // After get the response of lastMessageIds, put the future and result into `refreshMap`
             // and then filter out partitions that has been read to the lastMessageID.
             pendingRefreshRequests.put(completableFuture, lastMessageIds);
@@ -291,22 +296,28 @@ private CompletableFuture<Void> readAllExistingMessages(Reader<T> reader) {
         AtomicLong messagesRead = new AtomicLong();
 
         CompletableFuture<Void> future = new CompletableFuture<>();
-        getLastMessageIds(reader).thenAccept(maxMessageIds -> {
-            readAllExistingMessages(reader, future, startTime, messagesRead, maxMessageIds);
+        getLastMessageIdOfNonEmptyTopics(reader).thenAccept(lastMessageIds -> {
+            if (lastMessageIds.isEmpty()) {
+                future.complete(null);
+                return;
+            }
+            readAllExistingMessages(reader, future, startTime, messagesRead, lastMessageIds);
         }).exceptionally(ex -> {
             future.completeExceptionally(ex);
             return null;
         });
         return future;
     }
 
-    private CompletableFuture<Map<String, TopicMessageId>> getLastMessageIds(Reader<T> reader) {
+    private CompletableFuture<Map<String, TopicMessageId>> getLastMessageIdOfNonEmptyTopics(Reader<T> reader) {
         return reader.getLastMessageIdsAsync().thenApply(lastMessageIds -> {
-            Map<String, TopicMessageId> maxMessageIds = new ConcurrentHashMap<>();
+            Map<String, TopicMessageId> lastMessageIdMap = new ConcurrentHashMap<>();
             lastMessageIds.forEach(topicMessageId -> {
-                maxMessageIds.put(topicMessageId.getOwnerTopic(), topicMessageId);
+                if (((MessageIdAdv) topicMessageId).getEntryId() >= 0) {
+                    lastMessageIdMap.put(topicMessageId.getOwnerTopic(), topicMessageId);
+                } // else: a negative entry id represents an empty topic so that we don't have to read messages from it
             });
-            return maxMessageIds;
+            return lastMessageIdMap;
         });
     }