Bump project version to 5.0.0-M1-SNAPSHOT

mqtt-broker/pom.xml

@@ -20,7 +20,7 @@
     <parent>
         <artifactId>pulsar-protocol-handler-mqtt-parent</artifactId>
         <groupId>io.streamnative.pulsar.handlers</groupId>
-        <version>4.3.0-SNAPSHOT</version>
+        <version>5.0.0-M1-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>pulsar-protocol-handler-mqtt</artifactId>
@@ -56,4 +56,4 @@
             </plugin>
         </plugins>
     </build>
-</project>
+</project>

mqtt-common/pom.xml

@@ -20,7 +20,7 @@
     <parent>
         <artifactId>pulsar-protocol-handler-mqtt-parent</artifactId>
         <groupId>io.streamnative.pulsar.handlers</groupId>
-        <version>4.3.0-SNAPSHOT</version>
+        <version>5.0.0-M1-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>pulsar-protocol-handler-mqtt-common</artifactId>

mqtt-proxy/pom.xml

@@ -20,7 +20,7 @@
     <parent>
         <artifactId>pulsar-protocol-handler-mqtt-parent</artifactId>
         <groupId>io.streamnative.pulsar.handlers</groupId>
-        <version>4.3.0-SNAPSHOT</version>
+        <version>5.0.0-M1-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>pulsar-protocol-handler-mqtt-proxy</artifactId>

pom.xml

@@ -19,7 +19,7 @@
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <groupId>io.streamnative.pulsar.handlers</groupId>
     <artifactId>pulsar-protocol-handler-mqtt-parent</artifactId>
-    <version>4.3.0-SNAPSHOT</version>
+    <version>5.0.0-M1-SNAPSHOT</version>
     <name>StreamNative :: Pulsar Protocol Handler :: MoP Parent</name>
     <description>Parent for MQTT on Pulsar implemented using Pulsar Protocol Handler.</description>
 
@@ -50,8 +50,8 @@
         <mockito.version>2.22.0</mockito.version>
         <testng.version>6.14.3</testng.version>
         <awaitility.version>4.0.2</awaitility.version>
-        <pulsar.version>4.3.0-SNAPSHOT</pulsar.version>
-        <sn.bom.version>4.3.0-SNAPSHOT</sn.bom.version>
+        <pulsar.version>5.0.0-M1-SNAPSHOT</pulsar.version>
+        <sn.bom.version>5.0.0-M1-SNAPSHOT</sn.bom.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>

tests/pom.xml

@@ -20,7 +20,7 @@
     <parent>
         <artifactId>pulsar-protocol-handler-mqtt-parent</artifactId>
         <groupId>io.streamnative.pulsar.handlers</groupId>
-        <version>4.3.0-SNAPSHOT</version>
+        <version>5.0.0-M1-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>pulsar-protocol-handler-mqtt-tests</artifactId>
@@ -118,4 +118,4 @@
             <url>https://repo.eclipse.org/content/repositories/paho-releases/</url>
         </repository>
     </repositories>
-</project>
+</project>

tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/base/MQTTProtocolHandlerTestBase.java

@@ -278,8 +278,9 @@ protected void restartBroker() throws Exception {
     }
 
     protected void stopBroker() throws Exception {
-        for (PulsarService pulsarService : pulsarServiceList) {
-            pulsarService.close();
+        for (int i = 0; i < pulsarServiceList.size(); i++) {
+            mockZooKeeper.setSessionId(i + 1);
+            pulsarServiceList.get(i).close();
         }
         pulsarServiceList.clear();
         brokerPortList.clear();
@@ -295,6 +296,7 @@ protected void stopBroker() throws Exception {
     }
 
     public void stopBroker(int brokerIndex) throws Exception {
+        mockZooKeeper.setSessionId(brokerIndex + 1);
         pulsarServiceList.get(brokerIndex).close();
         pulsarServiceList.remove(brokerIndex);
         brokerPortList.remove(brokerIndex);
@@ -398,6 +400,7 @@ protected void startBroker(MQTTCommonConfiguration conf) throws Exception {
                 mqttProxyPort, mqttProxyTlsPort, mqttProxyTlsPskPort, mqttHttpPort);
         ConfigurationUtils.extractFieldToProperties(conf);
         setTLSConf(conf);
+        mockZooKeeper.setSessionId(pulsarServiceList.size() + 1);
         this.pulsarServiceList.add(doStartBroker(conf));
     }
 
@@ -424,7 +427,7 @@ protected void setupBrokerMocks(PulsarService pulsar) throws Exception {
 
     public static MockZooKeeper createMockZooKeeper() throws Exception {
         MockZooKeeper zk = MockZooKeeper.newInstance();
-        zk.setSessionId(-1);
+        zk.setSessionId(1);
         List<ACL> dummyAclList = new ArrayList<>(0);
 
         ZkUtils.createFullPathOptimistic(zk, "/ledgers/available/192.168.1.1:" + 5000,

tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/base/MQTTTestBase.java

@@ -22,7 +22,9 @@
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.RetentionPolicies;
 import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.conscrypt.Conscrypt;
 import org.fusesource.mqtt.client.MQTT;
+import org.testng.SkipException;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.DataProvider;
@@ -96,7 +98,6 @@ protected void setup() throws Exception {
         super.setup();
         log.info("success internal setup");
         setupClusterNamespaces();
-        setPulsarServiceState();
     }
 
     protected void setupClusterNamespaces() throws Exception {
@@ -137,6 +138,12 @@ protected void setupClusterNamespaces() throws Exception {
         }
     }
 
+    protected void assumeConscryptAvailable() {
+        if (!Conscrypt.isAvailable()) {
+            throw new SkipException("Conscrypt native library is not available on this platform");
+        }
+    }
+
     @AfterClass(alwaysRun = true)
     @Override
     protected void cleanup() throws Exception {

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

@@ -442,6 +442,7 @@ public void testInvalidClientId() throws Exception {
     @Test
     @SneakyThrows
     public void testTlsPskWithTlsv1() {
+        assumeConscryptAvailable();
         Bootstrap client = new Bootstrap();
         EventLoopGroup group = new NioEventLoopGroup();
         client.group(group);

tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/mqtt3/fusesource/proxy/ProxyTLSTest.java

@@ -156,6 +156,7 @@ public void testSendAndConsumeWithFilter(String topic, String filter) {
     @Test(timeOut = TIMEOUT, priority = 4)
     @SneakyThrows
     public void testTlsPsk() {
+        assumeConscryptAvailable();
         Bootstrap client = new Bootstrap();
         EventLoopGroup group = new NioEventLoopGroup();
         client.group(group);

tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/mqtt3/fusesource/proxy/ProxyTest.java

@@ -260,6 +260,7 @@ public void testSendAndConsumeWithFilter(String topic, String filter) {
     @Test
     @SneakyThrows
     public void testTlsPskWithTlsv1() {
+        assumeConscryptAvailable();
         Bootstrap client = new Bootstrap();
         EventLoopGroup group = new NioEventLoopGroup();
         client.group(group);
@@ -501,6 +502,7 @@ public void testRetainedMessageInCluster() throws Exception {
     @Test
     @SneakyThrows
     public void testAddPskIdentity() {
+        assumeConscryptAvailable();
         HttpClient httpClient = HttpClientBuilder.create().build();
         final String mopEndPoint = "http://localhost:" + brokerWebservicePortList.get(0) + "/mop/add_psk_identity";
         HttpPost request = new HttpPost();