Repository: moquette-io/moquette
Pull request #657 (2022-02-18 18:01:34)

Updates the command enqueuing to Session Event Loop to return the status of the operation
## Release notes
[rn:skip]

## What does this PR do?

Fix a design problem in the enqueuing of commands to Session Event loops.
In the first version the of the process publish methods, the `receivedPublish*` operations returned a `CompletableFuture` which wrapped the logic of the message processing. In case of validation error, was returned an already completed future.
For example the `receivedPublishQos1` returned  a future that was obtained as composition of the action to publish to subscribers, concatenated with the logic to send the PUBACK to the publisher. So it happened that the PUBACK was sent when [publish was forwarded to the matching subscribers](https://github.com/moquette-io/moquette/blob/15d5494975cf8f173a829733d60a0add93d51c33/broker/src/main/java/io/moquette/broker/PostOffice.java#L349).
The problem is that the operation to enqueue could be successful or fail, and in case of success there is a `CompleteableFuture` that rappresents the part of the forwarding to subscribers.
The PUBACK message should be sent when the message is taken into account from the broker and not when the publish has completely forwarded to all subscribers.

This PR changes the `receivedPublishQos0` `receivedPublishQos1` `receivedPublishQos2` to return a `RouteResult` object that contains the status of the enqueuing operation plus the future of the command task.

It touches a lot of parts of the code due to signature changes, but the core changes are all in `PostOffice`.


## Why is it important/What is the impact to the user?

The final user is not impated by this, the developer has a more straightforward and clean process logic

## Checklist

- [x] My code follows the style guidelines of this project
- [x] I have commented my code, particularly in hard-to-understand areas
- ~~[ ] I have made corresponding changes to the documentation~~
- ~~[ ] I have made corresponding change to the default configuration files (and/or docker env variables)~~
- [x] I have added tests that prove my fix is effective or that my feature works


## Related issues

<!-- Recommended
Link related issues below. Insert the issue link or reference after the word "Closes" if merging this should automatically close it.

- Closes #123
- Relates #123
- Requires #123
- Superseeds #123
-->
- Related #631







hylkevds: :+1:  That makes things a lot easier to follow!
andsel: Hi @hylkevds may I ask you for a review of this PR?
I know it's big, but the core part of the changes is in `PostOffice` class. Let me know :-)
hylkevds: > Hi @hylkevds may I ask you for a review of this PR?
> I know it's big, but the core part of the changes is in `PostOffice` class. Let me know :-)

Of course, no problem.
hylkevds: (Review in progress...)
Found one leak, when a publish fails because a Session command queue is full (at least for QoS 2). I think it is either because
- `MQTTConnection.processPublish` does a retain, but when the queue is full the release in `PostOffice.receivedPublishQos2` never runs.
- `PostOffice.publish2Subscribers:499` does a retain for each relevant queue, but the command lambda never executes in `PostOffice.publish2Subscribers:503`

Or maybe it's both reasons.

Maybe we can change `Server.internalPublish` for embedded applications to return the `RoutingResults` too, that way the embedded application can check if the publish was OK, and re-try if it wants to.
andsel: Hi @hylkevds thanks a ton to have spotted all those missed `release`.
Have fixed and aligned to `main`, now thi s PR is ready for a second run of review :-)
andsel: @hylkevds wait before review, because the test `testClientDoesntRemainSubscribedAfterASubscriptionAndServerRestart` started to hangup

Patch:
diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnection.java b/broker/src/main/java/io/moquette/broker/MQTTConnection.java
index f3ab7611..a12e0f54 100644
--- a/broker/src/main/java/io/moquette/broker/MQTTConnection.java
+++ b/broker/src/main/java/io/moquette/broker/MQTTConnection.java
@@ -25,14 +25,12 @@ import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelPipeline;
 import io.netty.handler.codec.mqtt.*;
 import io.netty.handler.timeout.IdleStateHandler;
