From 873f9382ad7ce8e9df76edd919af72bda58e49f3 Mon Sep 17 00:00:00 2001
From: Jia Zhai <zhaijia@apache.org>
Date: Thu, 11 Jun 2020 17:27:40 +0800
Subject: [PATCH] Fix the regression from #6428

---
 .../authorization/AuthorizationProvider.java       | 14 ++++++++------
 .../authorization/PulsarAuthorizationProvider.java | 10 +++++-----
 2 files changed, 13 insertions(+), 11 deletions(-)

diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java
index c4fcc5ed860e8..4eb5d93dbdc4e 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java
@@ -47,25 +47,27 @@ public interface AuthorizationProvider extends Closeable {
     /**
      * Check if specified role is a super user
      * @param role the role to check
+     * @param authenticationData authentication data related to the role
      * @return a CompletableFuture containing a boolean in which true means the role is a super user
      * and false if it is not
      */
-    default CompletableFuture<Boolean> isSuperUser(String role, ServiceConfiguration serviceConfiguration) {
+    default CompletableFuture<Boolean> isSuperUser(String role,
+                                                   AuthenticationDataSource authenticationData,
+                                                   ServiceConfiguration serviceConfiguration) {
         Set<String> superUserRoles = serviceConfiguration.getSuperUserRoles();
         return CompletableFuture.completedFuture(role != null && superUserRoles.contains(role) ? true : false);
     }
 
     /**
+     * @deprecated Use method {@link #isSuperUser(String, AuthenticationDataSource, ServiceConfiguration)}
      * Check if specified role is a super user
      * @param role the role to check
-     * @param authenticationData authentication data related to the role
      * @return a CompletableFuture containing a boolean in which true means the role is a super user
      * and false if it is not
      */
-    default CompletableFuture<Boolean> isSuperUser(String role,
-                                                   AuthenticationDataSource authenticationData,
-                                                   ServiceConfiguration serviceConfiguration) {
-        return isSuperUser(role, serviceConfiguration);
+    default CompletableFuture<Boolean> isSuperUser(String role, ServiceConfiguration serviceConfiguration) {
+        Set<String> superUserRoles = serviceConfiguration.getSuperUserRoles();
+        return CompletableFuture.completedFuture(role != null && superUserRoles.contains(role) ? true : false);
     }
 
     /**
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
index 40b20217d4200..1aa79bf965ae4 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
@@ -322,7 +322,7 @@ public CompletableFuture<Void> revokeSubscriptionPermissionAsync(NamespaceName n
             String role, String authDataJson) {
         return updateSubscriptionPermissionAsync(namespace, subscriptionName, Collections.singleton(role), true);
     }
-    
+
     private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName, Set<String> roles,
             boolean remove) {
         CompletableFuture<Void> result = new CompletableFuture<>();
@@ -549,7 +549,7 @@ public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topicName,
                     new IllegalStateException("TopicOperation is not supported."));
         }
 
-        CompletableFuture<Boolean> isSuperUserFuture = isSuperUser(role, conf);
+        CompletableFuture<Boolean> isSuperUserFuture = isSuperUser(role, authData, conf);
 
         return isSuperUserFuture
                 .thenCombine(isAuthorizedFuture, (isSuperUser, isAuthorized) -> isSuperUser || isAuthorized);
@@ -573,14 +573,14 @@ private CompletableFuture<Boolean> validateTenantAdminAccess(String tenantName,
 
             if (role != null && conf.getProxyRoles().contains(role)) {
                 // role check
-                CompletableFuture<Boolean> isRoleSuperUserFuture = isSuperUser(role, conf);
+                CompletableFuture<Boolean> isRoleSuperUserFuture = isSuperUser(role, authData, conf);
                 CompletableFuture<Boolean> isRoleTenantAdminFuture = isTenantAdmin(tenantName, role, tenantInfo, authData);
                 CompletableFuture<Boolean> isRoleAuthorizedFuture = isRoleSuperUserFuture
                         .thenCombine(isRoleTenantAdminFuture, (isRoleSuperUser, isRoleTenantAdmin) ->
                                 isRoleSuperUser || isRoleTenantAdmin);
 
                 // originalRole check
-                CompletableFuture<Boolean> isOriginalRoleSuperUserFuture = isSuperUser(originalRole, conf);
+                CompletableFuture<Boolean> isOriginalRoleSuperUserFuture = isSuperUser(originalRole, authData, conf);
                 CompletableFuture<Boolean> isOriginalRoleTenantAdminFuture = isTenantAdmin(tenantName, originalRole,
                         tenantInfo, authData);
                 CompletableFuture<Boolean> isOriginalRoleAuthorizedFuture = isOriginalRoleSuperUserFuture
@@ -593,7 +593,7 @@ private CompletableFuture<Boolean> validateTenantAdminAccess(String tenantName,
                                 isRoleAuthorized && isOriginalRoleAuthorized);
             } else {
                 // role check
-                CompletableFuture<Boolean> isRoleSuperUserFuture = isSuperUser(role, conf);
+                CompletableFuture<Boolean> isRoleSuperUserFuture = isSuperUser(role, authData, conf);
                 CompletableFuture<Boolean> isRoleTenantAdminFuture = isTenantAdmin(tenantName, role, tenantInfo, authData);
                 return isRoleSuperUserFuture
                         .thenCombine(isRoleTenantAdminFuture, (isRoleSuperUser, isRoleTenantAdmin) ->