diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java
index 115bbe56ffa35..58559131af82f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java
@@ -45,6 +45,7 @@
 import org.apache.pulsar.metadata.api.Notification;
 import org.apache.pulsar.metadata.api.NotificationType;
 import org.apache.pulsar.metadata.api.extended.CreateOption;
+import org.apache.pulsar.metadata.api.extended.SessionEvent;
 
 /**
  * The broker registry impl, base on the LockManager.
@@ -112,6 +113,7 @@ public synchronized void start() throws PulsarServerException {
             throw new PulsarServerException("Cannot start the broker registry in state " + state.get());
         }
         pulsar.getLocalMetadataStore().registerListener(this::handleMetadataStoreNotification);
+        pulsar.getLocalMetadataStore().registerSessionListener(this::handleMetadataSessionEvent);
         try {
             this.registerAsync().get(conf.getMetadataStoreOperationTimeoutSeconds(), TimeUnit.SECONDS);
         } catch (ExecutionException | InterruptedException | TimeoutException e) {
@@ -282,6 +284,18 @@ private void handleMetadataStoreNotification(Notification t) {
         }
     }
 
+    private void handleMetadataSessionEvent(SessionEvent event) {
+        if (!this.isStarted()) {
+            return;
+        }
+        if (log.isDebugEnabled()) {
+            log.debug("Handle metadata session event: [{}]", event);
+        }
+        if (event == SessionEvent.SessionReestablished || event == SessionEvent.Reconnected) {
+            this.registerAsyncWithRetries();
+        }
+    }
+
     @VisibleForTesting
     protected static boolean isVerifiedNotification(Notification t) {
        return t.getPath().startsWith(LOADBALANCE_BROKERS_ROOT + "/")
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ZkSessionExpireTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ZkSessionExpireTest.java
index f9632625e6820..e19fd09d0629f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ZkSessionExpireTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ZkSessionExpireTest.java
@@ -202,11 +202,11 @@ public void testTopicUnloadAfterSessionRebuild(boolean enableSystemTopic, Class
         metadataZKProxy.unRejectAllConnections();
         Awaitility.await().untilAsserted(() -> {
             Set<String> availableBrokers1 = getAvailableBrokers(pulsar1);
-            Set<String> availableBrokers2 = getAvailableBrokers(pulsar1);
+            Set<String> availableBrokers2 = getAvailableBrokers(pulsar2);
             log.info("Available brokers 1: {}", availableBrokers1);
             log.info("Available brokers 2: {}", availableBrokers2);
             assertEquals(availableBrokers1.size(), 2);
-            assertEquals(availableBrokers1.size(), 2);
+            assertEquals(availableBrokers2.size(), 2);
         });
 
         // Verify: the topic on broker-1 will be unloaded.