diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 0383c63b1f3bc..d77e224157a77 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -133,6 +133,7 @@
 import org.apache.pulsar.client.admin.PulsarAdminBuilder;
 import org.apache.pulsar.client.api.ClientBuilder;
 import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.SizeUnit;
 import org.apache.pulsar.client.impl.ClientBuilderImpl;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
@@ -970,6 +971,11 @@ public CompletableFuture<Optional<Topic>> getTopic(final TopicName topicName, bo
                 final CompletableFuture<Optional<TopicPolicies>> topicPoliciesFuture =
                         getTopicPoliciesBypassSystemTopic(topicName);
                 return topicPoliciesFuture.exceptionally(ex -> {
+                    if (!PulsarClientException.isRetriableError(ex.getCause())) {
+                        // Topic policies are not available, we should not prevent the creation of this topic
+                        log.warn("Failed to get topic policies due to non-retriable error: {}", ex.getMessage());
+                        return Optional.empty();
+                    }
                     final Throwable rc = FutureUtil.unwrapCompletionException(ex);
                     final String errorInfo = String.format("Topic creation encountered an exception by initialize"
                             + " topic policies service. topic_name=%s error_message=%s", topicName, rc.getMessage());
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java
index f8ae0279e08b7..d593f8d1d4b3f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java
@@ -45,6 +45,7 @@
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.DataProvider;
+import org.testng.annotations.Factory;
 import org.testng.annotations.Test;
 
 /**
@@ -58,14 +59,24 @@ public class TokenAuthenticatedProducerConsumerTest extends ProducerConsumerBase
 
     private final String ADMIN_TOKEN;
     private final String TOKEN_PUBLIC_KEY;
+    private final boolean brokerClientAuthEnabled;
+
+    @Factory
+    public static Object[] instances() throws NoSuchAlgorithmException {
+        return new Object[] {
+                new TokenAuthenticatedProducerConsumerTest(false),
+                new TokenAuthenticatedProducerConsumerTest(true),
+        };
+    }
 
-    TokenAuthenticatedProducerConsumerTest() throws NoSuchAlgorithmException {
+    TokenAuthenticatedProducerConsumerTest(boolean brokerClientAuthEnabled) throws NoSuchAlgorithmException {
         KeyPairGenerator kpg = KeyPairGenerator.getInstance("RSA");
         KeyPair kp = kpg.generateKeyPair();
 
         byte[] encodedPublicKey = kp.getPublic().getEncoded();
         TOKEN_PUBLIC_KEY = "data:;base64," + Base64.getEncoder().encodeToString(encodedPublicKey);
         ADMIN_TOKEN = generateToken(kp);
+        this.brokerClientAuthEnabled = brokerClientAuthEnabled;
     }
 
     private String generateToken(KeyPair kp) {
@@ -93,10 +104,13 @@ protected void setup() throws Exception {
         Set<String> providers = new HashSet<>();
         providers.add(AuthenticationProviderToken.class.getName());
         conf.setAuthenticationProviders(providers);
-        conf.setBrokerClientAuthenticationPlugin(AuthenticationToken.class.getName());
-        conf.setBrokerClientAuthenticationParameters("token:" + ADMIN_TOKEN);
+        if (brokerClientAuthEnabled) {
+            conf.setBrokerClientAuthenticationPlugin(AuthenticationToken.class.getName());
+            conf.setBrokerClientAuthenticationParameters("token:" + ADMIN_TOKEN);
+        }
 
-        conf.setClusterName("test");
+        final var clusterName = "test";
+        conf.setClusterName(clusterName);
 
         // Set provider domain name
         Properties properties = new Properties();
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
index 007308ec7ab46..369e4ecbcef01 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
@@ -1158,6 +1158,7 @@ public void setSequenceId(long sequenceId) {
 
     public static boolean isRetriableError(Throwable t) {
         if (t instanceof AuthorizationException
+                || t instanceof AuthenticationException
                 || t instanceof InvalidServiceURL
                 || t instanceof InvalidConfigurationException
                 || t instanceof NotFoundException