TestCI bugfix investigation

.github/workflows/temp_maven_build_debug.yml

@@ -0,0 +1,26 @@
+# This workflow is copied from maven_build.yml and is temporarily used to throubleshoot failure that appears only in CI
+
+name: Java CI with Maven
+
+on:
+    push:
+        branches: [ main ]
+    pull_request:
+        branches: [ main ]
+
+jobs:
+    test:
+        runs-on: ubuntu-latest
+        name: Temp test of CI
+        steps:
+            - name: Check out repository
+              uses: actions/checkout@v2
+
+            - name: Set up JDK
+              uses: actions/setup-java@v1
+              with:
+                  java-version: 17
+                  architecture: x64
+            - name: Test with Maven
+              run: ./mvnw -Dtest="SharedSubscriptionTest#givenSharedSubscriptionWithCertainQoSWhenSameClientWithSameShareSubscribeToSameTopicFilterThenQoSUpdates" test -pl broker --file pom.xml
+

broker/pom.xml

@@ -19,7 +19,7 @@
              https://github.com/netty/netty/blob/netty-4.1.100.Final/pom.xml#L650 -->
         <netty.tcnative.version>2.0.61.Final</netty.tcnative.version>
         <paho.version>1.2.5</paho.version>
-        <hivemqclient.version>1.3.0</hivemqclient.version>
+        <hivemqclient.version>1.3.3</hivemqclient.version>
         <h2.version>2.1.212</h2.version>
     </properties>
 

broker/src/main/java/io/moquette/broker/MQTTConnection.java

@@ -690,10 +690,10 @@ void sendPublish(MqttPublishMessage publishMsg) {
         final String topicName = publishMsg.variableHeader().topicName();
         MqttQoS qos = publishMsg.fixedHeader().qosLevel();
         if (LOG.isTraceEnabled()) {
-            LOG.trace("Sending PUBLISH({}) message. MessageId={}, topic={}, payload={}", qos, packetId, topicName,
-                DebugUtils.payload2Str(publishMsg.payload()));
+            LOG.trace("Sending PUBLISH({}) message. MessageId={}, topic={}, payload={} to {}", qos, packetId, topicName,
+                DebugUtils.payload2Str(publishMsg.payload()), getClientId());
         } else {
-            LOG.debug("Sending PUBLISH({}) message. MessageId={}, topic={}", qos, packetId, topicName);
+            LOG.debug("Sending PUBLISH({}) message. MessageId={}, topic={} to {}", qos, packetId, topicName, getClientId());
         }
         sendIfWritableElseDrop(publishMsg);
     }
@@ -703,7 +703,7 @@ void sendIfWritableElseDrop(MqttMessage msg) {
             LOG.debug("OUT {}", msg.fixedHeader().messageType());
         }
         if (channel.isWritable()) {
-
+            LOG.debug("Sending message {} on the wire to {}", msg.fixedHeader().messageType(), getClientId());
             // Sending to external, retain a duplicate. Just retain is not
             // enough, since the receiver must have full control.
             Object retainedDup = msg;
@@ -718,6 +718,8 @@ void sendIfWritableElseDrop(MqttMessage msg) {
                 channelFuture = channel.write(retainedDup);
             }
             channelFuture.addListener(FIRE_EXCEPTION_ON_FAILURE);
+        } else {
+            LOG.debug("Dropping message {} from the wire, msg: {}", msg.fixedHeader().messageType(), msg);
         }
     }
 

broker/src/test/java/io/moquette/integration/mqtt5/SharedSubscriptionTest.java

@@ -33,7 +33,9 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.Properties;
+import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 
 import static io.moquette.integration.mqtt5.ConnectTest.assertConnectionAccepted;
 import static io.moquette.integration.mqtt5.ConnectTest.verifyNoPublish;
@@ -193,6 +195,7 @@ private static void verifyPubPayload(MqttMessage received, String expectedPayloa
         MqttPublishMessage pub = (MqttPublishMessage) received;
         String payload = pub.payload().asByteBuf().toString(StandardCharsets.UTF_8);
         assertEquals(expectedPayload, payload);
+        assertTrue(pub.release(), "received message must be deallocated");
     }
 
     @NotNull
@@ -219,12 +222,12 @@ private Mqtt5BlockingClient createPublisherClient() {
         return client;
     }
 
