Bump mop version to 4.2.1.9

mqtt-broker/pom.xml

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

mqtt-common/pom.xml

@@ -20,7 +20,7 @@
     <parent>
         <artifactId>pulsar-protocol-handler-mqtt-parent</artifactId>
         <groupId>io.streamnative.pulsar.handlers</groupId>
-        <version>4.2.0-SNAPSHOT</version>
+        <version>4.2.1.9</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.2.0-SNAPSHOT</version>
+        <version>4.2.1.9</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.2.0-SNAPSHOT</version>
+    <version>4.2.1.9</version>
     <name>StreamNative :: Pulsar Protocol Handler :: MoP Parent</name>
     <description>Parent for MQTT on Pulsar implemented using Pulsar Protocol Handler.</description>
 
@@ -51,8 +51,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.2.0-SNAPSHOT</pulsar.version>
-        <sn.bom.version>4.2.0-SNAPSHOT</sn.bom.version>
+        <pulsar.version>4.2.1.9</pulsar.version>
+        <sn.bom.version>4.2.1.9</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.2.0-SNAPSHOT</version>
+        <version>4.2.1.9</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>pulsar-protocol-handler-mqtt-tests</artifactId>

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

@@ -14,25 +14,26 @@
 
 package io.streamnative.pulsar.handlers.mqtt.mqtt3.fusesource.proxy;
 
+import static org.awaitility.Awaitility.await;
 import com.google.gson.Gson;
 import io.streamnative.pulsar.handlers.mqtt.base.MQTTTestBase;
 import io.streamnative.pulsar.handlers.mqtt.common.MQTTCommonConfiguration;
-import java.io.BufferedReader;
-import java.io.InputStream;
-import java.io.InputStreamReader;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.http.HttpResponse;
-import org.apache.http.client.HttpClient;
 import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.util.EntityUtils;
 import org.fusesource.mqtt.client.BlockingConnection;
 import org.fusesource.mqtt.client.MQTT;
 import org.fusesource.mqtt.client.QoS;
 import org.testng.Assert;
 import org.testng.annotations.Test;
+
 /**
  * Integration tests for MQTT protocol handler with proxy.
  */
@@ -48,33 +49,36 @@ protected MQTTCommonConfiguration initConfig() throws Exception {
 
     @Test
     public void testGetDeviceList() throws Exception {
-        int index = random.nextInt(mqttProxyPortList.size());
         List<Integer> mqttProxyPortList = getMqttProxyPortList();
         List<Integer> mqttProxyHttpPortList = getMqttProxyHttpPortList();
+        int index = random.nextInt(mqttProxyPortList.size());
         MQTT mqttProducer = new MQTT();
         int port = mqttProxyPortList.get(index);
         String clientId = "device-list-client";
         mqttProducer.setHost("127.0.0.1", port);
         mqttProducer.setClientId(clientId);
         BlockingConnection producer = mqttProducer.blockingConnection();
-        producer.connect();
-        producer.publish("testHttp", "Hello MQTT".getBytes(StandardCharsets.UTF_8), QoS.AT_MOST_ONCE, false);
-        Thread.sleep(4000);
-        HttpClient httpClient = HttpClientBuilder.create().build();
-        final String mopEndPoint = "http://localhost:" + mqttProxyHttpPortList.get(index) + "/admin/devices/list";
-        HttpResponse response = httpClient.execute(new HttpGet(mopEndPoint));
-        InputStream inputStream = response.getEntity().getContent();
-        InputStreamReader isReader = new InputStreamReader(inputStream);
-        BufferedReader reader = new BufferedReader(isReader);
-        StringBuffer buffer = new StringBuffer();
-        String str;
-        while ((str = reader.readLine()) != null){
-            buffer.append(str);
+        try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) {
+            producer.connect();
+            producer.publish("testHttp", "Hello MQTT".getBytes(StandardCharsets.UTF_8), QoS.AT_MOST_ONCE, false);
+            final String mopEndPoint = "http://localhost:" + mqttProxyHttpPortList.get(index) + "/admin/devices/list";
+            await().atMost(30, TimeUnit.SECONDS).pollInterval(1, TimeUnit.SECONDS).untilAsserted(() -> {
+                String ret = getDeviceList(httpClient, mopEndPoint);
+                ArrayList<?> deviceList = new Gson().fromJson(ret, ArrayList.class);
+                Assert.assertNotNull(deviceList, "Invalid device list response: " + ret);
+                Assert.assertEquals(deviceList.size(), 1, "Unexpected device list response: " + ret);
+                Assert.assertTrue(deviceList.contains(clientId), "Unexpected device list response: " + ret);
+            });
+        } finally {
+            producer.disconnect();
         }
-        String ret = buffer.toString();
-        ArrayList deviceList = new Gson().fromJson(ret, ArrayList.class);
-        Assert.assertEquals(deviceList.size(), 1);
-        Assert.assertTrue(deviceList.contains(clientId));
     }
 
+    private String getDeviceList(CloseableHttpClient httpClient, String mopEndPoint) throws Exception {
+        HttpResponse response = httpClient.execute(new HttpGet(mopEndPoint));
+        String body = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
+        Assert.assertEquals(response.getStatusLine().getStatusCode(), 200,
+                "Unexpected HTTP response from " + mopEndPoint + ": " + body);
+        return body;
+    }
 }