diff --git a/pom.xml b/pom.xml
index 33f9ced9f1cd7..c1289c32107d0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -245,7 +245,7 @@ flexible messaging model and an intuitive client API.</description>
     <kotlin-stdlib.version>1.8.20</kotlin-stdlib.version>
     <nsq-client.version>1.0</nsq-client.version>
     <cron-utils.version>9.1.6</cron-utils.version>
-    <spring.version>5.3.27</spring.version>
+    <spring.version>6.1.13</spring.version>
     <apache-http-client.version>4.5.13</apache-http-client.version>
     <apache-httpcomponents.version>4.4.15</apache-httpcomponents.version>
     <jetcd.version>0.7.7</jetcd.version>
@@ -2153,7 +2153,7 @@ flexible messaging model and an intuitive client API.</description>
             <!-- <nvdDatafeedUrl>https://jeremylong.github.io/DependencyCheck/hb_nvd/</nvdDatafeedUrl> -->
           </configuration>
         </plugin>
-        <!-- 
+        <!--
           vscode-java and Eclipse Maven integration workaround:
           - Addresses maven-dependency-plugin exception (MDEP-187) issue by ignoring the plugin execution
           - Uses org.eclipse.m2e:lifecycle-mapping:1.0.0 as configuration placeholder in pluginManagement section
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java
index 624f8f9236c5e..fede660c2c716 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java
@@ -137,8 +137,17 @@ private static class Arguments {
                 hidden = false)
         private String configurationMetadataStore;
 
+        @Parameter(names = {"-mscp",
+                "--metadata-store-config-path"}, description = "Metadata Store config path", hidden = false)
+        private String metadataStoreConfigPath;
+
+        @Parameter(names = {"-cmscp",
+                "--configuration-metadata-store-config-path"}, description = "Configuration Metadata Store config path",
+                hidden = false)
+        private String configurationStoreConfigPath;
+
         @Parameter(names = {
-            "--initial-num-stream-storage-containers"
+                "--initial-num-stream-storage-containers"
         }, description = "Num storage containers of BookKeeper stream storage")
         private int numStreamStorageContainers = 16;
 
@@ -281,9 +290,11 @@ private static void initializeCluster(Arguments arguments, int bundleNumberForDe
         log.info("Setting up cluster {} with metadata-store={} configuration-metadata-store={}", arguments.cluster,
                 arguments.metadataStoreUrl, arguments.configurationMetadataStore);
 
-        MetadataStoreExtended localStore =
-                initLocalMetadataStore(arguments.metadataStoreUrl, arguments.zkSessionTimeoutMillis);
+        MetadataStoreExtended localStore = initLocalMetadataStore(arguments.metadataStoreUrl,
+                arguments.metadataStoreConfigPath,
+                arguments.zkSessionTimeoutMillis);
         MetadataStoreExtended configStore = initConfigMetadataStore(arguments.configurationMetadataStore,
+                arguments.configurationStoreConfigPath,
                 arguments.zkSessionTimeoutMillis);
 
         final String metadataStoreUrlNoIdentifer = MetadataStoreFactoryImpl
@@ -462,9 +473,17 @@ static void createPartitionedTopic(MetadataStore configStore, TopicName topicNam
         }
     }
 