-    private static void verifyPublishedMessage(Mqtt5BlockingClient subscriber, int timeout, String expectedPayload, String errorMessage) throws InterruptedException {
-        verifyPublishedMessage(subscriber, MqttQos.AT_MOST_ONCE, expectedPayload, errorMessage, timeout);
+    private static void verifyPublishedMessage(Mqtt5BlockingClient client, int timeout, String expectedPayload, String errorMessage) throws InterruptedException {
+        verifyPublishedMessage(client, MqttQos.AT_MOST_ONCE, expectedPayload, errorMessage, timeout);
     }
 
-    private static void verifyPublishedMessage(Mqtt5BlockingClient subscriber, MqttQos expectedQos, String expectedPayload, String errorMessage, int timeoutSeconds) throws InterruptedException {
-        try (Mqtt5BlockingClient.Mqtt5Publishes publishes = subscriber.publishes(MqttGlobalPublishFilter.ALL)) {
+    private static void verifyPublishedMessage(Mqtt5BlockingClient client, MqttQos expectedQos, String expectedPayload, String errorMessage, int timeoutSeconds) throws InterruptedException {
+        try (Mqtt5BlockingClient.Mqtt5Publishes publishes = client.publishes(MqttGlobalPublishFilter.ALL)) {
             Optional<Mqtt5Publish> publishMessage = publishes.receive(timeoutSeconds, TimeUnit.SECONDS);
             if (!publishMessage.isPresent()) {
                 fail("Expected to receive a publish message");
@@ -233,7 +236,22 @@ private static void verifyPublishedMessage(Mqtt5BlockingClient subscriber, MqttQ
             Mqtt5Publish msgPub = publishMessage.get();
             final String payload = new String(msgPub.getPayloadAsBytes(), StandardCharsets.UTF_8);
             assertEquals(expectedPayload, payload, errorMessage);
+            assertEquals(expectedQos, msgPub.getQos());
+        }
+    }
 
+    private static void verifyPublishedMessage(Mqtt5BlockingClient client, Consumer<Void> action, MqttQos expectedQos,
+                                               String expectedPayload, String errorMessage, int timeoutSeconds) throws Exception {
+        try (Mqtt5BlockingClient.Mqtt5Publishes publishes = client.publishes(MqttGlobalPublishFilter.ALL)) {
+            action.accept(null);
+            Optional<Mqtt5Publish> publishMessage = publishes.receive(timeoutSeconds, TimeUnit.SECONDS);
+            if (!publishMessage.isPresent()) {
+                fail("Expected to receive a publish message");
+                return;
+            }
+            Mqtt5Publish msgPub = publishMessage.get();
+            final String payload = new String(msgPub.getPayloadAsBytes(), StandardCharsets.UTF_8);
+            assertEquals(expectedPayload, payload, errorMessage);
             assertEquals(expectedQos, msgPub.getQos());
         }
     }
@@ -269,7 +287,7 @@ public void whenAClientSubscribeToASharedTopicThenDoesntReceiveAnyRetainedMessag
     }
 
     @Test
-    public void givenSharedSubscriptionWithCertainQoSWhenSameClientWithSameShareSubscribeToSameTopicFilterThenQoSUpdates() throws InterruptedException {
+    public void givenSharedSubscriptionWithCertainQoSWhenSameClientWithSameShareSubscribeToSameTopicFilterThenQoSUpdates() throws Exception {
         final Mqtt5BlockingClient subscriberClient = createSubscriberClient();
         subscribe(subscriberClient, "$share/collectors/metric/temperature/living", MqttQos.AT_MOST_ONCE);
 
@@ -279,14 +297,16 @@ public void givenSharedSubscriptionWithCertainQoSWhenSameClientWithSameShareSubs
         // because the PUB is at QoS1 and the subscription is at QoS0, the subscribed doesn't receive any message
         verifyNoPublish(subscriberClient,Duration.ofSeconds(1), "No message is expected at Qos0 subscription");
 
+        LOG.info("Before repeating with AT_LEAST_ONCE");
+
         // update QoS for shared subscription
         subscribe(subscriberClient, "$share/collectors/metric/temperature/living", MqttQos.AT_LEAST_ONCE);
 
-        // publish the message again
-        publish(publisherClient, "metric/temperature/living", MqttQos.AT_LEAST_ONCE);
-
         // This time the publish reaches the subscription
-        verifyPublishedMessage(subscriberClient, MqttQos.AT_LEAST_ONCE, "18", "Shared message must be received", 30);
+        verifyPublishedMessage(subscriberClient, v -> {
+            // publish the message again
+            publish(publisherClient, "metric/temperature/living", MqttQos.AT_LEAST_ONCE);
+        }, MqttQos.AT_LEAST_ONCE, "18", "Shared message must be received", 30);
     }
 
     private static void publish(Mqtt5BlockingClient publisherClient, String topicName, MqttQos mqttQos) {

broker/src/test/resources/log4j.properties

@@ -1,7 +1,8 @@
 #log4j.rootLogger=ERROR, stdout, file
 log4j.rootLogger=ERROR, stdout
 
-log4j.logger.io.moquette=WARN
+log4j.logger.io.moquette=DEBUG
+log4j.logger.com.hivemq.client=DEBUG
 #log4j.logger.io.moquette.broker=DEBUG
 #log4j.logger.io.moquette.broker.MQTTConnection=DEBUG
 #log4j.logger.io.moquette.broker.SessionRegistry=DEBUG