diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java
index d29497b5a27d7..279e975214e3c 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java
@@ -97,7 +97,7 @@ public CompletableFuture<Boolean> isSuperUser(String role, AuthenticationDataSou
         if (role != null && superUserRoles.contains(role)) {
             return CompletableFuture.completedFuture(true);
         }
-        Set<String> roles = getRoles(authenticationData);
+        Set<String> roles = getRoles(role, authenticationData);
         if (roles.isEmpty()) {
             return CompletableFuture.completedFuture(false);
         }
@@ -112,7 +112,7 @@ public CompletableFuture<Boolean> validateTenantAdminAccess(String tenantName, S
                     if (isSuperUser) {
                         return CompletableFuture.completedFuture(true);
                     }
-                    Set<String> roles = getRoles(authData);
+                    Set<String> roles = getRoles(role, authData);
                     if (roles.isEmpty()) {
                         return CompletableFuture.completedFuture(false);
                     }
@@ -143,7 +143,11 @@ public CompletableFuture<Boolean> validateTenantAdminAccess(String tenantName, S
                 });
     }
 
-    private Set<String> getRoles(AuthenticationDataSource authData) {
+    private Set<String> getRoles(String role, AuthenticationDataSource authData) {
+        if (authData == null) {
+            return Collections.singleton(role);
+        }
+
         String token = null;
 
         if (authData.hasDataFromCommand()) {
@@ -192,9 +196,9 @@ private Set<String> getRoles(AuthenticationDataSource authData) {
         return Collections.emptySet();
     }
 
-    public CompletableFuture<Boolean> authorize(AuthenticationDataSource authenticationData, Function<String,
-            CompletableFuture<Boolean>> authorizeFunc) {
-        Set<String> roles = getRoles(authenticationData);
+    public CompletableFuture<Boolean> authorize(String role, AuthenticationDataSource authenticationData,
+                                                Function<String, CompletableFuture<Boolean>> authorizeFunc) {
+        Set<String> roles = getRoles(role, authenticationData);
         if (roles.isEmpty()) {
             return CompletableFuture.completedFuture(false);
         }
@@ -212,7 +216,7 @@ public CompletableFuture<Boolean> authorize(AuthenticationDataSource authenticat
     @Override
     public CompletableFuture<Boolean> canProduceAsync(TopicName topicName, String role,
                                                       AuthenticationDataSource authenticationData) {
-        return authorize(authenticationData, r -> super.canProduceAsync(topicName, r, authenticationData));
+        return authorize(role, authenticationData, r -> super.canProduceAsync(topicName, r, authenticationData));
     }
 
     /**
@@ -227,7 +231,7 @@ public CompletableFuture<Boolean> canProduceAsync(TopicName topicName, String ro
     public CompletableFuture<Boolean> canConsumeAsync(TopicName topicName, String role,
                                                       AuthenticationDataSource authenticationData,
                                                       String subscription) {
-        return authorize(authenticationData, r -> super.canConsumeAsync(topicName, r, authenticationData,
+        return authorize(role, authenticationData, r -> super.canConsumeAsync(topicName, r, authenticationData,
                 subscription));
     }
 
@@ -244,25 +248,27 @@ public CompletableFuture<Boolean> canConsumeAsync(TopicName topicName, String ro
     @Override
     public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String role,
                                                      AuthenticationDataSource authenticationData) {
-        return authorize(authenticationData, r -> super.canLookupAsync(topicName, r, authenticationData));
+        return authorize(role, authenticationData, r -> super.canLookupAsync(topicName, r, authenticationData));
     }
 
     @Override
     public CompletableFuture<Boolean> allowFunctionOpsAsync(NamespaceName namespaceName, String role,
                                                             AuthenticationDataSource authenticationData) {
-        return authorize(authenticationData, r -> super.allowFunctionOpsAsync(namespaceName, r, authenticationData));
+        return authorize(role, authenticationData,
+                r -> super.allowFunctionOpsAsync(namespaceName, r, authenticationData));
     }
 
     @Override
     public CompletableFuture<Boolean> allowSourceOpsAsync(NamespaceName namespaceName, String role,
                                                           AuthenticationDataSource authenticationData) {
-        return authorize(authenticationData, r -> super.allowSourceOpsAsync(namespaceName, r, authenticationData));
+        return authorize(role, authenticationData,
+                r -> super.allowSourceOpsAsync(namespaceName, r, authenticationData));
     }
 
     @Override
     public CompletableFuture<Boolean> allowSinkOpsAsync(NamespaceName namespaceName, String role,
                                                         AuthenticationDataSource authenticationData) {
-        return authorize(authenticationData, r -> super.allowSinkOpsAsync(namespaceName, r, authenticationData));
+        return authorize(role, authenticationData, r -> super.allowSinkOpsAsync(namespaceName, r, authenticationData));
     }
 
     @Override
@@ -270,7 +276,7 @@ public CompletableFuture<Boolean> allowTenantOperationAsync(String tenantName,
                                                                 String role,
                                                                 TenantOperation operation,
                                                                 AuthenticationDataSource authData) {
-        return authorize(authData, r -> super.allowTenantOperationAsync(tenantName, r, operation, authData));
+        return authorize(role, authData, r -> super.allowTenantOperationAsync(tenantName, r, operation, authData));
     }
 
     @Override
@@ -278,7 +284,8 @@ public CompletableFuture<Boolean> allowNamespaceOperationAsync(NamespaceName nam
                                                                    String role,
                                                                    NamespaceOperation operation,
                                                                    AuthenticationDataSource authData) {
-        return authorize(authData, r -> super.allowNamespaceOperationAsync(namespaceName, r, operation, authData));
+        return authorize(role, authData,
+                r -> super.allowNamespaceOperationAsync(namespaceName, r, operation, authData));
     }
 
     @Override
@@ -287,8 +294,8 @@ public CompletableFuture<Boolean> allowNamespacePolicyOperationAsync(NamespaceNa
                                                                          PolicyOperation operation,
                                                                          String role,
                                                                          AuthenticationDataSource authData) {
-        return authorize(authData, r -> super.allowNamespacePolicyOperationAsync(namespaceName, policy, operation, r,
-                authData));
+        return authorize(role, authData,
+                r -> super.allowNamespacePolicyOperationAsync(namespaceName, policy, operation, r, authData));
     }
 
     @Override
@@ -296,7 +303,7 @@ public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topicName,
                                                                String role,
                                                                TopicOperation operation,
                                                                AuthenticationDataSource authData) {
-        return authorize(authData, r -> super.allowTopicOperationAsync(topicName, r, operation, authData));
+        return authorize(role, authData, r -> super.allowTopicOperationAsync(topicName, r, operation, authData));
     }
 
     @Override
@@ -305,7 +312,7 @@ public CompletableFuture<Boolean> allowTopicPolicyOperationAsync(TopicName topic
                                                                      PolicyName policyName,
                                                                      PolicyOperation policyOperation,
                                                                      AuthenticationDataSource authData) {
-        return authorize(authData, r -> super.allowTopicPolicyOperationAsync(topicName, r, policyName, policyOperation,
-                authData));
+        return authorize(role, authData,
+                r -> super.allowTopicPolicyOperationAsync(topicName, r, policyName, policyOperation, authData));
     }
 }
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProviderTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProviderTest.java
index 2ac9524fdef38..e5e3612eafb01 100644
--- a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProviderTest.java
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProviderTest.java
@@ -19,14 +19,20 @@
 package org.apache.pulsar.broker.authorization;
 
 import static org.mockito.Mockito.mock;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.assertFalse;
+
 import io.jsonwebtoken.Jwts;
 import io.jsonwebtoken.SignatureAlgorithm;
+
+import java.util.HashSet;
 import java.util.Properties;
+import java.util.function.Function;
+import lombok.Cleanup;
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
 import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
 import org.apache.pulsar.broker.resources.PulsarResources;
-import org.junit.Assert;
 import org.testng.annotations.Test;
 
 import javax.crypto.SecretKey;
@@ -60,18 +66,18 @@ public String getHttpHeader(String name) {
             }
         };
 
-        Assert.assertTrue(provider.authorize(ads, role -> {
+        assertTrue(provider.authorize("test", ads, role -> {
             if (role.equals(userB)) {
                 return CompletableFuture.completedFuture(true); // only userB has permission
             }
             return CompletableFuture.completedFuture(false);
         }).get());
 
-        Assert.assertTrue(provider.authorize(ads, role -> {
+        assertTrue(provider.authorize("test", ads, role -> {
             return CompletableFuture.completedFuture(true); // all users has permission
         }).get());
 
-        Assert.assertFalse(provider.authorize(ads, role -> {
+        assertFalse(provider.authorize("test", ads, role -> {
             return CompletableFuture.completedFuture(false); // all users has no permission
         }).get());
     }
@@ -99,7 +105,7 @@ public String getHttpHeader(String name) {
             }
         };
 
-        Assert.assertFalse(provider.authorize(ads, role -> CompletableFuture.completedFuture(false)).get());
+        assertFalse(provider.authorize("test", ads, role -> CompletableFuture.completedFuture(false)).get());
     }
 
     @Test
@@ -126,7 +132,7 @@ public String getHttpHeader(String name) {
             }
         };
 
-        Assert.assertTrue(provider.authorize(ads, role -> {
+        assertTrue(provider.authorize("test", ads, role -> {
             if (role.equals(testRole)) {
                 return CompletableFuture.completedFuture(true);
             }
@@ -134,6 +140,21 @@ public String getHttpHeader(String name) {
         }).get());
     }
 
+    @Test
+    public void testMultiRolesAuthzWithAnonymousUser() throws Exception {
+        @Cleanup
+        MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
+
+        Function<String, CompletableFuture<Boolean>> authorizeFunc = (String role) -> {
+            if (role.equals("test-role")) {
+                return CompletableFuture.completedFuture(true);
+            }
+            return CompletableFuture.completedFuture(false);
+        };
+        assertTrue(provider.authorize("test-role", null, authorizeFunc).get());
+        assertFalse(provider.authorize("test-role-x", null, authorizeFunc).get());
+    }
+
     @Test
     public void testMultiRolesNotFailNonJWT() throws Exception {
         String token = "a-non-jwt-token";
@@ -156,7 +177,7 @@ public String getHttpHeader(String name) {
             }
         };
 
-        Assert.assertFalse(provider.authorize(ads, role -> CompletableFuture.completedFuture(false)).get());
+        assertFalse(provider.authorize("test", ads, role -> CompletableFuture.completedFuture(false)).get());
     }
 
     @Test
@@ -190,8 +211,7 @@ public String getHttpHeader(String name) {
                 }
             }
         };
-
-        Assert.assertTrue(provider.authorize(ads, role -> {
+        assertTrue(provider.authorize("test", ads, role -> {
             if (role.equals(testRole)) {
                 return CompletableFuture.completedFuture(true);
             }
@@ -206,7 +226,9 @@ public void testMultiRolesAuthzWithSuperUser() throws Exception {
         String token = Jwts.builder().claim("sub", testAdminRole).signWith(secretKey).compact();
 
         ServiceConfiguration conf = new ServiceConfiguration();
-        conf.setSuperUserRoles(Set.of(testAdminRole));
+        Set rols = new HashSet();
+        rols.add(testAdminRole);
+        conf.setSuperUserRoles(rols);
 
         MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
         provider.initialize(conf, mock(PulsarResources.class));