-import io.netty.util.ReferenceCountUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.net.InetSocketAddress;
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -110,7 +108,7 @@ final class MQTTConnection {
 
     private void processPubComp(MqttMessage msg) {
         final int messageID = ((MqttMessageIdVariableHeader) msg.variableHeader()).messageId();
-        this.postOffice.routeCommand(bindedSession.getClientID(), () -> {
+        this.postOffice.routeCommand(bindedSession.getClientID(), "PUBCOMP", () -> {
             bindedSession.processPubComp(messageID);
             return bindedSession.getClientID();
         });
@@ -118,7 +116,7 @@ final class MQTTConnection {
 
     private void processPubRec(MqttMessage msg) {
         final int messageID = ((MqttMessageIdVariableHeader) msg.variableHeader()).messageId();
-        this.postOffice.routeCommand(bindedSession.getClientID(), () -> {
+        this.postOffice.routeCommand(bindedSession.getClientID(), "PUBREC", () -> {
             bindedSession.processPubRec(messageID);
             return null;
         });
@@ -132,13 +130,13 @@ final class MQTTConnection {
     private void processPubAck(MqttMessage msg) {
         final int messageID = ((MqttMessageIdVariableHeader) msg.variableHeader()).messageId();
         final String clientId = getClientId();
-        this.postOffice.routeCommand(clientId, () -> {
+        this.postOffice.routeCommand(clientId, "PUB ACK", () -> {
             bindedSession.pubAckReceived(messageID);
             return null;
         });
     }
 
-    Future<PostOffice.RouteResult> processConnect(MqttConnectMessage msg) {
+    PostOffice.RouteResult processConnect(MqttConnectMessage msg) {
         MqttConnectPayload payload = msg.payload();
         String clientId = payload.clientIdentifier();
         final String username = payload.userName();
@@ -147,20 +145,20 @@ final class MQTTConnection {
         if (isNotProtocolVersion(msg, MqttVersion.MQTT_3_1) && isNotProtocolVersion(msg, MqttVersion.MQTT_3_1_1)) {
             LOG.warn("MQTT protocol version is not valid. CId: {}", clientId);
             abortConnection(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION);
-            return CompletableFuture.completedFuture(null);
+            return PostOffice.RouteResult.failed(clientId);
         }
         final boolean cleanSession = msg.variableHeader().isCleanSession();
         if (clientId == null || clientId.length() == 0) {
             if (!brokerConfig.isAllowZeroByteClientId()) {
                 LOG.info("Broker doesn't permit MQTT empty client ID. Username: {}", username);
                 abortConnection(CONNECTION_REFUSED_IDENTIFIER_REJECTED);
-                return CompletableFuture.completedFuture(null);
+                return PostOffice.RouteResult.failed(clientId);
             }
 
             if (!cleanSession) {
                 LOG.info("MQTT client ID cannot be empty for persistent session. Username: {}", username);
                 abortConnection(CONNECTION_REFUSED_IDENTIFIER_REJECTED);
-                return CompletableFuture.completedFuture(null);
+                return PostOffice.RouteResult.failed(clientId);
             }
 
             // Generating client id.
@@ -171,11 +169,11 @@ final class MQTTConnection {
         if (!login(msg, clientId)) {
             abortConnection(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
             channel.close().addListener(CLOSE_ON_FAILURE);
-            return CompletableFuture.completedFuture(null);
+            return PostOffice.RouteResult.failed(clientId);
         }
 
         final String sessionId = clientId;
-        return this.postOffice.routeCommand(clientId, () -> {
+        return postOffice.routeCommand(clientId, "CONN", () -> {
             executeConnect(msg, sessionId);
             return null;
         });
@@ -305,7 +303,7 @@ final class MQTTConnection {
         }
         // this must not be done on the netty thread
         LOG.debug("Notifying connection lost event");
-        postOffice.routeCommand(clientID, () -> {
+        postOffice.routeCommand(clientID, "CONN LOST", () -> {
             if (isBoundToSession() || isSessionUnbound()) {
                 LOG.debug("Cleaning {}", clientID);
                 processConnectionLost(clientID);
@@ -341,15 +339,15 @@ final class MQTTConnection {
         channel.close().addListener(FIRE_EXCEPTION_ON_FAILURE);
     }
 
-    Future<PostOffice.RouteResult> processDisconnect(MqttMessage msg) {
+    PostOffice.RouteResult processDisconnect(MqttMessage msg) {
         final String clientID = NettyUtils.clientID(channel);
         LOG.trace("Start DISCONNECT");
         if (!connected) {
             LOG.info("DISCONNECT received on already closed connection");
-            return CompletableFuture.completedFuture(null);
+            return PostOffice.RouteResult.success(clientID, CompletableFuture.completedFuture(null));
         }
 
-        return this.postOffice.routeCommand(clientID, () -> {
+        return this.postOffice.routeCommand(clientID, "DISCONN", () -> {
             if (!isBoundToSession()) {
                 LOG.debug("NOT processing disconnect {}, not bound.", clientID);
                 return null;
@@ -364,15 +362,15 @@ final class MQTTConnection {
         });
     }
 
-    Future<PostOffice.RouteResult> processSubscribe(MqttSubscribeMessage msg) {
+    PostOffice.RouteResult processSubscribe(MqttSubscribeMessage msg) {
         final String clientID = NettyUtils.clientID(channel);
         if (!connected) {
             LOG.warn("SUBSCRIBE received on already closed connection");
             dropConnection();
-            return CompletableFuture.completedFuture(null);
+            return PostOffice.RouteResult.success(clientID, CompletableFuture.completedFuture(null));
         }
         final String username = NettyUtils.userName(channel);
-        return postOffice.routeCommand(clientID, () -> {
+        return postOffice.routeCommand(clientID, "SUB", () -> {
             if (isBoundToSession())
                 postOffice.subscribeClientToTopics(msg, clientID, username, this);
             return null;
@@ -389,7 +387,7 @@ final class MQTTConnection {
         final String clientID = NettyUtils.clientID(channel);
         final int messageId = msg.variableHeader().messageId();
 
-        postOffice.routeCommand(clientID, () -> {
+        postOffice.routeCommand(clientID, "UNSUB", () -> {
             if (!isBoundToSession())
                 return null;
             LOG.trace("Processing UNSUBSCRIBE message. topics: {}", topics);
@@ -408,7 +406,7 @@ final class MQTTConnection {
         LOG.trace("Client unsubscribed from topics <{}>", topics);
     }
 
-    Future<? extends Object> processPublish(MqttPublishMessage msg) {
+    PostOffice.RouteResult processPublish(MqttPublishMessage msg) {
         final MqttQoS qos = msg.fixedHeader().qosLevel();
         final String username = NettyUtils.userName(channel);
         final String topicName = msg.variableHeader().topicName();
@@ -426,33 +424,39 @@ final class MQTTConnection {
         msg.retain();
         switch (qos) {
             case AT_MOST_ONCE:
-                return postOffice.routeCommand(clientId, () -> {
+                return postOffice.routeCommand(clientId, "PUB QoS0", () -> {
                     if (!isBoundToSession())
                         return null;
                     postOffice.receivedPublishQos0(topic, username, clientId, msg);
                     return null;
-                });
+                }).ifFailed(msg::release);
             case AT_LEAST_ONCE:
-                return postOffice.routeCommand(clientId, () -> {
+                return postOffice.routeCommand(clientId, "PUB QoS1", () -> {
                     if (!isBoundToSession())
                         return null;
                     postOffice.receivedPublishQos1(this, topic, username, messageID, msg);
                     return null;
-                });
+                }).ifFailed(msg::release);
             case EXACTLY_ONCE: {
-                final CompletableFuture<PostOffice.RouteResult> firstStepFuture = postOffice.routeCommand(clientId, () -> {
+                final PostOffice.RouteResult firstStepResult = postOffice.routeCommand(clientId, "PUB QoS2", () -> {
                     if (!isBoundToSession())
                         return null;
                     bindedSession.receivedPublishQos2(messageID, msg);
                     return null;
                 });
-                return firstStepFuture.thenCompose(v -> postOffice.receivedPublishQos2(this, msg, username));
+                if (!firstStepResult.isSuccess()) {
+                    msg.release();
+                    LOG.trace("Failed to enqueue PUB QoS2 to session loop for  {}", clientId);
+                    return firstStepResult;
+                }
+                firstStepResult.completableFuture().thenRun(() ->
+                    postOffice.receivedPublishQos2(this, msg, username).completableFuture()
+                );
+                return firstStepResult;
             }
             default:
                 LOG.error("Unknown QoS-Type:{}", qos);
-                final CompletableFuture<Void> failed = new CompletableFuture<>();
-                failed.completeExceptionally(new Error("Unknown QoS-"));
-                return failed;
+                return PostOffice.RouteResult.failed(clientId, "Unknown QoS-");
         }
     }
 
@@ -466,7 +470,7 @@ final class MQTTConnection {
 
     private void processPubRel(MqttMessage msg) {
         final int messageID = ((MqttMessageIdVariableHeader) msg.variableHeader()).messageId();
-        this.postOffice.routeCommand(bindedSession.getClientID(), () -> {
+        postOffice.routeCommand(bindedSession.getClientID(), "PUBREL", () -> {
             executePubRel(messageID);
             return null;
         });
@@ -480,7 +484,6 @@ final class MQTTConnection {
     void sendPublish(MqttPublishMessage publishMsg) {
         final int packetId = publishMsg.variableHeader().packetId();
         final String topicName = publishMsg.variableHeader().topicName();
-        final String clientId = getClientId();
         MqttQoS qos = publishMsg.fixedHeader().qosLevel();
         if (LOG.isTraceEnabled()) {
             LOG.trace("Sending PUBLISH({}) message. MessageId={}, topic={}, payload={}", qos, packetId, topicName,
diff --git a/broker/src/main/java/io/moquette/broker/PostOffice.java b/broker/src/main/java/io/moquette/broker/PostOffice.java
index d4278852..da640489 100644
--- a/broker/src/main/java/io/moquette/broker/PostOffice.java
+++ b/broker/src/main/java/io/moquette/broker/PostOffice.java
@@ -126,34 +126,84 @@ class PostOffice {
 
         private final List<String> successedRoutings;
         private final List<String> failedRoutings;
+        private final CompletableFuture<Void> mergedAction;
 
-        public RoutingResults(List<String> successedRoutings, List<String> failedRoutings) {
+        public RoutingResults(List<String> successedRoutings, List<String> failedRoutings, CompletableFuture<Void> mergedAction) {
             this.successedRoutings = successedRoutings;
             this.failedRoutings = failedRoutings;
+            this.mergedAction = mergedAction;
         }
 
         public boolean isAllSuccess() {
             return failedRoutings.isEmpty();
         }
+
+        public boolean isAllFailed() {
+            return successedRoutings.isEmpty() && !failedRoutings.isEmpty();
+        }
+
+        public CompletableFuture<Void> completableFuture() {
+            return mergedAction;
+        }
+
+        public static RoutingResults preroutingError() {
+            // WARN this is a special case failed is empty, but this result is to be considered as error.
+            return new RoutingResults(Collections.emptyList(), Collections.emptyList(), CompletableFuture.completedFuture(null));
+        }
+
+        @Override
+        public String toString() {
+            return "RoutingResults{" +
+                "successedRoutings=" + successedRoutings +
+                ", failedRoutings=" + failedRoutings +
+                ", mergedAction=" + mergedAction +
+                '}';
+        }
     }
 
     static class RouteResult {
         private final String clientId;
         private final Status status;
+        private CompletableFuture queuedFuture;
 
         enum Status {SUCCESS, FAIL}
 
-        public static RouteResult success(String clientId) {
-            return new RouteResult(clientId, Status.SUCCESS);
+        public static RouteResult success(String clientId, CompletableFuture queuedFuture) {
+            return new RouteResult(clientId, Status.SUCCESS, queuedFuture);
         }
 
         public static RouteResult failed(String clientId) {
-            return new RouteResult(clientId, Status.FAIL);
+            return failed(clientId, null);
         }
 
-        private RouteResult(String clientId, RouteResult.Status status) {
+        public static RouteResult failed(String clientId, String error) {
+           final CompletableFuture<Void> failed = new CompletableFuture<>();
+            failed.completeExceptionally(new Error(error));
+            return new RouteResult(clientId, Status.FAIL, failed);
+        }
+
+        private RouteResult(String clientId, Status status, CompletableFuture queuedFuture) {
             this.clientId = clientId;
             this.status = status;
+            this.queuedFuture = queuedFuture;
+        }
+
+        public CompletableFuture completableFuture() {
+            if (status == Status.FAIL) {
+                throw new IllegalArgumentException("Accessing completable future on a failed result");
+            }
+            return queuedFuture;
+        }
+
+        public boolean isSuccess() {
+            return status == Status.SUCCESS;
+        }
+
+        public RouteResult ifFailed(Runnable action) {
+            if (!isSuccess()) {
+                action.run();
+            }
+            return this;
         }
     }
 
@@ -307,11 +357,17 @@ class PostOffice {
     CompletableFuture<Void> receivedPublishQos0(Topic topic, String username, String clientID, MqttPublishMessage msg) {
         if (!authorizator.canWrite(topic, username, clientID)) {
             LOG.error("client is not authorized to publish on topic: {}", topic);
+            ReferenceCountUtil.release(msg);
+            return CompletableFuture.completedFuture(null);
+        }
+        final RoutingResults publishResult = publish2Subscribers(msg.payload(), topic, AT_MOST_ONCE);
+        if (publishResult.isAllFailed()) {
+            LOG.info("No one publish was successfully enqueued to session loops");
+            ReferenceCountUtil.release(msg);
             return CompletableFuture.completedFuture(null);
         }
-        final CompletableFuture<RoutingResults> publishFuture = publish2Subscribers(msg.payload(), topic, AT_MOST_ONCE);
 
-        return publishFuture.thenRun(() -> {
+        return publishResult.completableFuture().thenRun(() -> {
             if (msg.fixedHeader().isRetain()) {
                 // QoS == 0 && retain => clean old retained
                 retainedRepository.cleanRetained(topic);
@@ -322,45 +378,49 @@ class PostOffice {
         });
     }
 
-    CompletableFuture<RoutingResults> receivedPublishQos1(MQTTConnection connection, Topic topic, String username, int messageID,
+    RoutingResults receivedPublishQos1(MQTTConnection connection, Topic topic, String username, int messageID,
                                                 MqttPublishMessage msg) {
         // verify if topic can be written
         topic.getTokens();
         if (!topic.isValid()) {
             LOG.warn("Invalid topic format, force close the connection");
             connection.dropConnection();
-            return CompletableFuture.completedFuture(null);
+            ReferenceCountUtil.release(msg);
+            return RoutingResults.preroutingError();
         }
         final String clientId = connection.getClientId();
         if (!authorizator.canWrite(topic, username, clientId)) {
             LOG.error("MQTT client: {} is not authorized to publish on topic: {}", clientId, topic);
-            return CompletableFuture.completedFuture(null);
+            ReferenceCountUtil.release(msg);
+            return RoutingResults.preroutingError();
         }
 
         ByteBuf payload = msg.payload();
-        final CompletableFuture<RoutingResults> publishFuture;
+        final RoutingResults routes;
         if (msg.fixedHeader().isDup()) {
             final Set<String> failedClients = failedPublishes.listFailed(clientId, messageID);
-            publishFuture = publish2Subscribers(payload, topic, AT_LEAST_ONCE, failedClients);
+            routes = publish2Subscribers(payload, topic, AT_LEAST_ONCE, failedClients);
+        } else {
+            routes = publish2Subscribers(payload, topic, AT_LEAST_ONCE);
+        }
+        if (LOG.isTraceEnabled()) {
+            LOG.trace("subscriber routes: {}", routes);
+        }
+        if (routes.isAllSuccess()) {
+            // QoS1 message was enqueued successfully to every event loop
+            connection.sendPubAck(messageID);
+            manageRetain(topic, msg);
+            interceptor.notifyTopicPublished(msg, clientId, username);
         } else {
-            publishFuture = publish2Subscribers(payload, topic, AT_LEAST_ONCE);
+            // some session event loop enqueue raised a problem
+            failedPublishes.insertAll(messageID, clientId, routes.failedRoutings);
         }
+        ReferenceCountUtil.release(msg);
 
-        return publishFuture.whenComplete((RoutingResults routings, Throwable err) -> {
-            if (routings.isAllSuccess()) {
-                // QoS1 message was enqueued successfully to every event loop
-                connection.sendPubAck(messageID);
-                manageRetain(topic, msg);
-                interceptor.notifyTopicPublished(msg, clientId, username);
-            } else {
-                // some session event loop enqueue raised a problem
-                failedPublishes.insertAll(messageID, clientId, routings.failedRoutings);
-            }
-            ReferenceCountUtil.release(msg);
+        // cleanup success resends from the failed publishes cache
+        failedPublishes.removeAll(messageID, clientId, routes.successedRoutings);
 
-            // cleanup success resends from the failed publishes cache
-            failedPublishes.removeAll(messageID, clientId, routings.successedRoutings);
-        });
+        return routes;
     }
 
     private void manageRetain(Topic topic, MqttPublishMessage msg) {
@@ -374,7 +434,7 @@ class PostOffice {
         }
     }
 
-    private CompletableFuture<RoutingResults> publish2Subscribers(ByteBuf payload, Topic topic, MqttQoS publishingQos) {
+    private RoutingResults publish2Subscribers(ByteBuf payload, Topic topic, MqttQoS publishingQos) {
         return publish2Subscribers(payload, topic, publishingQos, NO_FILTER);
     }
 
@@ -399,8 +459,8 @@ class PostOffice {
             return Math.abs(clientId.hashCode()) % this.eventLoops;
         }
 
-        List<CompletableFuture<RouteResult>> routeBatchedPublishes(Consumer<List<Subscription>> action) {
-            List<CompletableFuture<RouteResult>> publishFutures = new ArrayList<>(this.eventLoops);
+        List<RouteResult> routeBatchedPublishes(Consumer<List<Subscription>> action) {
+            List<RouteResult> publishResults = new ArrayList<>(this.eventLoops);
 
             for (List<Subscription> subscriptionsBatch : subscriptions) {
                 if (subscriptionsBatch == null) {
@@ -414,23 +474,39 @@ class PostOffice {
                     final int loopId = subscriberEventLoop(clientId);
                     LOG.trace("Routing PUBLISH to eventLoop {}  for subscriptions [{}]", loopId, subscriptionsDetails);
                 }
-                publishFutures.add(routeCommand(clientId, () -> {
+                publishResults.add(routeCommand(clientId, "batched PUB", () -> {
                     action.accept(subscriptionsBatch);
                     return null;
                 }));
             }
-            return publishFutures;
+            return publishResults;
         }
 
         Collection<String> subscriberIdsByEventLoop(String clientId) {
             final int targetQueueId = subscriberEventLoop(clientId);
             return subscriptions[targetQueueId].stream().map(Subscription::getClientId).collect(Collectors.toList());
         }
+
+        public int countBatches() {
+            int count = 0;
+            for (List<Subscription> subscriptionsBatch : subscriptions) {
+                if (subscriptionsBatch == null) {
+                    continue;
+                }
+                count ++;
+            }
+            return count;
+        }
     }
 
-    private CompletableFuture<RoutingResults> publish2Subscribers(ByteBuf payload, Topic topic, MqttQoS publishingQos,
-                                                                  Set<String> filterTargetClients) {
+    private RoutingResults publish2Subscribers(ByteBuf payload, Topic topic, MqttQoS publishingQos,
+                                               Set<String> filterTargetClients) {
         Set<Subscription> topicMatchingSubscriptions = subscriptions.matchQosSharpening(topic);
+        if (topicMatchingSubscriptions.isEmpty()) {
+            // no matching subscriptions, clean exit
+            LOG.trace("No matching subscriptions for topic: {}", topic);
+            return new RoutingResults(Collections.emptyList(), Collections.emptyList(), CompletableFuture.completedFuture(null));
+        }
 
         final BatchingPublishesCollector collector = new BatchingPublishesCollector(eventLoops);
 
@@ -439,26 +515,30 @@ class PostOffice {
                 collector.add(sub);
             }
         }
+        payload.retain(collector.countBatches());
 
-        List<CompletableFuture<RouteResult>> publishFutures = collector.routeBatchedPublishes((batch) -> {
+        List<RouteResult> publishResults = collector.routeBatchedPublishes((batch) -> {
             publishToSession(payload, topic, batch, publishingQos);
+            payload.release();
         });
 
-        final CompletableFuture<Void> publishes = CompletableFuture.allOf(publishFutures.toArray(new CompletableFuture[0]));
-        return publishes.handle((result, exception) -> {
-            final List<String> failedRoutings = new ArrayList<>();
-            final List<String> successedRoutings = new ArrayList<>();
-            for (CompletableFuture<RouteResult> cf : publishFutures) {
-                RouteResult rr = cf.join();
-                Collection<String> subscibersIds = collector.subscriberIdsByEventLoop(rr.clientId);
-                if (rr.status == RouteResult.Status.FAIL) {
-                    failedRoutings.addAll(subscibersIds);
-                } else {
-                    successedRoutings.addAll(subscibersIds);
-                }
+        final CompletableFuture[] publishFutures = publishResults.stream()
+            .filter(RouteResult::isSuccess)
+            .map(RouteResult::completableFuture).toArray(CompletableFuture[]::new);
+        final CompletableFuture<Void> publishes = CompletableFuture.allOf(publishFutures);
+
+        final List<String> failedRoutings = new ArrayList<>();
+        final List<String> successedRoutings = new ArrayList<>();
+        for (RouteResult rr : publishResults) {
+            Collection<String> subscibersIds = collector.subscriberIdsByEventLoop(rr.clientId);
+            if (rr.status == RouteResult.Status.FAIL) {
+                failedRoutings.addAll(subscibersIds);
+                payload.release();
+            } else {
+                successedRoutings.addAll(subscibersIds);
             }
-            return new RoutingResults(successedRoutings, failedRoutings);
-        });
+        }
+        return new RoutingResults(successedRoutings, failedRoutings, publishes);
     }
 
     private void publishToSession(ByteBuf payload, Topic topic, Collection<Subscription> subscriptions, MqttQoS publishingQos) {
@@ -489,7 +569,7 @@ class PostOffice {
      * subscribers.
      * @return
      */
-    CompletableFuture<RoutingResults> receivedPublishQos2(MQTTConnection connection, MqttPublishMessage msg, String username) {
+    RoutingResults receivedPublishQos2(MQTTConnection connection, MqttPublishMessage msg, String username) {
         LOG.trace("Processing PUB QoS2 message on connection: {}", connection);
         final Topic topic = new Topic(msg.variableHeader().topicName());
         final ByteBuf payload = msg.payload();
@@ -497,33 +577,34 @@ class PostOffice {
         final String clientId = connection.getClientId();
         if (!authorizator.canWrite(topic, username, clientId)) {
             LOG.error("MQTT client is not authorized to publish on topic: {}", topic);
-            return CompletableFuture.completedFuture(null);
+            ReferenceCountUtil.release(msg);
+            // WARN this is a special case failed is empty, but this result is to be considered as error.
+            return RoutingResults.preroutingError();
         }
 
         final int messageID = msg.variableHeader().packetId();
-        final CompletableFuture<RoutingResults> publishFuture;
+        final RoutingResults publishRoutings;
         if (msg.fixedHeader().isDup()) {
             final Set<String> failedClients = failedPublishes.listFailed(clientId, messageID);
-            publishFuture = publish2Subscribers(payload, topic, EXACTLY_ONCE, failedClients);
+            publishRoutings = publish2Subscribers(payload, topic, EXACTLY_ONCE, failedClients);
+        } else {
+            publishRoutings = publish2Subscribers(payload, topic, EXACTLY_ONCE);
+        }
+        if (publishRoutings.isAllSuccess()) {
+            // QoS2 PUB message was enqueued successfully to every event loop
+            connection.sendPubRec(messageID);
+            manageRetain(topic, msg);
+            interceptor.notifyTopicPublished(msg, clientId, username);
         } else {
-            publishFuture = publish2Subscribers(payload, topic, EXACTLY_ONCE);
+            // some session event loop enqueue raised a problem
+            failedPublishes.insertAll(messageID, clientId, publishRoutings.failedRoutings);
         }
+        ReferenceCountUtil.release(msg);
 
-        return publishFuture.whenComplete((RoutingResults routings, Throwable err) -> {
-            if (routings.isAllSuccess()) {
-                // QoS2 PUB message was enqueued successfully to every event loop
-                connection.sendPubRec(messageID);
-                manageRetain(topic, msg);
-                interceptor.notifyTopicPublished(msg, clientId, username);
-            } else {
-                // some session event loop enqueue raised a problem
-                failedPublishes.insertAll(messageID, clientId, routings.failedRoutings);
-            }
-            ReferenceCountUtil.release(msg);
+        // cleanup success resends from the failed publishes cache
+        failedPublishes.removeAll(messageID, clientId, publishRoutings.successedRoutings);
 
-            // cleanup success resends from the failed publishes cache
-            failedPublishes.removeAll(messageID, clientId, routings.successedRoutings);
-        });
+        return publishRoutings;
     }
 
     static MqttQoS lowerQosToTheSubscriptionDesired(Subscription sub, MqttQoS qos) {
@@ -543,25 +624,27 @@ class PostOffice {
      * @param msg
      *            the message to publish
      * @return
+     *            the result of the enqueuing operation to session loops.
      */
-    public CompletableFuture<Void> internalPublish(MqttPublishMessage msg) {
+    public RoutingResults internalPublish(MqttPublishMessage msg) {
         final MqttQoS qos = msg.fixedHeader().qosLevel();
         final Topic topic = new Topic(msg.variableHeader().topicName());
         final ByteBuf payload = msg.payload();
         LOG.info("Sending internal PUBLISH message Topic={}, qos={}", topic, qos);
 
-        final CompletableFuture<RoutingResults> publishFuture = publish2Subscribers(payload, topic, qos);
-        return publishFuture.thenRun(() -> {
-            if (!msg.fixedHeader().isRetain()) {
-                return;
-            }
-            if (qos == AT_MOST_ONCE || payload.readableBytes() == 0) {
-                // QoS == 0 && retain => clean old retained
-                retainedRepository.cleanRetained(topic);
-                return;
-            }
-            retainedRepository.retain(topic, msg);
-        });
+        final RoutingResults publishResult = publish2Subscribers(payload, topic, qos);
+        LOG.trace("after routed publishes: {}", publishResult);
+
+        if (!msg.fixedHeader().isRetain()) {
+            return publishResult;
+        }
+        if (qos == AT_MOST_ONCE || payload.readableBytes() == 0) {
+            // QoS == 0 && retain => clean old retained
+            retainedRepository.cleanRetained(topic);
+            return publishResult;
+        }
+        retainedRepository.retain(topic, msg);
+        return publishResult;
     }
 
     /**
@@ -583,10 +666,10 @@ class PostOffice {
     /**
      * Route the command to the owning SessionEventLoop
      * */
-    public CompletableFuture<RouteResult> routeCommand(String clientId, Callable<String> action) {
+    public RouteResult routeCommand(String clientId, String actionDescription, Callable<String> action) {
         SessionCommand cmd = new SessionCommand(clientId, action);
         final int targetQueueId = Math.abs(cmd.getSessionId().hashCode()) % this.eventLoops;
-        LOG.debug("Routing cmd for session [{}] to event processor {}", cmd.getSessionId(), targetQueueId);
+        LOG.debug("Routing cmd [{}] for session [{}] to event processor {}", actionDescription, cmd.getSessionId(), targetQueueId);
         final FutureTask<String> task = new FutureTask<>(() -> {
             cmd.execute();
             cmd.complete();
@@ -594,22 +677,27 @@ class PostOffice {
         });
         if (Thread.currentThread() == sessionExecutors[targetQueueId]) {
             SessionEventLoop.executeTask(task);
-            return cmd.completableFuture().thenApply(RouteResult::success);
+            return RouteResult.success(clientId, cmd.completableFuture());
         }
         if (this.sessionQueues[targetQueueId].offer(task)) {
-            return cmd.completableFuture().thenApply(RouteResult::success);
+            return RouteResult.success(clientId, cmd.completableFuture());
         } else {
-            LOG.warn("Session command queue {} is full", targetQueueId);
-            final CompletableFuture<RouteResult> failed = new CompletableFuture<>();
-            failed.complete(RouteResult.failed(cmd.getSessionId()));
-            return failed;
+            LOG.warn("Session command queue {} is full executing action {}", targetQueueId, actionDescription);
+            return RouteResult.failed(clientId);
         }
     }
 
     public void terminate() {
-        for (Thread processor:sessionExecutors) {
+        for (Thread processor : sessionExecutors) {
             processor.interrupt();
         }
+        for (Thread processor : sessionExecutors) {
+            try {
+                processor.join(5_000);
+            } catch (InterruptedException ex) {
+                LOG.info("Interrupted while joining session event loop {}", processor.getName(), ex);
+            }
+        }
     }
 
     /**
diff --git a/broker/src/main/java/io/moquette/broker/Server.java b/broker/src/main/java/io/moquette/broker/Server.java
index 42b9aa3c..177fcab1 100644
--- a/broker/src/main/java/io/moquette/broker/Server.java
+++ b/broker/src/main/java/io/moquette/broker/Server.java
@@ -35,7 +35,6 @@ import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.text.ParseException;
 import java.util.*;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 
@@ -307,7 +306,7 @@ public class Server {
      * @throws IllegalStateException if the integration is not yet started
      * @return
      */
-    public CompletableFuture<Void> internalPublish(MqttPublishMessage msg, final String clientId) {
+    public PostOffice.RoutingResults internalPublish(MqttPublishMessage msg, final String clientId) {
         final int messageID = msg.variableHeader().packetId();
         if (!initialized) {
             LOG.error("Moquette is not started, internal message cannot be published. CId: {}, messageId: {}", clientId,
@@ -315,8 +314,9 @@ public class Server {
             throw new IllegalStateException("Can't publish on a integration is not yet started");
         }
         LOG.trace("Internal publishing message CId: {}, messageId: {}", clientId, messageID);
-        return dispatcher.internalPublish(msg)
-            .thenRun(() -> msg.payload().release());
+        final PostOffice.RoutingResults routingResults = dispatcher.internalPublish(msg);
+        msg.payload().release();
+        return routingResults;
     }
 
     public void stopServer() {
diff --git a/broker/src/main/java/io/moquette/broker/SessionEventLoop.java b/broker/src/main/java/io/moquette/broker/SessionEventLoop.java
index a25ca9a1..dea036ba 100644
--- a/broker/src/main/java/io/moquette/broker/SessionEventLoop.java
+++ b/broker/src/main/java/io/moquette/broker/SessionEventLoop.java
@@ -11,14 +11,23 @@ final class SessionEventLoop implements Runnable {
     private static final Logger LOG = LoggerFactory.getLogger(SessionEventLoop.class);
 
     private final BlockingQueue<FutureTask<String>> sessionQueue;
+    private final boolean flushOnExit;
 
     public SessionEventLoop(BlockingQueue<FutureTask<String>> sessionQueue) {
+        this(sessionQueue, true);
+    }
+
+    /**
+     * @param flushOnExit consume the commands queue before exit.
+     * */
+    public SessionEventLoop(BlockingQueue<FutureTask<String>> sessionQueue, boolean flushOnExit) {
         this.sessionQueue = sessionQueue;
+        this.flushOnExit = flushOnExit;
     }
 
     @Override
     public void run() {
-        while (!Thread.interrupted()) {
+        while (!Thread.interrupted() || (Thread.interrupted() && !sessionQueue.isEmpty() && flushOnExit)) {
             try {
                 // blocking call
                 final FutureTask<String> task = this.sessionQueue.take();
@@ -28,6 +37,7 @@ final class SessionEventLoop implements Runnable {
                 Thread.currentThread().interrupt();
             }
         }
+        LOG.info("SessionEventLoop {} exit", Thread.currentThread().getName());
     }
 
     public static void executeTask(final FutureTask<String> task) {


Test patch:
diff --git a/broker/src/test/java/io/moquette/broker/ConnectionTestUtils.java b/broker/src/test/java/io/moquette/broker/ConnectionTestUtils.java
index 6f580b65..d45ef96e 100644
--- a/broker/src/test/java/io/moquette/broker/ConnectionTestUtils.java
+++ b/broker/src/test/java/io/moquette/broker/ConnectionTestUtils.java
@@ -23,6 +23,7 @@ import io.netty.handler.codec.mqtt.*;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.ExecutionException;
 
 import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED;
 import static org.junit.jupiter.api.Assertions.*;
@@ -35,6 +36,10 @@ public final class ConnectionTestUtils {
     private ConnectionTestUtils() {
     }
 
+    static void assertConnectAccepted(MQTTConnection connection) {
+        assertConnectAccepted((EmbeddedChannel) connection.channel);
+    }
+
     static void assertConnectAccepted(EmbeddedChannel channel) {
         MqttConnAckMessage connAck = channel.readOutbound();
         final MqttConnectReturnCode connAckReturnCode = connAck.variableHeader().connectReturnCode();
@@ -69,8 +74,13 @@ public final class ConnectionTestUtils {
         assertTrue(receivedPublish.fixedHeader().isRetain(), "MUST be retained publish");
     }
 
+    static void verifyPublishIsReceived(MQTTConnection connection, MqttQoS expectedQos, String expectedPayload) {
+        verifyPublishIsReceived((EmbeddedChannel) connection.channel, expectedQos, expectedPayload);
+    }
+
     static void verifyPublishIsReceived(EmbeddedChannel embCh, MqttQoS expectedQos, String expectedPayload) {
         final MqttPublishMessage publishReceived = embCh.flushOutbound().readOutbound();
+        assertNotNull(publishReceived, "A PUB message must be received");
         final String payloadMessage = DebugUtils.payload2Str(publishReceived.payload());
         assertEquals(expectedPayload, payloadMessage, "Sent and received payload must be identical");
         assertEquals(expectedQos, publishReceived.fixedHeader().qosLevel(), "Expected QoS don't match");
@@ -93,4 +103,13 @@ public final class ConnectionTestUtils {
             .cleanSession(false)
             .build();
     }
+
+    static void connect(MQTTConnection connection, MqttConnectMessage connectMessage) {
+        try {
+            connection.processConnect(connectMessage).completableFuture().get();
+        } catch (InterruptedException | ExecutionException e) {
+            fail(e);
+        }
+        assertConnectAccepted((EmbeddedChannel) connection.channel);
+    }
 }
diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java
index ca73c87d..08e2358d 100644
--- a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java
+++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java
@@ -101,7 +101,7 @@ public class MQTTConnectionConnectTest {
             .cleanSession(true)
             .build();
 
-        sut.processConnect(msg).get();
+        sut.processConnect(msg).completableFuture().get();
         assertEqualsConnAck("Connection must be accepted", CONNECTION_ACCEPTED, channel.readOutbound());
         assertNotNull("unique clientid must be generated", sut.getClientId());
         assertTrue(sessionRegistry.retrieve(sut.getClientId()).isClean(), "clean session flag must be true");
@@ -128,7 +128,7 @@ public class MQTTConnectionConnectTest {
         connMsg.clientId("extremely_long_clientID_greater_than_23").build();
 
         // Exercise
-        sut.processConnect(connMsg.clientId("extremely_long_clientID_greater_than_23").build()).get();
+        sut.processConnect(connMsg.clientId("extremely_long_clientID_greater_than_23").build()).completableFuture().get();
 
         // Verify
         assertEqualsConnAck(CONNECTION_ACCEPTED, channel.readOutbound());
@@ -140,7 +140,7 @@ public class MQTTConnectionConnectTest {
             .willTopic("topic").willMessage("Topic message").build();
 
         // Exercise
-        sut.processConnect(msg).get();
+        sut.processConnect(msg).completableFuture().get();
 
         // Verify
         assertEqualsConnAck(CONNECTION_ACCEPTED, channel.readOutbound());
@@ -156,7 +156,7 @@ public class MQTTConnectionConnectTest {
 
         MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).willFlag(true)
             .willTopic("topic").willMessage("Topic message").build();
-        sut.processConnect(msg).get();
+        sut.processConnect(msg).completableFuture().get();
 
         // Exercise
         sut.handleConnectionLost();
@@ -171,7 +171,7 @@ public class MQTTConnectionConnectTest {
         MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build();
 
         // Exercise
-        sut.processConnect(msg).get();
+        sut.processConnect(msg).completableFuture().get();
 
         // Verify
         assertEqualsConnAck(CONNECTION_ACCEPTED, channel.readOutbound());
@@ -184,7 +184,7 @@ public class MQTTConnectionConnectTest {
             .username(TEST_USER).password(TEST_PWD).build();
 
         // Exercise
-        sut.processConnect(msg).get();
+        sut.processConnect(msg).completableFuture().get();
 
         // Verify
         assertEqualsConnAck(CONNECTION_ACCEPTED, channel.readOutbound());
@@ -277,7 +277,7 @@ public class MQTTConnectionConnectTest {
             .username(TEST_USER)
             .password(TEST_PWD)
             .build();
-        sut.processConnect(msg).get();
+        sut.processConnect(msg).completableFuture().get();
         assertEqualsConnAck(CONNECTION_ACCEPTED, channel.readOutbound());
 
         // create another connect same clientID but with bad credentials
@@ -309,12 +309,12 @@ public class MQTTConnectionConnectTest {
             .username(TEST_USER)
             .password(TEST_PWD)
             .build();
-        sut.processConnect(msg).get();
+        sut.processConnect(msg).completableFuture().get();
         assertEqualsConnAck(CONNECTION_ACCEPTED, channel.readOutbound());
 
         // now create another connection and check the new one closes the older
         MQTTConnection anotherConnection = createMQTTConnection(CONFIG);
-        anotherConnection.processConnect(msg).get();
+        anotherConnection.processConnect(msg).completableFuture().get();
         EmbeddedChannel anotherChannel = (EmbeddedChannel) anotherConnection.channel;
         assertEqualsConnAck(CONNECTION_ACCEPTED, anotherChannel.readOutbound());
 
diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java
index 9fd9b530..7d605de8 100644
--- a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java
+++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java
@@ -92,7 +92,7 @@ public class MQTTConnectionPublishTest {
             .qos(MqttQoS.AT_MOST_ONCE)
             .payload(payload).build();
 
-        sut.processPublish(publish).get();
+        sut.processPublish(publish).completableFuture().get();
 
         // Verify
         assertFalse(channel.isOpen(), "Connection should be closed by the broker");
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java
index a5b05ffd..1a1f7570 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java
@@ -31,7 +31,6 @@ import org.junit.jupiter.api.Test;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 
 import static io.moquette.broker.PostOfficeUnsubscribeTest.CONFIG;
 import static io.netty.handler.codec.mqtt.MqttQoS.*;
@@ -61,14 +60,14 @@ public class PostOfficeInternalPublishTest {
 
     @BeforeEach
     public void setUp() throws ExecutionException, InterruptedException {
-        sessionRegistry = initPostOfficeAndSubsystems();
+        initPostOfficeAndSubsystems();
 
         mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD));
         connection = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID);
 
         connectMessage = ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID);
 
-        connection.processConnect(connectMessage).get();
+        connection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
     }
 
@@ -81,7 +80,7 @@ public class PostOfficeInternalPublishTest {
         return new MQTTConnection(channel, config, mockAuthenticator, sessionRegistry, sut);
     }
 
-    private SessionRegistry initPostOfficeAndSubsystems() {
+    private void initPostOfficeAndSubsystems() {
         subscriptions = new CTrieSubscriptionDirectory();
         ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository();
         subscriptions.init(subscriptionsRepository);
@@ -90,10 +89,9 @@ public class PostOfficeInternalPublishTest {
 
         final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy();
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
-        SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll);
+        sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll);
         sut = new PostOffice(subscriptions, retainedRepository, sessionRegistry,
                              ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, 1024);
-        return sessionRegistry;
     }
 
     private void internalPublishNotRetainedTo(String topic) {
@@ -110,10 +108,11 @@ public class PostOfficeInternalPublishTest {
             .retained(retained)
             .qos(qos)
             .payload(Unpooled.copiedBuffer(PAYLOAD.getBytes(UTF_8))).build();
+        final PostOffice.RoutingResults res = sut.internalPublish(publish);
         try {
-            sut.internalPublish(publish).get(5, TimeUnit.SECONDS);
-        } catch (InterruptedException | ExecutionException | TimeoutException e) {
-            throw new RuntimeException(e);
+            res.completableFuture().get(5, TimeUnit.SECONDS);
+        } catch (Exception ex) {
+            throw new RuntimeException(ex);
         }
     }
 
@@ -127,6 +126,7 @@ public class PostOfficeInternalPublishTest {
         internalPublishNotRetainedTo(topic);
 
         subscribe(AT_MOST_ONCE, topic, connection);
+        completeEventLoopCommands();
 
         // Verify
         verifyNoPublishIsReceived(channel);
@@ -139,6 +139,7 @@ public class PostOfficeInternalPublishTest {
         internalPublishRetainedTo(topic);
 
         subscribe(AT_MOST_ONCE, topic, connection);
+        completeEventLoopCommands();
 
         // Verify
         verifyNoPublishIsReceived(channel);
@@ -150,6 +151,7 @@ public class PostOfficeInternalPublishTest {
 
         // Exercise
         internalPublishNotRetainedTo("/topic");
+        completeEventLoopCommands();
 
         // Verify
         ConnectionTestUtils.verifyPublishIsReceived(channel, AT_MOST_ONCE, PAYLOAD);
@@ -161,17 +163,24 @@ public class PostOfficeInternalPublishTest {
 
         // Exercise
         internalPublishRetainedTo("/topic");
+        completeEventLoopCommands();
 
         // Verify
         ConnectionTestUtils.verifyPublishIsReceived(channel, AT_MOST_ONCE, PAYLOAD);
     }
 
+    private void completeEventLoopCommands() {
+        // shutdown event loop threads to complete all queued commands
+        sut.terminate();
+    }
+
     @Test
     public void testClientSubscribeBeforeNotRetainedQoS1IsSent() {
         subscribe(AT_LEAST_ONCE, "/topic", connection);
 
         // Exercise
         internalPublishTo("/topic", AT_LEAST_ONCE, false);
+        completeEventLoopCommands();
 
         // Verify
         ConnectionTestUtils.verifyPublishIsReceived(channel, AT_LEAST_ONCE, PAYLOAD);
@@ -182,17 +191,19 @@ public class PostOfficeInternalPublishTest {
         // Exercise
         internalPublishTo("/topic", AT_LEAST_ONCE, false);
         subscribe(AT_LEAST_ONCE, "/topic", connection);
+        completeEventLoopCommands();
 
         // Verify
         verifyNoPublishIsReceived(channel);
     }
 
     @Test
-    public void testClientSubscribeBeforeRetainedQoS1IsSent() {
+    public void testClientSubscribeBeforeRetainedQoS1IsSent() throws Exception {
         subscribe(AT_LEAST_ONCE, "/topic", connection);
 
         // Exercise
         internalPublishTo("/topic", AT_LEAST_ONCE, true);
+        completeEventLoopCommands();
 
         // Verify
         ConnectionTestUtils.verifyPublishIsReceived(channel, AT_LEAST_ONCE, PAYLOAD);
@@ -203,17 +214,19 @@ public class PostOfficeInternalPublishTest {
         // Exercise
         internalPublishTo("/topic", AT_LEAST_ONCE, true);
         subscribe(AT_LEAST_ONCE, "/topic", connection);
+        completeEventLoopCommands();
 
         // Verify
         ConnectionTestUtils.verifyPublishIsReceived(channel, AT_LEAST_ONCE, PAYLOAD);
     }
 
     @Test
-    public void testClientSubscribeBeforeNotRetainedQoS2IsSent() {
+    public void testClientSubscribeBeforeNotRetainedQoS2IsSent() throws Exception {
         subscribe(EXACTLY_ONCE, "/topic", connection);
 
         // Exercise
         internalPublishTo("/topic", EXACTLY_ONCE, false);
+        completeEventLoopCommands();
 
         // Verify
         ConnectionTestUtils.verifyPublishIsReceived(channel, EXACTLY_ONCE, PAYLOAD);
@@ -224,6 +237,7 @@ public class PostOfficeInternalPublishTest {
         // Exercise
         internalPublishTo("/topic", EXACTLY_ONCE, false);
         subscribe(EXACTLY_ONCE, "/topic", connection);
+        completeEventLoopCommands();
 
         // Verify
         verifyNoPublishIsReceived(channel);
@@ -235,8 +249,10 @@ public class PostOfficeInternalPublishTest {
 
         // Exercise
         internalPublishTo("/topic", EXACTLY_ONCE, true);
+        completeEventLoopCommands();
 
         // Verify
+        assertNotNull(channel, "channel can't be null");
         ConnectionTestUtils.verifyPublishIsReceived(channel, EXACTLY_ONCE, PAYLOAD);
     }
 
@@ -245,6 +261,7 @@ public class PostOfficeInternalPublishTest {
         // Exercise
         internalPublishTo("/topic", EXACTLY_ONCE, true);
         subscribe(EXACTLY_ONCE, "/topic", connection);
+        completeEventLoopCommands();
 
         // Verify
         ConnectionTestUtils.verifyPublishIsReceived(channel, EXACTLY_ONCE, PAYLOAD);
@@ -256,6 +273,7 @@ public class PostOfficeInternalPublishTest {
         connection.processDisconnect(null);
 
         internalPublishTo("foo", AT_MOST_ONCE, false);
+        completeEventLoopCommands();
 
         verifyNoPublishIsReceived(channel);
     }
@@ -272,12 +290,13 @@ public class PostOfficeInternalPublishTest {
             .clientId(FAKE_CLIENT_ID)
             .cleanSession(false)
             .build();
-        anotherConn.processConnect(connectMessage).get();
-        ConnectionTestUtils.assertConnectAccepted((EmbeddedChannel) anotherConn.channel);
+        anotherConn.processConnect(connectMessage).completableFuture().get();
+        ConnectionTestUtils.assertConnectAccepted(anotherConn);
 
         assertEquals(1, subscriptions.size());
         internalPublishTo("foo", MqttQoS.AT_MOST_ONCE, false);
-        ConnectionTestUtils.verifyPublishIsReceived((EmbeddedChannel) anotherConn.channel, AT_MOST_ONCE, PAYLOAD);
+        completeEventLoopCommands();
+        ConnectionTestUtils.verifyPublishIsReceived(anotherConn, AT_MOST_ONCE, PAYLOAD);
     }
 
     private void subscribe(MqttQoS topic, String newsTopic, MQTTConnection connection) {
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java
index caa02ff8..6a8ed3bb 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java
@@ -28,6 +28,7 @@ import io.netty.buffer.UnpooledByteBufAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.codec.mqtt.*;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -71,24 +72,29 @@ public class PostOfficePublishTest {
 
     @BeforeEach
     public void setUp() {
-        sessionRegistry = initPostOfficeAndSubsystems();
+        initPostOfficeAndSubsystems();
 
         mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD));
         connection = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID);
+        channel = (EmbeddedChannel) connection.channel;
 
         connectMessage = ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID);
     }
 
+    @AfterEach
+    public void tearDown() {
+        sut.terminate();
+    }
+
     private MQTTConnection createMQTTConnection(BrokerConfiguration config) {
-        channel = new EmbeddedChannel();
-        return createMQTTConnection(config, channel);
+        return createMQTTConnection(config, new EmbeddedChannel());
     }
 
     private MQTTConnection createMQTTConnection(BrokerConfiguration config, Channel channel) {
         return new MQTTConnection(channel, config, mockAuthenticator, sessionRegistry, sut);
     }
 
-    private SessionRegistry initPostOfficeAndSubsystems() {
+    private void initPostOfficeAndSubsystems() {
         subscriptions = new CTrieSubscriptionDirectory();
         ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository();
         subscriptions.init(subscriptionsRepository);
@@ -97,15 +103,14 @@ public class PostOfficePublishTest {
 
         final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy();
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
-        SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll);
+        sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll);
         sut = new PostOffice(subscriptions, retainedRepository, sessionRegistry,
                              ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, 1024);
-        return sessionRegistry;
     }
 
     @Test
     public void testPublishQoS0ToItself() throws ExecutionException, InterruptedException, TimeoutException {
-        connection.processConnect(connectMessage).get();
+        connection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
 
         // subscribe
@@ -148,7 +153,7 @@ public class PostOfficePublishTest {
             .qos(MqttQoS.EXACTLY_ONCE)
             .retained(true)
             .messageId(2)
-            .topicName(NEWS_TOPIC).build(), "username").get(5, TimeUnit.SECONDS);
+            .topicName(NEWS_TOPIC).build(), "username").completableFuture().get(5, TimeUnit.SECONDS);
 
         // Verify
         assertFalse(clientXA.channel.isOpen(), "First 'subscriber' channel MUST be closed by the broker");
@@ -158,12 +163,7 @@ public class PostOfficePublishTest {
     private MQTTConnection connectAs(String clientId) {
         EmbeddedChannel channel = new EmbeddedChannel();
         MQTTConnection connection = createMQTTConnection(CONFIG, channel);
-        try {
-            connection.processConnect(ConnectionTestUtils.buildConnect(clientId)).get();
-        } catch (InterruptedException | ExecutionException e) {
-            fail(e);
-        }
-        ConnectionTestUtils.assertConnectAccepted(channel);
+        ConnectionTestUtils.connect(connection, ConnectionTestUtils.buildConnect(clientId));
         return connection;
     }
 
@@ -204,12 +204,12 @@ public class PostOfficePublishTest {
         mockAuthenticator = new MockAuthenticator(clientIds, singletonMap(TEST_USER, TEST_PWD));
         EmbeddedChannel channel1 = new EmbeddedChannel();
         MQTTConnection connection1 = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID, channel1);
-        connection1.processConnect(ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID)).get();
+        connection1.processConnect(ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID)).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel1);
 
         EmbeddedChannel channel2 = new EmbeddedChannel();
         MQTTConnection connection2 = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID, channel2);
-        connection2.processConnect(ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID2)).get();
+        connection2.processConnect(ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID2)).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel2);
 
         // subscribe
@@ -234,7 +234,7 @@ public class PostOfficePublishTest {
 
     @Test
     public void testPublishWithEmptyPayloadClearRetainedStore() throws ExecutionException, InterruptedException {
-        connection.processConnect(connectMessage).get();
+        connection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
 
         final ByteBuf payload1 = ByteBufUtil.writeAscii(UnpooledByteBufAllocator.DEFAULT, "Hello world!");
@@ -262,18 +262,22 @@ public class PostOfficePublishTest {
 
     @Test
     public void testPublishWithQoS1() throws ExecutionException, InterruptedException, TimeoutException {
-        connection.processConnect(connectMessage).get();
+        connection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         subscribe(connection, NEWS_TOPIC, AT_LEAST_ONCE);
 
+        MQTTConnection senderConnection = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID);
+        senderConnection.processConnect(ConnectionTestUtils.buildConnect("Publisher")).completableFuture().get();
+        ConnectionTestUtils.assertConnectAccepted(senderConnection);
+
         // Exercise
         final ByteBuf anyPayload = Unpooled.copiedBuffer("Any payload", Charset.defaultCharset());
-        sut.receivedPublishQos1(connection, new Topic(NEWS_TOPIC), TEST_USER, 1,
+        sut.receivedPublishQos1(senderConnection, new Topic(NEWS_TOPIC), TEST_USER, 1,
             MqttMessageBuilders.publish()
-                .payload(Unpooled.copiedBuffer("Any payload", Charset.defaultCharset()))
+                .payload(anyPayload)
                 .qos(MqttQoS.AT_LEAST_ONCE)
                 .retained(true)
-                .topicName(NEWS_TOPIC).build()).get(5, TimeUnit.SECONDS);
+                .topicName(NEWS_TOPIC).build()).completableFuture().get(5, TimeUnit.SECONDS);
 
         // Verify
         ConnectionTestUtils.verifyPublishIsReceived(channel, AT_LEAST_ONCE, "Any payload");
@@ -281,18 +285,22 @@ public class PostOfficePublishTest {
 
     @Test
     public void testPublishWithQoS2() throws ExecutionException, InterruptedException, TimeoutException {
-        connection.processConnect(connectMessage).get();
+        connection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         subscribe(connection, NEWS_TOPIC, EXACTLY_ONCE);
 
+        MQTTConnection senderConnection = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID);
+        senderConnection.processConnect(ConnectionTestUtils.buildConnect("Publisher")).completableFuture().get();
+        ConnectionTestUtils.assertConnectAccepted(senderConnection);
+
         // Exercise
         final ByteBuf anyPayload = Unpooled.copiedBuffer("Any payload", Charset.defaultCharset());
-        sut.receivedPublishQos2(connection, MqttMessageBuilders.publish()
+        sut.receivedPublishQos2(senderConnection, MqttMessageBuilders.publish()
                 .payload(anyPayload)
                 .qos(MqttQoS.EXACTLY_ONCE)
                 .retained(true)
                 .messageId(2)
-                .topicName(NEWS_TOPIC).build(), "username").get(5000, TimeUnit.SECONDS);
+                .topicName(NEWS_TOPIC).build(), "username").completableFuture().get(5000, TimeUnit.SECONDS);
 
         // Verify
         ConnectionTestUtils.verifyPublishIsReceived(channel, EXACTLY_ONCE, "Any payload");
@@ -301,7 +309,7 @@ public class PostOfficePublishTest {
     // aka testPublishWithQoS1_notCleanSession
     @Test
     public void forwardQoS1PublishesWhenNotCleanSessionReconnects() throws ExecutionException, InterruptedException {
-        connection.processConnect(ConnectionTestUtils.buildConnectNotClean(FAKE_CLIENT_ID)).get();
+        connection.processConnect(ConnectionTestUtils.buildConnectNotClean(FAKE_CLIENT_ID)).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         subscribe(connection, NEWS_TOPIC, AT_LEAST_ONCE);
         connection.processDisconnect(null);
@@ -309,7 +317,7 @@ public class PostOfficePublishTest {
         // publish a QoS 1 message from another client publish a message on the topic
         EmbeddedChannel pubChannel = new EmbeddedChannel();
         MQTTConnection pubConn = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID, pubChannel);
-        pubConn.processConnect(ConnectionTestUtils.buildConnect(PUBLISHER_ID)).get();
+        pubConn.processConnect(ConnectionTestUtils.buildConnect(PUBLISHER_ID)).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(pubChannel);
 
         final ByteBuf anyPayload = Unpooled.copiedBuffer("Any payload", Charset.defaultCharset());
@@ -321,8 +329,9 @@ public class PostOfficePublishTest {
 
         // simulate a reconnection from the other client
         connection = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID);
+        channel = (EmbeddedChannel) connection.channel;
         connectMessage = ConnectionTestUtils.buildConnectNotClean(FAKE_CLIENT_ID);
-        connection.processConnect(connectMessage).get();
+        connection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
 
         // Verify
@@ -332,7 +341,7 @@ public class PostOfficePublishTest {
     @Test
     public void checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession() throws ExecutionException, InterruptedException, TimeoutException {
         // first connect - subscribe -disconnect
-        connection.processConnect(ConnectionTestUtils.buildConnectNotClean(FAKE_CLIENT_ID)).get();
+        connection.processConnect(ConnectionTestUtils.buildConnectNotClean(FAKE_CLIENT_ID)).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         subscribe(connection, NEWS_TOPIC, AT_LEAST_ONCE);
         connection.processDisconnect(null);
@@ -340,14 +349,14 @@ public class PostOfficePublishTest {
         // connect - subscribe from another connection but with same ClientID
         EmbeddedChannel secondChannel = new EmbeddedChannel();
         MQTTConnection secondConn = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID, secondChannel);
-        secondConn.processConnect(ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID)).get();
+        secondConn.processConnect(ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID)).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(secondChannel);
         subscribe(secondConn, NEWS_TOPIC, AT_LEAST_ONCE);
 
         // publish a QoS 1 message another client publish a message on the topic
         EmbeddedChannel pubChannel = new EmbeddedChannel();
         MQTTConnection pubConn = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID, pubChannel);
-        pubConn.processConnect(ConnectionTestUtils.buildConnect(PUBLISHER_ID)).get();
+        pubConn.processConnect(ConnectionTestUtils.buildConnect(PUBLISHER_ID)).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(pubChannel);
 
         final ByteBuf anyPayload = Unpooled.copiedBuffer("Any payload", Charset.defaultCharset());
@@ -356,7 +365,7 @@ public class PostOfficePublishTest {
                 .payload(anyPayload.retainedDuplicate())
                 .qos(MqttQoS.AT_LEAST_ONCE)
                 .retained(true)
-                .topicName(NEWS_TOPIC).build()).get(5, TimeUnit.SECONDS);
+                .topicName(NEWS_TOPIC).build()).completableFuture().get(5, TimeUnit.SECONDS);
 
         // Verify that after a reconnection the client receive the message
         ConnectionTestUtils.verifyPublishIsReceived(secondChannel, AT_LEAST_ONCE, "Any payload");
@@ -365,7 +374,7 @@ public class PostOfficePublishTest {
     @Test
     public void noPublishToInactiveSession() throws ExecutionException, InterruptedException {
         // create an inactive session for Subscriber
-        connection.processConnect(ConnectionTestUtils.buildConnectNotClean(SUBSCRIBER_ID)).get();
+        connection.processConnect(ConnectionTestUtils.buildConnectNotClean(SUBSCRIBER_ID)).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         subscribe(connection, NEWS_TOPIC, AT_LEAST_ONCE);
         connection.processDisconnect(null);
@@ -373,7 +382,7 @@ public class PostOfficePublishTest {
         // Exercise
         EmbeddedChannel pubChannel = new EmbeddedChannel();
         MQTTConnection pubConn = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID, pubChannel);
-        pubConn.processConnect(ConnectionTestUtils.buildConnect(PUBLISHER_ID)).get();
+        pubConn.processConnect(ConnectionTestUtils.buildConnect(PUBLISHER_ID)).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(pubChannel);
 
         final ByteBuf anyPayload = Unpooled.copiedBuffer("Any payload", Charset.defaultCharset());
@@ -394,9 +403,13 @@ public class PostOfficePublishTest {
 
     @Test
     public void cleanRetainedMessageStoreWhenPublishWithRetainedQos0IsReceived() throws ExecutionException, InterruptedException {
-        connection.processConnect(connectMessage).get();
+        connection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
 
+        MQTTConnection senderConnection = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID);
+        senderConnection.processConnect(ConnectionTestUtils.buildConnect("Publisher")).completableFuture().get();
+        ConnectionTestUtils.assertConnectAccepted(senderConnection);
+
         // publish a QoS1 retained message
         final ByteBuf anyPayload = Unpooled.copiedBuffer("Any payload", Charset.defaultCharset());
         final MqttPublishMessage publishMsg = MqttMessageBuilders.publish()
@@ -405,7 +418,7 @@ public class PostOfficePublishTest {
             .retained(true)
             .topicName(NEWS_TOPIC)
             .build();
-        sut.receivedPublishQos1(connection, new Topic(NEWS_TOPIC), TEST_USER, 1,
+        sut.receivedPublishQos1(senderConnection, new Topic(NEWS_TOPIC), TEST_USER, 1,
                                 publishMsg);
 
         assertMessageIsRetained(NEWS_TOPIC, anyPayload);
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java
index 30a638b8..5318d6ca 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java
@@ -115,7 +115,7 @@ public class PostOfficeSubscribeTest {
 
     @Test
     public void testSubscribe() throws ExecutionException, InterruptedException {
-        connection.processConnect(connectMessage).get();
+        connection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
 
         // Exercise & verify
@@ -172,7 +172,7 @@ public class PostOfficeSubscribeTest {
         sut = new PostOffice(subscriptions, new MemoryRetainedRepository(), sessionRegistry,
                              ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, new Authorizator(prohibitReadOnNewsTopic), 1024);
 
-        connection.processConnect(connectMessage).get();
+        connection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
 
         //Exercise
@@ -195,7 +195,7 @@ public class PostOfficeSubscribeTest {
 
     @Test
     public void testDoubleSubscribe() throws ExecutionException, InterruptedException {
-        connection.processConnect(connectMessage).get();
+        connection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         assertEquals(0, subscriptions.size(), "After CONNECT subscription MUST be empty");
         subscribe(channel, NEWS_TOPIC, AT_MOST_ONCE);
@@ -207,7 +207,7 @@ public class PostOfficeSubscribeTest {
 
     @Test
     public void testSubscribeWithBadFormattedTopic() throws ExecutionException, InterruptedException {
-        connection.processConnect(connectMessage).get();
+        connection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         assertEquals(0, subscriptions.size(), "After CONNECT subscription MUST be empty");
 
@@ -225,7 +225,7 @@ public class PostOfficeSubscribeTest {
 
     @Test
     public void testCleanSession_maintainClientSubscriptions() throws ExecutionException, InterruptedException, TimeoutException {
-        connection.processConnect(connectMessage).get();
+        connection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         assertEquals(0, subscriptions.size(), "After CONNECT subscription MUST be empty");
 
@@ -238,7 +238,7 @@ public class PostOfficeSubscribeTest {
 
         EmbeddedChannel anotherChannel = new EmbeddedChannel();
         MQTTConnection anotherConn = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID, anotherChannel);
-        anotherConn.processConnect(ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID)).get();
+        anotherConn.processConnect(ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID)).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(anotherChannel);
         assertEquals(1, subscriptions.size(), "After a reconnect, subscription MUST be still present");
 
@@ -260,7 +260,7 @@ public class PostOfficeSubscribeTest {
      */
     @Test
     public void testCleanSession_correctlyClientSubscriptions() throws ExecutionException, InterruptedException, TimeoutException {
-        connection.processConnect(connectMessage).get();
+        connection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         assertEquals(0, subscriptions.size(), "After CONNECT subscription MUST be empty");
 
@@ -270,10 +270,10 @@ public class PostOfficeSubscribeTest {
             .addSubscription(AT_MOST_ONCE, NEWS_TOPIC)
             .messageId(1)
             .build();
-        connection.processSubscribe(subscribeMsg).get(5, TimeUnit.SECONDS);
+        connection.processSubscribe(subscribeMsg).completableFuture().get(5, TimeUnit.SECONDS);
         assertEquals(1, subscriptions.size(), "Subscribe MUST contain one subscription");
 
-        connection.processDisconnect(null).get(5, TimeUnit.SECONDS);
+        connection.processDisconnect(null).completableFuture().get(5, TimeUnit.SECONDS);
         assertEquals(1, subscriptions.size(), "Disconnection MUSTN'T clear subscriptions");
 
         connectMessage = MqttMessageBuilders.connect()
@@ -282,7 +282,7 @@ public class PostOfficeSubscribeTest {
             .build();
         channel = new EmbeddedChannel();
         connection = createMQTTConnection(CONFIG, channel);
-        connection.processConnect(connectMessage).get();
+        connection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         assertEquals(0, subscriptions.size(), "After CONNECT with clean, subscription MUST be empty");
 
@@ -302,7 +302,7 @@ public class PostOfficeSubscribeTest {
     @Test
     public void testReceiveRetainedPublishRespectingSubscriptionQoSAndNotPublisher() throws ExecutionException, InterruptedException {
         // publisher publish a retained message on topic /news
-        connection.processConnect(connectMessage).get();
+        connection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         final ByteBuf payload = Unpooled.copiedBuffer("Hello world!", Charset.defaultCharset());
         final MqttPublishMessage retainedPubQoS1Msg = MqttMessageBuilders.publish()
@@ -316,7 +316,7 @@ public class PostOfficeSubscribeTest {
         // subscriber connects subscribe to topic /news and receive the last retained message
         EmbeddedChannel subChannel = new EmbeddedChannel();
         MQTTConnection subConn = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID, subChannel);
-        subConn.processConnect(ConnectionTestUtils.buildConnect(SUBSCRIBER_ID)).get();
+        subConn.processConnect(ConnectionTestUtils.buildConnect(SUBSCRIBER_ID)).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(subChannel);
         subscribe(subConn, NEWS_TOPIC, MqttQoS.AT_MOST_ONCE);
 
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java
index 177117fc..579a64fa 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java
@@ -96,18 +96,9 @@ public class PostOfficeUnsubscribeTest {
         return new MQTTConnection(channel, config, mockAuthenticator, sessionRegistry, sut);
     }
 
-    protected void connect(MQTTConnection connection, String clientId) {
+    protected static void connect(MQTTConnection connection, String clientId) {
         MqttConnectMessage connectMessage = ConnectionTestUtils.buildConnect(clientId);
-        connect(connection, connectMessage);
-    }
-
-    protected void connect(MQTTConnection connection, MqttConnectMessage connectMessage) {
-        try {
-            connection.processConnect(connectMessage).get();
-        } catch (InterruptedException | ExecutionException e) {
-            Assertions.fail(e);
-        }
-        ConnectionTestUtils.assertConnectAccepted((EmbeddedChannel) connection.channel);
+        ConnectionTestUtils.connect(connection, connectMessage);
     }
 
     protected void subscribe(MQTTConnection connection, String topic, MqttQoS desiredQos) {
@@ -204,7 +195,7 @@ public class PostOfficeUnsubscribeTest {
         // connect on another channel
         EmbeddedChannel anotherChannel = new EmbeddedChannel();
         MQTTConnection anotherConnection = createMQTTConnection(CONFIG, anotherChannel);
-        anotherConnection.processConnect(connectMessage).get();
+        anotherConnection.processConnect(connectMessage).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(anotherChannel);
 
         // publish on /news
@@ -222,7 +213,7 @@ public class PostOfficeUnsubscribeTest {
     @Test
     public void avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1() throws ExecutionException, InterruptedException {
         final MqttConnectMessage notCleanConnect = ConnectionTestUtils.buildConnectNotClean(FAKE_CLIENT_ID);
-        connect(connection, notCleanConnect);
+        ConnectionTestUtils.connect(connection, notCleanConnect);
         subscribe(connection, NEWS_TOPIC, AT_LEAST_ONCE);
         connection.processDisconnect(null);
 
@@ -233,7 +224,7 @@ public class PostOfficeUnsubscribeTest {
         // reconnect FAKE_CLIENT on another channel
         EmbeddedChannel anotherChannel2 = new EmbeddedChannel();
         MQTTConnection anotherConnection2 = createMQTTConnection(CONFIG, anotherChannel2);
-        anotherConnection2.processConnect(notCleanConnect).get();
+        anotherConnection2.processConnect(notCleanConnect).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(anotherChannel2);
 
         ConnectionTestUtils.verifyPublishIsReceived(anotherChannel2, MqttQoS.AT_LEAST_ONCE, firstPayload);
@@ -245,7 +236,7 @@ public class PostOfficeUnsubscribeTest {
 
         EmbeddedChannel anotherChannel3 = new EmbeddedChannel();
         MQTTConnection anotherConnection3 = createMQTTConnection(CONFIG, anotherChannel3);
-        anotherConnection3.processConnect(notCleanConnect).get();
+        anotherConnection3.processConnect(notCleanConnect).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(anotherChannel3);
 
         ConnectionTestUtils.verifyPublishIsReceived(anotherChannel3, MqttQoS.AT_LEAST_ONCE, secondPayload);
@@ -270,24 +261,14 @@ public class PostOfficeUnsubscribeTest {
     private MQTTConnection connectNotCleanAs(String clientId) {
         EmbeddedChannel channel = new EmbeddedChannel();
         MQTTConnection connection = createMQTTConnection(CONFIG, channel);
-        try {
-            connection.processConnect(ConnectionTestUtils.buildConnectNotClean(clientId)).get();
-        } catch (InterruptedException | ExecutionException e) {
-            Assertions.fail(e);
-        }
-        ConnectionTestUtils.assertConnectAccepted(channel);
+        ConnectionTestUtils.connect(connection, ConnectionTestUtils.buildConnectNotClean(clientId));
         return connection;
     }
 
     private MQTTConnection connectAs(String clientId) {
         EmbeddedChannel channel = new EmbeddedChannel();
         MQTTConnection connection = createMQTTConnection(CONFIG, channel);
-        try {
-            connection.processConnect(ConnectionTestUtils.buildConnect(clientId)).get();
-        } catch (InterruptedException | ExecutionException e) {
-            Assertions.fail(e);
-        }
-        ConnectionTestUtils.assertConnectAccepted(channel);
+        ConnectionTestUtils.connect(connection, ConnectionTestUtils.buildConnect(clientId));
         return connection;
     }
 
@@ -311,7 +292,7 @@ public class PostOfficeUnsubscribeTest {
         final MqttConnectMessage notCleanConnect = ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID);
         EmbeddedChannel subscriberChannel = new EmbeddedChannel();
         MQTTConnection subscriberConnection = createMQTTConnection(CONFIG, subscriberChannel);
-        subscriberConnection.processConnect(notCleanConnect).get();
+        subscriberConnection.processConnect(notCleanConnect).completableFuture().get();
         ConnectionTestUtils.assertConnectAccepted(subscriberChannel);
 
         subscribe(subscriberConnection, NEWS_TOPIC, AT_MOST_ONCE);
@@ -326,7 +307,7 @@ public class PostOfficeUnsubscribeTest {
 
         ConnectionTestUtils.verifyPublishIsReceived(subscriberChannel, AT_MOST_ONCE, "Hello world2!");
 
-        subscriberConnection.processDisconnect(null).get();
+        subscriberConnection.processDisconnect(null).completableFuture().get();
 
         assertFalse(subscriberChannel.isOpen(), "after a disconnect the client should be disconnected");
     }
@@ -341,7 +322,7 @@ public class PostOfficeUnsubscribeTest {
         // publish from another channel
         publishQos1(publisher, NEWS_TOPIC, "Hello world MQTT!!-1", 99);
         ConnectionTestUtils.verifyPublishIsReceived(channel, AT_LEAST_ONCE, "Hello world MQTT!!-1");
-        connection.processDisconnect(null).get();
+        connection.processDisconnect(null).completableFuture().get();
 
         publishQos1(publisher, NEWS_TOPIC, "Hello world MQTT!!-2", 100);
         publishQos1(publisher, NEWS_TOPIC, "Hello world MQTT!!-3", 101);
@@ -360,7 +341,7 @@ public class PostOfficeUnsubscribeTest {
                     .payload(bytePayload)
                     .qos(MqttQoS.AT_LEAST_ONCE)
                     .retained(false)
-                    .topicName(NEWS_TOPIC).build()).get(5, TimeUnit.SECONDS);
+                    .topicName(NEWS_TOPIC).build()).completableFuture().get(5, TimeUnit.SECONDS);
         } catch (InterruptedException | ExecutionException | TimeoutException e) {
             throw new RuntimeException(e);
         }
@@ -372,6 +353,7 @@ public class PostOfficeUnsubscribeTest {
             .payload(bytePayload)
             .qos(MqttQoS.EXACTLY_ONCE)
             .retained(true)
+            .messageId(1)
             .topicName(topic).build(), "username");
     }
 
@@ -382,7 +364,7 @@ public class PostOfficeUnsubscribeTest {
      */
     @Test
     public void checkQoS2SubscriberDisconnectReceivePersistedPublishes() {
-        connect(this.connection, FAKE_CLIENT_ID);
+        connect(connection, FAKE_CLIENT_ID);
         subscribe(connection, NEWS_TOPIC, EXACTLY_ONCE);
         connection.processDisconnect(null);
 
@@ -401,7 +383,7 @@ public class PostOfficeUnsubscribeTest {
      * message "hello" on "a/b" subscriber must receive only a single message not twice
      */
     @Test
-    public void checkSinglePublishOnOverlappingSubscriptions() {
+    public void checkSinglePubPostOfficeUnsubscribeTestlishOnOverlappingSubscriptions() {
         final MQTTConnection publisher = connectAs("Publisher");
 
         connect(this.connection, FAKE_CLIENT_ID);
diff --git a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
index d6884c42..78d1c9f8 100644
--- a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
+++ b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
@@ -120,9 +120,9 @@ public class SessionRegistryTest {
     public void connectWithCleanSessionUpdateClientSession() throws ExecutionException, InterruptedException {
         // first connect with clean session true
         MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).cleanSession(true).build();
-        connection.processConnect(msg).get();
+        connection.processConnect(msg).completableFuture().get();
         assertEqualsConnAck(CONNECTION_ACCEPTED, channel.readOutbound());
-        connection.processDisconnect(null).get();
+        connection.processDisconnect(null).completableFuture().get();
         assertFalse(channel.isOpen());
 
         // second connect with clean session false
@@ -134,7 +134,7 @@ public class SessionRegistryTest {
             .protocolVersion(MqttVersion.MQTT_3_1)
             .build();
 
-        anotherConnection.processConnect(secondConnMsg).get();
+        anotherConnection.processConnect(secondConnMsg).completableFuture().get();
         assertEqualsConnAck(CONNECTION_ACCEPTED, anotherChannel.readOutbound());
 
         // Verify client session is clean false
diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationRestartTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationRestartTest.java
index d8dd23fc..705a855c 100644
--- a/broker/src/test/java/io/moquette/integration/ServerIntegrationRestartTest.java
+++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationRestartTest.java
@@ -33,6 +33,8 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.nio.file.Path;
@@ -42,6 +44,7 @@ import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 public class ServerIntegrationRestartTest {
+    private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationRestartTest.class);
 
     static MqttConnectOptions CLEAN_SESSION_OPT = new MqttConnectOptions();
 
@@ -101,6 +104,7 @@ public class ServerIntegrationRestartTest {
     @DisplayName("given not clean session after a server restart, the session is still present")
     @Test
     public void testNotCleanSessionIsVisibleAfterServerRestart() throws Exception {
+        LOG.info("*** testNotCleanSessionIsVisibleAfterServerRestart ***");
         m_subscriber.connect(CLEAN_SESSION_OPT);
         m_subscriber.subscribe("/topic", 1);
         m_subscriber.disconnect();
@@ -123,6 +127,7 @@ public class ServerIntegrationRestartTest {
 
     @Test
     public void checkRestartCleanSubscriptionTree() throws Exception {
+        LOG.info("*** checkRestartCleanSubscriptionTree ***");
         // subscribe to /topic
         m_subscriber.connect(CLEAN_SESSION_OPT);
         m_subscriber.subscribe("/topic", 1);
@@ -154,6 +159,7 @@ public class ServerIntegrationRestartTest {
 
     @Test
     public void checkDontPublishInactiveClientsAfterServerRestart() throws Exception {
+        LOG.info("*** checkDontPublishInactiveClientsAfterServerRestart ***");
         IMqttClient conn = subscribeAndPublish("/topic");
         conn.disconnect();
 
@@ -169,6 +175,7 @@ public class ServerIntegrationRestartTest {
 
     @Test
     public void testClientDoesntRemainSubscribedAfterASubscriptionAndServerRestart() throws Exception {
+        LOG.info("*** testClientDoesntRemainSubscribedAfterASubscriptionAndServerRestart ***");
         // subscribe to /topic
         m_subscriber.connect();
         // subscribe /topic