Fix subscription authorization mode

Author: coderzc

mqtt-broker/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/processor/MQTTBrokerProtocolMethodProcessor.java

@@ -393,21 +393,30 @@ public void processSubscribe(MqttAdapterMessage adapter) {
             for (MqttTopicSubscription topic: msg.payload().topicSubscriptions()) {
                 String finalUserRole = userRole;
                 authorizationFutures.add(this.authorizationService.canConsumeAsync(
-                        getEncodedPulsarTopicName(topic.topicName()), userRole, authData, userRole)
+                        getEncodedPulsarTopicName(topic.topicName()), userRole, authData, clientId)
                         .thenAccept((authorized) -> {
                             if (!authorized) {
                                 authorizedFlag.set(false);
                                 log.warn("[Subscribe] no authorization to sub topic={}, userRole={}, CId= {}",
                                         topic.topicName(), finalUserRole, clientId);
                             }
-                        }));
+                }).exceptionally(ex -> {
+                    if (ex != null) {
+                        ex = FutureUtil.unwrapCompletionException(ex);
+                        authorizedFlag.set(false);
+                        log.warn("[Subscribe] failed to authorize sub topic={}, userRole={}, CId= {}, err= {}",
+                            topic.topicName(), finalUserRole, clientId, ex.getMessage());
+                    }
+                    return null;
+                }));
             }
             FutureUtil.waitForAll(authorizationFutures).thenAccept(__ -> {
                 if (!authorizedFlag.get()) {
                     MqttAck subAck = MqttSubAck.errorBuilder(connection.getProtocolVersion())
                             .packetId(packetId)
                             .errorReason(MqttSubAck.ErrorReason.AUTHORIZATION_FAIL)
                             .build();
+                    log.warn("send failed sub ack to clientId={}, subAck={}", clientId, subAck);
                     connection.sendAckThenClose(subAck);
                 } else {
                     doSubscribe(msg);

tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/mqtt3/fusesource/base/AuthorizationTest.java

@@ -21,6 +21,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.admin.GrantTopicPermissionOptions;
 import org.apache.pulsar.common.policies.data.AuthAction;
+import org.apache.pulsar.common.policies.data.SubscriptionAuthMode;
 import org.awaitility.Awaitility;
 import org.fusesource.mqtt.client.BlockingConnection;
 import org.fusesource.mqtt.client.MQTT;
@@ -113,6 +114,58 @@ public void testAuthorizedOnTopic() throws Exception {
         consumer.disconnect();
     }
 
+    @Test(timeOut = TIMEOUT)
+    public void testSubscriptionAuthMode() throws Exception {
+        Set<AuthAction> user1Actions = new HashSet<>();
+        user1Actions.add(AuthAction.produce);
+        admin.namespaces().grantPermissionOnNamespace("public/default", "user1", user1Actions);
+
+        Set<AuthAction> user2Actions = new HashSet<>();
+        user2Actions.add(AuthAction.consume);
+        admin.namespaces().grantPermissionOnNamespace("public/default", "user2", user2Actions);
+
+        admin.namespaces().setSubscriptionAuthMode("public/default", SubscriptionAuthMode.Prefix);
+
+        String topicName = "persistent://public/default/testSubscriptionAuthMode";
+        MQTT mqttConsumer = createMQTTClient();
+        mqttConsumer.setClientId("user2-11");
+        mqttConsumer.setUserName("user2");
+        mqttConsumer.setPassword("pass2");
+        BlockingConnection consumer = mqttConsumer.blockingConnection();
+        consumer.connect();
+        Topic[] topics = {new Topic(topicName, QoS.AT_LEAST_ONCE)};
+        consumer.subscribe(topics);
+
+        MQTT mqttProducer = createMQTTClient();
+        mqttProducer.setUserName("user1");
+        mqttProducer.setPassword("pass1");
+        BlockingConnection producer = mqttProducer.blockingConnection();
+        producer.connect();
+        String message = "Hello MQTT";
+        producer.publish(topicName, message.getBytes(), QoS.AT_MOST_ONCE, false);
+
+        Message receive = consumer.receive();
+        Assert.assertEquals(new String(receive.getPayload()), message);
+        producer.disconnect();
+        consumer.disconnect();
+
+        MQTT mqttConsumer3 = createMQTTClient();
+        mqttConsumer3.setClientId("user3-11");
+        mqttConsumer3.setUserName("user2");
+        mqttConsumer3.setPassword("pass2");
+        mqttConsumer3.setConnectAttemptsMax(0);
+        mqttConsumer3.setReconnectAttemptsMax(0);
+        BlockingConnection connection3 = mqttConsumer3.blockingConnection();
+        connection3.connect();
+        Awaitility.await().untilAsserted(() -> {
+            Assert.assertTrue(connection3.isConnected());
+        });
+        connection3.subscribe(topics);
+        Awaitility.await().untilAsserted(() -> {
+            Assert.assertFalse(connection3.isConnected());
+        });
+    }
+
     @Test
     public void testNotAuthorized() throws Exception {
         Set<AuthAction> user3Actions = new HashSet<>();