-    public static MetadataStoreExtended initLocalMetadataStore(String connection, int sessionTimeout) throws Exception {
+    public static MetadataStoreExtended initLocalMetadataStore(String connection,
+                                                               int sessionTimeout) throws Exception {
+        return initLocalMetadataStore(connection, null, sessionTimeout);
+    }
+
+    public static MetadataStoreExtended initLocalMetadataStore(String connection,
+                                                               String configPath,
+                                                               int sessionTimeout) throws Exception {
         MetadataStoreExtended store = MetadataStoreExtended.create(connection, MetadataStoreConfig.builder()
                 .sessionTimeoutMillis(sessionTimeout)
+                .configFilePath(configPath)
                 .metadataStoreName(MetadataStoreConfig.METADATA_STORE)
                 .build());
         if (store instanceof MetadataStoreLifecycle) {
@@ -473,10 +492,19 @@ public static MetadataStoreExtended initLocalMetadataStore(String connection, in
         return store;
     }
 
-    public static MetadataStoreExtended initConfigMetadataStore(String connection, int sessionTimeout)
+    public static MetadataStoreExtended initConfigMetadataStore(String connection,
+                                                                int sessionTimeout)
+            throws Exception {
+        return initConfigMetadataStore(connection, null, sessionTimeout);
+    }
+
+    public static MetadataStoreExtended initConfigMetadataStore(String connection,
+                                                                String configPath,
+                                                                int sessionTimeout)
             throws Exception {
         MetadataStoreExtended store = MetadataStoreExtended.create(connection, MetadataStoreConfig.builder()
                 .sessionTimeoutMillis(sessionTimeout)
+                .configFilePath(configPath)
                 .metadataStoreName(MetadataStoreConfig.CONFIGURATION_METADATA_STORE)
                 .build());
         if (store instanceof MetadataStoreLifecycle) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarInitialNamespaceSetup.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarInitialNamespaceSetup.java
index 22b38e59676ec..ac7ef9961d338 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarInitialNamespaceSetup.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarInitialNamespaceSetup.java
@@ -40,6 +40,11 @@ private static class Arguments {
                 "--configuration-store" }, description = "Configuration Store connection string", required = true)
         private String configurationStore;
 
+        @Parameter(names = {"-cmscp",
+                "--configuration-metadata-store-config-path"}, description = "Configuration Metadata Store config path",
+                hidden = false)
+        private String configurationStoreConfigPath;
+
         @Parameter(names = {
                 "--zookeeper-session-timeout-ms"
         }, description = "Local zookeeper session timeout ms")
@@ -82,8 +87,10 @@ public static int doMain(String[] args) throws Exception {
             return 1;
         }
 
-        try (MetadataStore configStore = PulsarClusterMetadataSetup
-                .initConfigMetadataStore(arguments.configurationStore, arguments.zkSessionTimeoutMillis)) {
+        try (MetadataStore configStore = PulsarClusterMetadataSetup.initConfigMetadataStore(
+                arguments.configurationStore,
+                arguments.configurationStoreConfigPath,
+                arguments.zkSessionTimeoutMillis)) {
             PulsarResources pulsarResources = new PulsarResources(null, configStore);
             for (String namespace : arguments.namespaces) {
                 NamespaceName namespaceName = null;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarTransactionCoordinatorMetadataSetup.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarTransactionCoordinatorMetadataSetup.java
index 6aedfe13a5b50..ec4c7f237ba3b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarTransactionCoordinatorMetadataSetup.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarTransactionCoordinatorMetadataSetup.java
@@ -41,6 +41,11 @@ private static class Arguments {
                 "--configuration-store" }, description = "Configuration Store connection string", required = true)
         private String configurationStore;
 
+        @Parameter(names = {"-cmscp",
+                "--configuration-metadata-store-config-path"}, description = "Configuration Metadata Store config path",
+                hidden = false)
+        private String configurationStoreConfigPath;
+
         @Parameter(names = {
                 "--zookeeper-session-timeout-ms"
         }, description = "Local zookeeper session timeout ms")
@@ -90,8 +95,10 @@ public static void main(String[] args) throws Exception {
             System.exit(1);
         }
 
-        try (MetadataStoreExtended configStore = PulsarClusterMetadataSetup
-                .initConfigMetadataStore(arguments.configurationStore, arguments.zkSessionTimeoutMillis)) {
+        try (MetadataStoreExtended configStore = PulsarClusterMetadataSetup.initConfigMetadataStore(
+                arguments.configurationStore,
+                arguments.configurationStoreConfigPath,
+                arguments.zkSessionTimeoutMillis)) {
             PulsarResources pulsarResources = new PulsarResources(null, configStore);
             // Create system tenant
             PulsarClusterMetadataSetup
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
index e344d892b3137..c5fd889881bef 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
@@ -127,7 +127,16 @@ public CompletableFuture<Void> deleteTopicPoliciesAsync(TopicName topicName) {
         if (NamespaceService.isHeartbeatNamespace(topicName.getNamespaceObject())) {
             return CompletableFuture.completedFuture(null);
         }
-        return sendTopicPolicyEvent(topicName, ActionType.DELETE, null);
+        TopicName changeEvents = NamespaceEventsSystemTopicFactory.getEventsTopicName(topicName.getNamespaceObject());
+        return pulsarService.getNamespaceService().checkTopicExists(changeEvents).thenCompose(topicExistsInfo -> {
+            // If the system topic named "__change_events" has been deleted, it means all the data in the topic have
+            // been deleted, so we do not need to delete the message that we want to delete again.
+            if (!topicExistsInfo.isExists()) {
+                log.info("Skip delete topic-level policies because {} has been removed before", changeEvents);
+                return CompletableFuture.completedFuture(null);
+            }
+            return sendTopicPolicyEvent(topicName, ActionType.DELETE, null);
+        });
     }
 
     @Override
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/NamespaceEventsSystemTopicFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/NamespaceEventsSystemTopicFactory.java
index f5e6c7748d10b..199026bc4c445 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/NamespaceEventsSystemTopicFactory.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/NamespaceEventsSystemTopicFactory.java
@@ -37,12 +37,16 @@ public NamespaceEventsSystemTopicFactory(PulsarClient client) {
     }
 
     public TopicPoliciesSystemTopicClient createTopicPoliciesSystemTopicClient(NamespaceName namespaceName) {
-        TopicName topicName = TopicName.get(TopicDomain.persistent.value(), namespaceName,
-                SystemTopicNames.NAMESPACE_EVENTS_LOCAL_NAME);
+        TopicName topicName = getEventsTopicName(namespaceName);
         log.info("Create topic policies system topic client {}", topicName.toString());
         return new TopicPoliciesSystemTopicClient(client, topicName);
     }
 
+    public static TopicName getEventsTopicName(NamespaceName namespaceName) {
+        return TopicName.get(TopicDomain.persistent.value(), namespaceName,
+                SystemTopicNames.NAMESPACE_EVENTS_LOCAL_NAME);
+    }
+
     public <T> TransactionBufferSnapshotBaseSystemTopicClient<T> createTransactionBufferSystemTopicClient(
             TopicName systemTopicName, SystemTopicTxnBufferSnapshotService<T>
             systemTopicTxnBufferSnapshotService, Class<T> schemaType) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
index 42912d17c1c2b..bf4c1f938f035 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
@@ -1893,4 +1893,76 @@ public void testCreateMissingPartitions() throws Exception {
         String topicName = "persistent://" + testTenant + "/" + testNamespaceLocal + "/testCreateMissingPartitions";
         assertThrows(PulsarAdminException.NotFoundException.class, () -> admin.topics().createMissedPartitions(topicName));
     }
+
+    @Test
+    public void testPeekMessageWithProperties() throws Exception {
+        String topicName = "persistent://" + testTenant + "/" + testNamespaceLocal + "/testPeekMessageWithProperties";
+        admin.topics().createNonPartitionedTopic(topicName);
+
+        // Test non-batch messages
+        @Cleanup
+        Producer<String> nonBatchProducer = pulsarClient.newProducer(Schema.STRING)
+                .topic(topicName)
+                .enableBatching(false)
+                .create();
+
+        Map<String, String> props1 = new HashMap<>();
+        props1.put("key1", "value1");
+        props1.put("KEY2", "VALUE2");
+        props1.put("KeY3", "VaLuE3");
+
+        nonBatchProducer.newMessage()
+                .properties(props1)
+                .value("non-batch-message")
+                .send();
+
+        Message<byte[]> peekedMessage = admin.topics().peekMessages(topicName, "sub-peek", 1).get(0);
+        assertEquals(new String(peekedMessage.getData()), "non-batch-message");
+        assertEquals(peekedMessage.getProperties().size(), 3);
+        assertEquals(peekedMessage.getProperties().get("key1"), "value1");
+        assertEquals(peekedMessage.getProperties().get("KEY2"), "VALUE2");
+        assertEquals(peekedMessage.getProperties().get("KeY3"), "VaLuE3");
+
+        admin.topics().truncate(topicName);
+
+        // Test batch messages
+        @Cleanup
+        Producer<String> batchProducer = pulsarClient.newProducer(Schema.STRING)
+                .topic(topicName)
+                .enableBatching(true)
+                .batchingMaxPublishDelay(1, TimeUnit.SECONDS)
+                .batchingMaxMessages(2)
+                .create();
+
+        Map<String, String> props2 = new HashMap<>();
+        props2.put("batch-key1", "batch-value1");
+        props2.put("BATCH-KEY2", "BATCH-VALUE2");
+        props2.put("BaTcH-kEy3", "BaTcH-vAlUe3");
+
+        batchProducer.newMessage()
+                .properties(props2)
+                .value("batch-message-1")
+                .sendAsync();
+
+        batchProducer.newMessage()
+                .properties(props2)
+                .value("batch-message-2")
+                .send();
+
+        List<Message<byte[]>> peekedMessages = admin.topics().peekMessages(topicName, "sub-peek", 2);
+        assertEquals(peekedMessages.size(), 2);
+
+        for (int i = 0; i < 2; i++) {
+            Message<byte[]> batchMessage = peekedMessages.get(i);
+            assertEquals(new String(batchMessage.getData()), "batch-message-" + (i + 1));
+            assertEquals(batchMessage.getProperties().size(),
+                    3 + 2 // 3 properties from the message + 2 properties from the batch
+            );
+            assertEquals(batchMessage.getProperties().get("X-Pulsar-num-batch-message"), "2");
+            assertNotNull(batchMessage.getProperties().get("X-Pulsar-batch-size"));
+            assertEquals(batchMessage.getProperties().get("batch-key1"), "batch-value1");
+            assertEquals(batchMessage.getProperties().get("BATCH-KEY2"), "BATCH-VALUE2");
+            assertEquals(batchMessage.getProperties().get("BaTcH-kEy3"), "BaTcH-vAlUe3");
+        }
+    }
 }
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
index d99969fbaa7e5..949ac17124d6a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
@@ -19,17 +19,25 @@
 package org.apache.pulsar.broker.service;
 
 import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.HashSet;
+import java.util.Optional;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.broker.BrokerTestUtil;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.RetentionPolicies;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
 import org.awaitility.Awaitility;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
@@ -173,4 +181,42 @@ public void testDifferentTopicCreationRule(ReplicationMode replicationMode) thro
     public void testReplicationCountMetrics() throws Exception {
         super.testReplicationCountMetrics();
     }
+
+    @Test
+    public void testRemoveCluster() throws Exception {
+        // Initialize.
+        final String ns1 = defaultTenant + "/" + "ns_73b1a31afce34671a5ddc48fe5ad7fc8";
+        final String topic = "persistent://" + ns1 + "/___tp-5dd50794-7af8-4a34-8a0b-06188052c66a";
+        final String topicChangeEvents = "persistent://" + ns1 + "/__change_events";
+        admin1.namespaces().createNamespace(ns1);
+        admin1.namespaces().setNamespaceReplicationClusters(ns1, new HashSet<>(Arrays.asList(cluster1, cluster2)));
+        admin1.topics().createNonPartitionedTopic(topic);
+
+        // Wait for loading topic up.
+        Producer<String> p = client1.newProducer(Schema.STRING).topic(topic).create();
+        Awaitility.await().untilAsserted(() -> {
+            ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> tps
+                    = pulsar1.getBrokerService().getTopics();
+            assertTrue(tps.containsKey(topic));
+            assertTrue(tps.containsKey(topicChangeEvents));
+        });
+
+        // The topics under the namespace of the cluster-1 will be deleted.
+        // Verify the result.
+        admin1.namespaces().setNamespaceReplicationClusters(ns1, new HashSet<>(Arrays.asList(cluster2)));
+        Awaitility.await().atMost(Duration.ofSeconds(120)).untilAsserted(() -> {
+            ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> tps
+                    = pulsar1.getBrokerService().getTopics();
+            assertFalse(tps.containsKey(topic));
+            assertFalse(tps.containsKey(topicChangeEvents));
+            assertFalse(pulsar1.getNamespaceService().checkTopicExists(TopicName.get(topic)).join().isExists());
+            assertFalse(pulsar1.getNamespaceService()
+                    .checkTopicExists(TopicName.get(topicChangeEvents)).join().isExists());
+        });
+
+        // cleanup.
+        p.close();
+        admin2.topics().delete(topic);
+        admin2.namespaces().deleteNamespace(ns1);
+    }
 }
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/ClusterMetadataSetupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/ClusterMetadataSetupTest.java
index 4267c7564fa6f..6a7d54dbf7cae 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/ClusterMetadataSetupTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/ClusterMetadataSetupTest.java
@@ -74,10 +74,11 @@ public void testReSetupClusterMetadata() throws Exception {
             "--cluster", "testReSetupClusterMetadata-cluster",
             "--zookeeper", "127.0.0.1:" + localZkS.getZookeeperPort(),
             "--configuration-store", "127.0.0.1:" + localZkS.getZookeeperPort(),
+            "--configuration-metadata-store-config-path", "src/test/resources/conf/zk_client_enable_sasl.conf",
             "--web-service-url", "http://127.0.0.1:8080",
             "--web-service-url-tls", "https://127.0.0.1:8443",
             "--broker-service-url", "pulsar://127.0.0.1:6650",
-            "--broker-service-url-tls","pulsar+ssl://127.0.0.1:6651"
+            "--broker-service-url-tls", "pulsar+ssl://127.0.0.1:6651"
         };
         PulsarClusterMetadataSetup.main(args);
         SortedMap<String, String> data1 = localZkS.dumpData();
@@ -505,11 +506,9 @@ static class ZookeeperServerTest implements Closeable {
         private ZooKeeperServer zks;
         private NIOServerCnxnFactory serverFactory;
         private final int zkPort;
-        private final String hostPort;
 
         public ZookeeperServerTest(int zkPort) throws IOException {
             this.zkPort = zkPort;
-            this.hostPort = "127.0.0.1:" + zkPort;
             this.zkTmpDir = File.createTempFile("zookeeper", "test");
             log.info("**** Start GZK on {} ****", zkTmpDir);
             if (!zkTmpDir.delete() || !zkTmpDir.mkdir()) {
@@ -519,15 +518,17 @@ public ZookeeperServerTest(int zkPort) throws IOException {
 
         public void start() throws IOException {
             try {
+                System.setProperty("zookeeper.4lw.commands.whitelist", "*");
                 zks = new ZooKeeperServer(zkTmpDir, zkTmpDir, ZooKeeperServer.DEFAULT_TICK_TIME);
                 zks.setMaxSessionTimeout(20000);
                 serverFactory = new NIOServerCnxnFactory();
-                serverFactory.configure(new InetSocketAddress(zkPort), 1000);
+                serverFactory.configure(new InetSocketAddress("127.0.0.1", zkPort), 1000);
                 serverFactory.startup(zks);
             } catch (Exception e) {
                 log.error("Exception while instantiating ZooKeeper", e);
             }
 
+            String hostPort = "127.0.0.1:" + serverFactory.getLocalPort();
             LocalBookkeeperEnsemble.waitForServerUp(hostPort, 30000);
             log.info("ZooKeeper started at {}", hostPort);
         }
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SimpleProduceConsumeIoTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SimpleProduceConsumeIoTest.java
new file mode 100644
index 0000000000000..4da3ce2573334
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SimpleProduceConsumeIoTest.java
@@ -0,0 +1,134 @@
+/*
+ * 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.client.impl;
+
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
+import org.awaitility.Awaitility;
+import org.awaitility.reflect.WhiteboxImpl;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+public class SimpleProduceConsumeIoTest extends ProducerConsumerBase {
+
+    private PulsarClientImpl singleConnectionPerBrokerClient;
+
+    @BeforeClass(alwaysRun = true)
+    @Override
+    protected void setup() throws Exception {
+        super.internalSetup();
+        super.producerBaseSetup();
+        singleConnectionPerBrokerClient = (PulsarClientImpl) PulsarClient.builder().connectionsPerBroker(1)
+                .serviceUrl(lookupUrl.toString()).build();
+    }
+
+    @AfterClass(alwaysRun = true)
+    @Override
+    protected void cleanup() throws Exception {
+        if (singleConnectionPerBrokerClient != null) {
+            singleConnectionPerBrokerClient.close();
+        }
+        super.internalCleanup();
+    }
+
+    /**
+     * 1. Create a producer with a pooled connection.
+     * 2. When executing "producer.connectionOpened", the pooled connection has been closed due to a network issue.
+     * 3. Verify: the producer can be created successfully.
+     */
+    @Test
+    public void testUnstableNetWorkWhenCreatingProducer() throws Exception {
+        final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+        admin.topics().createNonPartitionedTopic(topic);
+        // Trigger a pooled connection creation.
+        ProducerImpl p = (ProducerImpl) singleConnectionPerBrokerClient.newProducer().topic(topic).create();
+        ClientCnx cnx = p.getClientCnx();
+        p.close();
+
+        // 1. Create a new producer with the pooled connection(since there is a pooled connection, the new producer
+        // will reuse it).
+        // 2. Trigger a network issue.
+        CountDownLatch countDownLatch = new CountDownLatch(1);
+        // A task for trigger network issue.
+        new Thread(() -> {
+            try {
+                countDownLatch.await();
+                cnx.ctx().close();
+            } catch (Exception ex) {
+            }
+        }).start();
+        // Create a new producer with the pooled connection.
+        AtomicReference<CompletableFuture<Producer<byte[]>>> p2FutureWrap = new AtomicReference<>();
+        new Thread(() -> {
+            ProducerBuilder producerBuilder = singleConnectionPerBrokerClient.newProducer().topic(topic);
+            ProducerConfigurationData producerConf = WhiteboxImpl.getInternalState(producerBuilder, "conf");
+            CompletableFuture<Producer<byte[]>> p2Future = new CompletableFuture();
+            p2FutureWrap.set(p2Future);
+            new ProducerImpl<>(singleConnectionPerBrokerClient, "public/default/tp1", producerConf, p2Future,
+                    -1, Schema.BYTES, null, Optional.empty()) {
+                @Override
+                public CompletableFuture<Void> connectionOpened(final ClientCnx cnx) {
+                    // Mock a network issue, and wait for the issue occurred.
+                    countDownLatch.countDown();
+                    try {
+                        Thread.sleep(1500);
+                    } catch (InterruptedException e) {
+                    }
+                    // Call the real implementation.
+                    return super.connectionOpened(cnx);
+                }
+            };
+        }).start();
+
+        // Verify: the producer can be created successfully.
+        Awaitility.await().untilAsserted(() -> {
+            assertNotNull(p2FutureWrap.get());
+            assertTrue(p2FutureWrap.get().isDone());
+        });
+        // Print log.
+        p2FutureWrap.get().exceptionally(ex -> {
+            log.error("Failed to create producer", ex);
+            return null;
+        });
+        Awaitility.await().untilAsserted(() -> {
+            assertFalse(p2FutureWrap.get().isCompletedExceptionally());
+            assertTrue("Ready".equals(
+                    WhiteboxImpl.getInternalState(p2FutureWrap.get().join(), "state").toString()));
+        });
+
+        // Cleanup.
+        p2FutureWrap.get().join().close();
+        admin.topics().delete(topic);
+    }
+}
diff --git a/pulsar-broker/src/test/resources/conf/zk_client_enable_sasl.conf b/pulsar-broker/src/test/resources/conf/zk_client_enable_sasl.conf
new file mode 100644
index 0000000000000..c59e093450d39
--- /dev/null
+++ b/pulsar-broker/src/test/resources/conf/zk_client_enable_sasl.conf
@@ -0,0 +1,20 @@
+#
+# 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.
+#
+
+zookeeper.sasl.client=true
diff --git a/pulsar-client-admin-shaded/pom.xml b/pulsar-client-admin-shaded/pom.xml
index 544c4cd406d74..ba77dd1892ce7 100644
--- a/pulsar-client-admin-shaded/pom.xml
+++ b/pulsar-client-admin-shaded/pom.xml
@@ -120,7 +120,6 @@
                   <include>org.reactivestreams:reactive-streams</include>
                   <include>com.typesafe.netty:netty-reactive-streams</include>
                   <include>org.javassist:javassist</include>
-                  <include>com.google.protobuf:protobuf-java</include>
                   <include>com.google.guava:guava</include>
                   <include>com.google.code.gson:gson</include>
                   <include>com.spotify:completable-futures</include>
@@ -153,6 +152,7 @@
                 </includes>
                 <excludes>
                   <exclude>com.fasterxml.jackson.core:jackson-annotations</exclude>
+                  <exclude>com.google.protobuf:protobuf-java</exclude>
                 </excludes>
               </artifactSet>
               <filters>
diff --git a/pulsar-client-all/pom.xml b/pulsar-client-all/pom.xml
index 6d33cfc2d5208..c80b84f5b64b3 100644
--- a/pulsar-client-all/pom.xml
+++ b/pulsar-client-all/pom.xml
@@ -160,7 +160,6 @@
                   <include>org.reactivestreams:reactive-streams</include>
                   <include>com.typesafe.netty:netty-reactive-streams</include>
                   <include>org.javassist:javassist</include>
-                  <include>com.google.protobuf:protobuf-java</include>
                   <include>com.google.guava:*</include>
                   <include>org.checkerframework:*</include>
                   <include>com.google.code.findbugs:*</include>
@@ -210,6 +209,7 @@
                 </includes>
                 <excludes>
                   <exclude>com.fasterxml.jackson.core:jackson-annotations</exclude>
+                  <exclude>com.google.protobuf:protobuf-java</exclude>
                 </excludes>
               </artifactSet>
               <filters>
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java
index 9c73684b356a5..fbacd9496c22d 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java
@@ -242,7 +242,7 @@ public CompletableFuture<ClientCnx> getConnection(InetSocketAddress logicalAddre
             }
             // Try use exists connection.
             if (clientCnx.getIdleState().tryMarkUsingAndClearIdleTime()) {
-                return CompletableFuture.completedFuture(clientCnx);
+                return CompletableFuture.supplyAsync(() -> clientCnx, clientCnx.ctx().executor());
             } else {
                 // If connection already release, create a new one.
                 cleanupConnection(logicalAddress, randomKey, completableFuture);