allow superuser to consume without subscription_auth_mode check (apac…

…he#1273)
vruc · Feb 23, 2018 · 7eb1002 · 7eb1002
1 parent adbf818
commit 7eb1002
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 1 deletion.
diff --git a/...mon/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java b/...mon/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
@@ -109,7 +109,7 @@ public CompletableFuture<Boolean> canConsumeAsync(DestinationName destination, S
                         log.debug("Policies node couldn't be found for destination : {}", destination);
                     }
                 } else {
-                    if (isNotBlank(subscription)) {
+                    if (isNotBlank(subscription) && !isSuperUser(role)) {
                         switch (policies.get().subscription_auth_mode) {
                         case Prefix:
                             if (!subscription.startsWith(role)) {

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationTest.java
@@ -49,6 +49,7 @@ protected void setup() throws Exception {
         conf.setClusterName("c1");
         conf.setAuthorizationEnabled(true);
         conf.setAuthorizationAllowWildcardsMatching(true);
+        conf.setSuperUserRoles(Sets.newHashSet("pulsar.super_user"));
         internalSetup();
     }
 
@@ -206,6 +207,7 @@ void simple() throws Exception {
 
         assertEquals(auth.canConsume(DestinationName.get("persistent://p1/c1/ns1/ds1"), "role1", null, "role1-sub1"), true);
         assertEquals(auth.canConsume(DestinationName.get("persistent://p1/c1/ns1/ds1"), "role2", null, "role2-sub2"), true);
+        assertEquals(auth.canConsume(DestinationName.get("persistent://p1/c1/ns1/ds1"), "pulsar.super_user", null, "role3-sub1"), true);
 
         admin.namespaces().deleteNamespace("p1/c1/ns1");
         admin.properties().deleteProperty("p1");