[pulsar] 03/20: Fix lookup permission error (#7234)

[rxl]

This is an automated email from the ASF dual-hosted git repository.

rxl pushed a commit to branch branch-2.6
in repository https://gitbox.apache.org/repos/asf/pulsar.git

commit e752e0f805cf2e8d2915c7dc77d8feb987343880
Author: ltamber <ltambe...@gmail.com>
AuthorDate: Thu Jun 11 15:59:22 2020 +0800

    Fix lookup permission error (#7234)
    
    ### Motivation
    Currently，when pulsar AuthorizationService check lookup permission, if the 
role canProducer **or** canConsumer mean that canLookup, but actually in the 
code 
https://github.com/apache/pulsar/blob/master/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java#L267,
 if the method canProduce or canConsume throw exception, `canLookup` will just 
throw the exception and won't check the other permission.
    
    ### Modification
    invoke `canLookupAsync` instead.
    
    (cherry picked from commit 834e2cb78e6354a8e74146f599b20ab2c75af5d9)
---
 .../pulsar/broker/authorization/AuthorizationService.java  | 14 ++++++++++++--
 1 file changed, 12 insertions(+), 2 deletions(-)

diff --git 
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
 
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
index 0ced2bf..10b35ef 100644
--- 
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
+++ 
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
@@ -264,8 +264,18 @@ public class AuthorizationService {
      */
     public boolean canLookup(TopicName topicName, String role, 
AuthenticationDataSource authenticationData)
             throws Exception {
-        return canProduce(topicName, role, authenticationData)
-                || canConsume(topicName, role, authenticationData, null);
+        try {
+            return canLookupAsync(topicName, role, authenticationData)
+                    .get(conf.getZooKeeperOperationTimeoutSeconds(), SECONDS);
+        } catch (InterruptedException e) {
+            log.warn("Time-out {} sec while checking authorization on {} ", 
conf.getZooKeeperOperationTimeoutSeconds(),
+                    topicName);
+            throw e;
+        } catch (Exception e) {
+            log.warn("Role - {} failed to get lookup permissions for topic - 
{}. {}", role, topicName,
+                    e.getMessage());
+            throw e;
+        }
     }
 
     /**