Fix subscription authorization PREFIX mode

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

@@ -393,14 +393,22 @@ 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()) {

pom.xml

@@ -53,13 +53,13 @@
         <awaitility.version>4.0.2</awaitility.version>
         <pulsar.version>4.1.0-SNAPSHOT</pulsar.version>
         <sn.bom.version>4.1.0-SNAPSHOT</sn.bom.version>
-        <mqtt.codec.version>4.1.115.Final</mqtt.codec.version>
         <log4j2.version>2.18.0</log4j2.version>
         <fusesource.client.version>1.16</fusesource.client.version>
         <hivemq.mqtt.client.version>1.2.2</hivemq.mqtt.client.version>
         <apache.commons.bean-utils.version>1.9.4</apache.commons.bean-utils.version>
         <grpc.version>1.45.1</grpc.version>
         <jackson.version>2.14.2</jackson.version>
+        <netty.version>4.1.122.Final</netty.version>
         <!-- plugins -->
         <javac.source>17</javac.source>
         <javac.target>17</javac.target>
@@ -77,11 +77,22 @@
         <dependency>
             <groupId>io.streamnative</groupId>
             <artifactId>pulsar-broker</artifactId>
+            <exclusions>
+                <exclusion>
+                    <artifactId>netty-codec-http2</artifactId>
+                    <groupId>io.netty</groupId>
+                </exclusion>
+            </exclusions>
         </dependency>
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
         </dependency>
+        <dependency>
+            <artifactId>netty-codec-http2</artifactId>
+            <groupId>io.netty</groupId>
+            <version>${netty.version}</version>
+        </dependency>
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-codec-mqtt</artifactId>

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<>();