Repository: moquette-io/moquette
Pull request #631 (2021-09-15 19:11:23)

Adds session events processors and queues of commands
## Release notes
Fix: introduces sessions event processors to segregate changes to a session in one single thread, simplifying concurrency and code

## What does this PR do?
Introduces the concept of sticky sessions, all the change operations to a session is done always by the same thread, so this simplifies the concurrency access, segregating the access to only one thread. The same thread can handle multiple sessions.

Every session event processor is an event loop, that peeks commands out of a queue and executes them.
This means that there is an isolation queue between the IO threads that executes the lower layer of protocol  and the event loops that mutate and managed the upper logic, mainly bound to Session's related structures.

The length of those queue can be configured with option `session_queue_size`, by default is has 1024 slots, but it could be that if there is an heavy publisher, the queue becomes full and this case the only thing to do is log an error. So it could happen that for a SUB command no SUBACK is sent (the same is valid for PUB) if a queue is full. In MQTT 3.1.1 there is no timeout specification for missed SUBACK, so the client will not send a second SUB (dup=1) and we potentially could loose SUBscriptions (and PUBlishes).

This commit also implement a couple of optimizations on this event loop side:
- if a command processed by an EventLoop generated the sending of other commands to the same EventLoop, then those commands are inlined and doesn't pass through the queue.
- when a single PUB command involves many subscribers, then all the subsequent PUB that goes to the same queue are compressed on only one command on the queue, instead of insert many single PUB commands

## 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
-->
- Relates  #608 



Patch:
diff --git a/ChangeLog.txt b/ChangeLog.txt
index e970ab75..c18fa96d 100644
--- a/ChangeLog.txt
+++ b/ChangeLog.txt
@@ -1,5 +1,6 @@
 Version 0.16-SNAPSHOT:
    [build] drop generation of broker-test, removed distribution and embedding_moquette modules from deploy phase (#616)
+   [fix] introduces sessions event processors to segregate changes to a session in one single thread, simplifying concurrency and code (#631)
 
 Version 0.15.1:
    [fix] avoid double subscription (#612)
diff --git a/broker/pom.xml b/broker/pom.xml
index 2c68fa29..a489968f 100644
--- a/broker/pom.xml
+++ b/broker/pom.xml
@@ -196,26 +196,26 @@
 <!--                </executions>-->
 <!--            </plugin>-->
 
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-compiler-plugin</artifactId>
-                <version>3.8.0</version>
-                <configuration>
-                    <source>8</source>
-                    <target>8</target>
-                    <compilerArgs>
-                        <arg>-XDcompilePolicy=simple</arg>
-                        <arg>-Xplugin:ErrorProne</arg>
-                    </compilerArgs>
-                    <annotationProcessorPaths>
-                        <path>
-                            <groupId>com.google.errorprone</groupId>
-                            <artifactId>error_prone_core</artifactId>
-                            <version>2.4.0</version>
-                        </path>
-                    </annotationProcessorPaths>
-                </configuration>
-            </plugin>
+<!--            <plugin>-->
+<!--                <groupId>org.apache.maven.plugins</groupId>-->
+<!--                <artifactId>maven-compiler-plugin</artifactId>-->
+<!--                <version>3.8.0</version>-->
+<!--                <configuration>-->
+<!--                    <source>1.8</source>-->
+<!--                    <target>1.8</target>-->
+<!--                    <compilerArgs>-->
+<!--                        <arg>-XDcompilePolicy=simple</arg>-->
+<!--                        <arg>-Xplugin:ErrorProne</arg>-->
+<!--                    </compilerArgs>-->
+<!--                    <annotationProcessorPaths>-->
+<!--                        <path>-->
+<!--                            <groupId>com.google.errorprone</groupId>-->
+<!--                            <artifactId>error_prone_core</artifactId>-->
+<!--                            <version>2.4.0</version>-->
+<!--                        </path>-->
+<!--                    </annotationProcessorPaths>-->
+<!--                </configuration>-->
+<!--            </plugin>-->
 
             <plugin>
                 <groupId>com.github.spotbugs</groupId>
diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java
index 4949fd46..95a60df9 100644
--- a/broker/src/main/java/io/moquette/BrokerConstants.java
+++ b/broker/src/main/java/io/moquette/BrokerConstants.java
@@ -34,6 +34,7 @@ public final class BrokerConstants {
     public static final String WSS_PORT_PROPERTY_NAME = "secure_websocket_port";
     public static final String WEB_SOCKET_PATH_PROPERTY_NAME = "websocket_path";
     public static final String WEB_SOCKET_MAX_FRAME_SIZE_PROPERTY_NAME = "websocket_max_frame_size";
+    public static final String SESSION_QUEUE_SIZE = "session_queue_size";
 
     /**
      * Defines the SSL implementation to use, default to "JDK".
diff --git a/broker/src/main/java/io/moquette/broker/DebugUtils.java b/broker/src/main/java/io/moquette/broker/DebugUtils.java
index 02779114..15972ee5 100644
--- a/broker/src/main/java/io/moquette/broker/DebugUtils.java
+++ b/broker/src/main/java/io/moquette/broker/DebugUtils.java
@@ -29,6 +29,7 @@ public final class DebugUtils {
             final int size = copy.readableBytes();
             bytesContent = new byte[size];
             copy.readBytes(bytesContent);
+            copy.release();
         } else {
             bytesContent = copy.array();
         }
diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnection.java b/broker/src/main/java/io/moquette/broker/MQTTConnection.java
index 83fd5acb..4781504b 100644
--- a/broker/src/main/java/io/moquette/broker/MQTTConnection.java
+++ b/broker/src/main/java/io/moquette/broker/MQTTConnection.java
@@ -30,6 +30,8 @@ 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;
 
@@ -38,7 +40,6 @@ import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE;
 import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
 import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
 import static io.netty.handler.codec.mqtt.MqttQoS.*;
-import io.netty.util.ReferenceCountUtil;
 
 final class MQTTConnection {
 
@@ -108,12 +109,18 @@ final class MQTTConnection {
 
     private void processPubComp(MqttMessage msg) {
         final int messageID = ((MqttMessageIdVariableHeader) msg.variableHeader()).messageId();
-        bindedSession.processPubComp(messageID);
+        this.postOffice.routeCommand(bindedSession.getClientID(), () -> {
+            bindedSession.processPubComp(messageID);
+            return bindedSession.getClientID();
+        });
     }
 
     private void processPubRec(MqttMessage msg) {
         final int messageID = ((MqttMessageIdVariableHeader) msg.variableHeader()).messageId();
-        bindedSession.processPubRec(messageID);
+        this.postOffice.routeCommand(bindedSession.getClientID(), () -> {
+            bindedSession.processPubRec(messageID);
+            return null;
+        });
     }
 
     static MqttMessage pubrel(int messageID) {
@@ -123,10 +130,14 @@ final class MQTTConnection {
 
     private void processPubAck(MqttMessage msg) {
         final int messageID = ((MqttMessageIdVariableHeader) msg.variableHeader()).messageId();
-        bindedSession.pubAckReceived(messageID);
+        final String clientId = getClientId();
+        this.postOffice.routeCommand(clientId, () -> {
+            bindedSession.pubAckReceived(messageID);
+            return null;
+        });
     }
 
-    void processConnect(MqttConnectMessage msg) {
+    Future<PostOffice.RouteResult> processConnect(MqttConnectMessage msg) {
         MqttConnectPayload payload = msg.payload();
         String clientId = payload.clientIdentifier();
         final String username = payload.userName();
@@ -135,20 +146,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;
+            return CompletableFuture.completedFuture(null);
         }
         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;
+                return CompletableFuture.completedFuture(null);
             }
 
             if (!cleanSession) {
                 LOG.info("MQTT client ID cannot be empty for persistent session. Username: {}", username);
                 abortConnection(CONNECTION_REFUSED_IDENTIFIER_REJECTED);
-                return;
+                return CompletableFuture.completedFuture(null);
             }
 
             // Generating client id.
@@ -159,9 +170,20 @@ final class MQTTConnection {
         if (!login(msg, clientId)) {
             abortConnection(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
             channel.close().addListener(CLOSE_ON_FAILURE);
-            return;
+            return CompletableFuture.completedFuture(null);
         }
 
+        final String sessionId = clientId;
+        return this.postOffice.routeCommand(clientId, () -> {
+            executeConnect(msg, sessionId);
+            return null;
+        });
+    }
+
+    /**
+     * Invoked by the Session's event loop.
+     * */
+    private void executeConnect(MqttConnectMessage msg, String clientId) {
         final SessionRegistry.SessionCreationResult result;
         try {
             LOG.trace("Binding MQTTConnection to session");
@@ -203,7 +225,7 @@ final class MQTTConnection {
                         setupInflightResender(channel);
 
                         postOffice.dispatchConnection(msg);
-                        LOG.trace("dispatch connection: {}", msg.toString());
+                        LOG.trace("dispatch connection: {}", msg);
                     }
                 } else {
                     bindedSession.disconnect();
@@ -276,11 +298,19 @@ final class MQTTConnection {
     }
 
     void handleConnectionLost() {
-        String clientID = NettyUtils.clientID(channel);
+        final String clientID = NettyUtils.clientID(channel);
         if (clientID == null || clientID.isEmpty()) {
             return;
         }
+        // this must not be done on the netty thread
         LOG.info("Notifying connection lost event");
+        postOffice.routeCommand(clientID, () -> {
+            processConnectionLost(clientID);
+            return null;
+        });
+    }
+
+    private void processConnectionLost(String clientID) {
         if (bindedSession.hasWill()) {
             postOffice.fireWill(bindedSession.getWill());
         }
@@ -305,30 +335,38 @@ final class MQTTConnection {
         channel.close().addListener(FIRE_EXCEPTION_ON_FAILURE);
     }
 
-    void processDisconnect(MqttMessage msg) {
+    Future<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;
+            return CompletableFuture.completedFuture(null);
         }
-        bindedSession.disconnect();
-        connected = false;
-        channel.close().addListener(FIRE_EXCEPTION_ON_FAILURE);
-        LOG.trace("Processed DISCONNECT");
-        String userName = NettyUtils.userName(channel);
-        postOffice.dispatchDisconnection(clientID, userName);
-        LOG.trace("dispatch disconnection userName={}", userName);
+
+        return this.postOffice.routeCommand(clientID, () -> {
+            bindedSession.disconnect();
+            connected = false;
+            channel.close().addListener(FIRE_EXCEPTION_ON_FAILURE);
+            LOG.trace("Processed DISCONNECT");
+            String userName = NettyUtils.userName(channel);
+            postOffice.clientDisconnected(clientID, userName);
+            LOG.trace("dispatch disconnection userName={}", userName);
+            return null;
+        });
     }
 
-    void processSubscribe(MqttSubscribeMessage msg) {
+    Future<PostOffice.RouteResult> processSubscribe(MqttSubscribeMessage msg) {
         final String clientID = NettyUtils.clientID(channel);
         if (!connected) {
             LOG.warn("SUBSCRIBE received on already closed connection");
             dropConnection();
-            return;
+            return CompletableFuture.completedFuture(null);
         }
-        postOffice.subscribeClientToTopics(msg, clientID, NettyUtils.userName(channel), this);
+        final String username = NettyUtils.userName(channel);
+        return postOffice.routeCommand(clientID, () -> {
+            postOffice.subscribeClientToTopics(msg, clientID, username, this);
+            return null;
+        });
     }
 
     void sendSubAckMessage(int messageID, MqttSubAckMessage ackMessage) {
@@ -338,10 +376,14 @@ final class MQTTConnection {
 
     private void processUnsubscribe(MqttUnsubscribeMessage msg) {
         List<String> topics = msg.payload().topics();
-        String clientID = NettyUtils.clientID(channel);
+        final String clientID = NettyUtils.clientID(channel);
+        final int messageId = msg.variableHeader().messageId();
 
-        LOG.trace("Processing UNSUBSCRIBE message. topics: {}", topics);
-        postOffice.unsubscribe(topics, this, msg.variableHeader().messageId());
+        postOffice.routeCommand(clientID, () -> {
+            LOG.trace("Processing UNSUBSCRIBE message. topics: {}", topics);
+            postOffice.unsubscribe(topics, this, messageId);
+            return null;
+        });
     }
 
     void sendUnsubAckMessage(List<String> topics, String clientID, int messageID) {
@@ -354,7 +396,7 @@ final class MQTTConnection {
         LOG.trace("Client unsubscribed from topics <{}>", topics);
     }
 
-    void processPublish(MqttPublishMessage msg) {
+    Future<? extends Object> processPublish(MqttPublishMessage msg) {
         final MqttQoS qos = msg.fixedHeader().qosLevel();
         final String username = NettyUtils.userName(channel);
         final String topicName = msg.variableHeader().topicName();
@@ -366,26 +408,32 @@ final class MQTTConnection {
             LOG.debug("Drop connection because of invalid topic format");
             dropConnection();
         }
+
+        // retain else msg is cleaned by the NewNettyMQTTHandler and is not available
+        // in execution by SessionEventLoop
+        msg.retain();
         switch (qos) {
             case AT_MOST_ONCE:
-                postOffice.receivedPublishQos0(topic, username, clientId, msg);
-                break;
+                return postOffice.receivedPublishQos0(topic, username, clientId, msg);
             case AT_LEAST_ONCE: {
-                postOffice.receivedPublishQos1(this, topic, username, messageID, msg);
-                break;
+                return postOffice.receivedPublishQos1(this, topic, username, messageID, msg);
             }
             case EXACTLY_ONCE: {
-                bindedSession.receivedPublishQos2(messageID, msg);
-                postOffice.receivedPublishQos2(this, msg, username);
-                break;
+                final CompletableFuture<PostOffice.RouteResult> firstStepFuture = postOffice.routeCommand(clientId, () -> {
+                    bindedSession.receivedPublishQos2(messageID, msg);
+                    return null;
+                });
+                return firstStepFuture.thenCompose(v -> postOffice.receivedPublishQos2(this, msg, username));
             }
             default:
                 LOG.error("Unknown QoS-Type:{}", qos);
-                break;
+                final CompletableFuture<Void> failed = new CompletableFuture<>();
+                failed.completeExceptionally(new Error("Unknown QoS-"));
+                return failed;
         }
     }
 
-    void sendPublishReceived(int messageID) {
+    void sendPubRec(int messageID) {
         LOG.trace("sendPubRec invoked, messageID: {}", messageID);
         MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC, false, AT_MOST_ONCE,
             false, 0);
@@ -395,6 +443,13 @@ final class MQTTConnection {
 
     private void processPubRel(MqttMessage msg) {
         final int messageID = ((MqttMessageIdVariableHeader) msg.variableHeader()).messageId();
+        this.postOffice.routeCommand(bindedSession.getClientID(), () -> {
+            executePubRel(messageID);
+            return null;
+        });
+    }
+
+    private void executePubRel(int messageID) {
         bindedSession.receivedPubRelQos2(messageID);
         sendPubCompMessage(messageID);
     }
@@ -533,4 +588,8 @@ final class MQTTConnection {
     public void flush() {
         channel.flush();
     }
+
+    public void bindSession(Session session) {
+        bindedSession = session;
+    }
 }
diff --git a/broker/src/main/java/io/moquette/broker/PostOffice.java b/broker/src/main/java/io/moquette/broker/PostOffice.java
index ab24b6a5..e23ba2ec 100644
--- a/broker/src/main/java/io/moquette/broker/PostOffice.java
+++ b/broker/src/main/java/io/moquette/broker/PostOffice.java
@@ -22,12 +22,19 @@ import io.moquette.broker.subscriptions.Topic;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.handler.codec.mqtt.*;
+import io.netty.util.ReferenceCountUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.*;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
 import static io.moquette.broker.Utils.messageId;
@@ -36,21 +43,153 @@ import static io.netty.handler.codec.mqtt.MqttQoS.*;
 
 class PostOffice {
 
+    /**
+     * Maps the failed packetID per clientId (id client source, id_packet) -> [id client target]
+     * */
+    private static class FailedPublishCollection {
+
+        static class PacketId {
+
+            private final String clientId;
+            private final int idPacket;
+
+            PacketId(String clientId, int idPacket) {
+                this.clientId = clientId;
+                this.idPacket = idPacket;
+            }
+            @Override
+            public boolean equals(Object o) {
+                if (this == o) return true;
+                if (o == null || getClass() != o.getClass()) return false;
+                PacketId packetId = (PacketId) o;
+                return idPacket == packetId.idPacket && Objects.equals(clientId, packetId.clientId);
+            }
+
+            @Override
+            public int hashCode() {
+                return Objects.hash(clientId, idPacket);
+            }
+
+            public boolean belongToClient(String clientId) {
+                return this.clientId.equals(clientId);
+            }
+        }
+
+        private final ConcurrentMap<PacketId, Set<String>> packetsMap = new ConcurrentHashMap<>();
+
+        private void insert(String clientId, int messageID, String failedClientId) {
+            final PacketId packetId = new PacketId(clientId, messageID);
+            packetsMap.computeIfAbsent(packetId, k -> new HashSet<>())
+                .add(failedClientId);
+        }
+
+        public void remove(String clientId, int messageID, String targetClientId) {
+            final PacketId packetId = new PacketId(clientId, messageID);
+            packetsMap.computeIfPresent(packetId, (key, clientsSet) -> {
+               clientsSet.remove(targetClientId);
+               if (clientsSet.isEmpty()) {
+                   // the mapping key, value is removed
+                   return null;
+               } else {
+                   return clientsSet;
+               }
+            });
+        }
+
+        private void removeAll(int messageID, String clientId, Collection<String> routings) {
+            for (String targetClientId : routings) {
+                remove(clientId, messageID, targetClientId);
+            }
+        }
+
+        void cleanupForClient(String clientId) {
+            // the only way to linear scan the map to collect of PacketIds references,
+            // for a following step of cleaning
+            packetsMap.keySet().stream()
+                .filter(packet -> packet.belongToClient(clientId))
+                .forEach(packetsMap::remove);
+        }
+
+        void insertAll(int messageID, String clientId, Collection<String> routings) {
+            for (String targetClientId : routings) {
+                insert(clientId, messageID, targetClientId);
+            }
+        }
+
+        Set<String> listFailed(String clientId, int messageID) {
+            final PacketId packetId = new PacketId(clientId, messageID);
+            return packetsMap.getOrDefault(packetId, Collections.emptySet());
+        }
+    }
+
+    static class RoutingResults {
+
+        private final List<String> successedRoutings;
+        private final List<String> failedRoutings;
+
+        public RoutingResults(List<String> successedRoutings, List<String> failedRoutings) {
+            this.successedRoutings = successedRoutings;
+            this.failedRoutings = failedRoutings;
+        }
+
+        public boolean isAllSuccess() {
+            return failedRoutings.isEmpty();
+        }
+    }
+
+    static class RouteResult {
+        private final String clientId;
+        private final Status status;
+
+        enum Status {SUCCESS, FAIL}
+
+        public static RouteResult success(String clientId) {
+            return new RouteResult(clientId, Status.SUCCESS);
+        }
+
+        public static RouteResult failed(String clientId) {
+            return new RouteResult(clientId, Status.FAIL);
+        }
+
+        private RouteResult(String clientId, RouteResult.Status status) {
+            this.clientId = clientId;
+            this.status = status;
+        }
+    }
+
     private static final Logger LOG = LoggerFactory.getLogger(PostOffice.class);
 
+    private static final Set<String> NO_FILTER = new HashSet<>();
+
     private final Authorizator authorizator;
     private final ISubscriptionsDirectory subscriptions;
     private final IRetainedRepository retainedRepository;
     private SessionRegistry sessionRegistry;
     private BrokerInterceptor interceptor;
 
+    private final Thread[] sessionExecutors;
+    private final BlockingQueue<FutureTask<String>>[] sessionQueues;
+    private final int eventLoops = Runtime.getRuntime().availableProcessors();
+    private final FailedPublishCollection failedPublishes = new FailedPublishCollection();
+
     PostOffice(ISubscriptionsDirectory subscriptions, IRetainedRepository retainedRepository,
-               SessionRegistry sessionRegistry, BrokerInterceptor interceptor, Authorizator authorizator) {
+               SessionRegistry sessionRegistry, BrokerInterceptor interceptor, Authorizator authorizator, int sessionQueueSize) {
         this.authorizator = authorizator;
         this.subscriptions = subscriptions;
         this.retainedRepository = retainedRepository;
         this.sessionRegistry = sessionRegistry;
         this.interceptor = interceptor;
+
+        this.sessionQueues = new BlockingQueue[eventLoops];
+        for (int i = 0; i < eventLoops; i++) {
+            this.sessionQueues[i] = new ArrayBlockingQueue<>(sessionQueueSize);
+        }
+        this.sessionExecutors = new Thread[eventLoops];
+        for (int i = 0; i < eventLoops; i++) {
+            this.sessionExecutors[i] = new Thread(new SessionEventLoop(this.sessionQueues[i]));
+            this.sessionExecutors[i].setName("Session Executor " + i);
+            this.sessionExecutors[i].start();
+        }
     }
 
     public void init(SessionRegistry sessionRegistry) {
@@ -135,6 +274,13 @@ class PostOffice {
     public void unsubscribe(List<String> topics, MQTTConnection mqttConnection, int messageId) {
         final String clientID = mqttConnection.getClientId();
         final Session session = sessionRegistry.retrieve(clientID);
+        if (session == null) {
+            // Session is already destroyed.
+            // Just ack the client
+            LOG.warn("Session not found when unsubscribing {}", clientID);
+            mqttConnection.sendUnsubAckMessage(topics, clientID, messageId);
+            return;
+        }
         for (String t : topics) {
             Topic topic = new Topic(t);
             boolean validTopic = topic.isValid();
@@ -158,102 +304,225 @@ class PostOffice {
         mqttConnection.sendUnsubAckMessage(topics, clientID, messageId);
     }
 
-    void receivedPublishQos0(Topic topic, String username, String clientID, MqttPublishMessage msg) {
+    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);
-            return;
+            return CompletableFuture.completedFuture(null);
         }
-        publish2Subscribers(msg.payload(), topic, AT_MOST_ONCE);
+        final CompletableFuture<RoutingResults> publishFuture = publish2Subscribers(msg.payload(), topic, AT_MOST_ONCE);
 
-        if (msg.fixedHeader().isRetain()) {
-            // QoS == 0 && retain => clean old retained
-            retainedRepository.cleanRetained(topic);
-        }
+        return publishFuture.thenRun(() -> {
+            if (msg.fixedHeader().isRetain()) {
+                // QoS == 0 && retain => clean old retained
+                retainedRepository.cleanRetained(topic);
+            }
 
-        interceptor.notifyTopicPublished(msg, clientID, username);
+            interceptor.notifyTopicPublished(msg, clientID, username);
+        });
     }
 
-    void receivedPublishQos1(MQTTConnection connection, Topic topic, String username, int messageID,
-                             MqttPublishMessage msg) {
-        // verify if topic can be write
+    CompletableFuture<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;
+            return CompletableFuture.completedFuture(null);
         }
         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;
+            return CompletableFuture.completedFuture(null);
         }
 
         ByteBuf payload = msg.payload();
-        publish2Subscribers(payload, topic, AT_LEAST_ONCE);
+        final CompletableFuture<RoutingResults> publishFuture;
+        if (msg.fixedHeader().isDup()) {
+            final Set<String> failedClients = failedPublishes.listFailed(clientId, messageID);
+            publishFuture = publish2Subscribers(payload, topic, AT_LEAST_ONCE, failedClients);
+        } else {
+            publishFuture = publish2Subscribers(payload, topic, AT_LEAST_ONCE);
+        }
 
-        connection.sendPubAck(messageID);
+        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, routings.successedRoutings);
+        });
+    }
 
+    private void manageRetain(Topic topic, MqttPublishMessage msg) {
         if (msg.fixedHeader().isRetain()) {
-            if (!payload.isReadable()) {
+            if (!msg.payload().isReadable()) {
                 retainedRepository.cleanRetained(topic);
             } else {
                 // before wasn't stored
                 retainedRepository.retain(topic, msg);
             }
         }
-        interceptor.notifyTopicPublished(msg, clientId, username);
     }
 
-    private void publish2Subscribers(ByteBuf payload, Topic topic, MqttQoS publishingQos) {
+    private CompletableFuture<RoutingResults> publish2Subscribers(ByteBuf payload, Topic topic, MqttQoS publishingQos) {
+        return publish2Subscribers(payload, topic, publishingQos, NO_FILTER);
+    }
+
+    private class BatchingPublishesCollector {
+        final List<Subscription>[] subscriptions;
+        private final int eventLoops;
+
+        BatchingPublishesCollector(int eventLoops) {
+            subscriptions = new List[eventLoops];
+            this.eventLoops = eventLoops;
+        }
+
+        public void add(Subscription sub) {
+            final int targetQueueId = subscriberEventLoop(sub.getClientId());
+            if (subscriptions[targetQueueId] == null) {
+                subscriptions[targetQueueId] = new ArrayList<>();
+            }
+            subscriptions[targetQueueId].add(sub);
+        }
+
+        private int subscriberEventLoop(String clientId) {
+            return Math.abs(clientId.hashCode()) % this.eventLoops;
+        }
+
+        List<CompletableFuture<RouteResult>> routeBatchedPublishes(Consumer<List<Subscription>> action) {
+            List<CompletableFuture<RouteResult>> publishFutures = new ArrayList<>(this.eventLoops);
+
+            for (List<Subscription> subscriptionsBatch : subscriptions) {
+                if (subscriptionsBatch == null) {
+                    continue;
+                }
+                final String clientId = subscriptionsBatch.get(0).getClientId();
+                if (LOG.isTraceEnabled()) {
+                    final String subscriptionsDetails = subscriptionsBatch.stream()
+                        .map(Subscription::toString)
+                        .collect(Collectors.joining(",\n"));
+                    final int loopId = subscriberEventLoop(clientId);
+                    LOG.trace("Routing PUBLISH to eventLoop {}  for subscriptions [{}]", loopId, subscriptionsDetails);
+                }
+                publishFutures.add(routeCommand(clientId, () -> {
+                    action.accept(subscriptionsBatch);
+                    return null;
+                }));
+            }
+            return publishFutures;
+        }
+
+        Collection<String> subscriberIdsByEventLoop(String clientId) {
+            final int targetQueueId = subscriberEventLoop(clientId);
+            return subscriptions[targetQueueId].stream().map(Subscription::getClientId).collect(Collectors.toList());
+        }
+    }
+
+    private CompletableFuture<RoutingResults> publish2Subscribers(ByteBuf payload, Topic topic, MqttQoS publishingQos,
+                                                                  Set<String> filterTargetClients) {
         Set<Subscription> topicMatchingSubscriptions = subscriptions.matchQosSharpening(topic);
 
+        final BatchingPublishesCollector collector = new BatchingPublishesCollector(eventLoops);
+
         for (final Subscription sub : topicMatchingSubscriptions) {
-            MqttQoS qos = lowerQosToTheSubscriptionDesired(sub, publishingQos);
-            Session targetSession = this.sessionRegistry.retrieve(sub.getClientId());
+            if (filterTargetClients == NO_FILTER || filterTargetClients.contains(sub.getClientId())) {
+                collector.add(sub);
+            }
+        }
 
-            boolean isSessionPresent = targetSession != null;
-            if (isSessionPresent) {
-                LOG.debug("Sending PUBLISH message to active subscriber CId: {}, topicFilter: {}, qos: {}",
-                          sub.getClientId(), sub.getTopicFilter(), qos);
-                targetSession.sendPublishOnSessionAtQos(topic, qos, payload);
-            } else {
-                // If we are, the subscriber disconnected after the subscriptions tree selected that session as a
-                // destination.
-                LOG.debug("PUBLISH to not yet present session. CId: {}, topicFilter: {}, qos: {}", sub.getClientId(),
-                          sub.getTopicFilter(), qos);
+        List<CompletableFuture<RouteResult>> publishFutures = collector.routeBatchedPublishes((batch) -> {
+            publishToSession(payload, topic, batch, publishingQos);
+        });
+
+        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);
+                }
             }
+            return new RoutingResults(successedRoutings, failedRoutings);
+        });
+    }
+
+    private void publishToSession(ByteBuf payload, Topic topic, Collection<Subscription> subscriptions, MqttQoS publishingQos) {
+        for (Subscription sub : subscriptions) {
+            MqttQoS qos = lowerQosToTheSubscriptionDesired(sub, publishingQos);
+            publishToSession(payload, topic, sub, qos);
+        }
+    }
+
+    private void publishToSession(ByteBuf payload, Topic topic, Subscription sub, MqttQoS qos) {
+        Session targetSession = this.sessionRegistry.retrieve(sub.getClientId());
+
+        boolean isSessionPresent = targetSession != null;
+        if (isSessionPresent) {
+            LOG.debug("Sending PUBLISH message to active subscriber CId: {}, topicFilter: {}, qos: {}",
+                      sub.getClientId(), sub.getTopicFilter(), qos);
+            targetSession.sendPublishOnSessionAtQos(topic, qos, payload);
+        } else {
+            // If we are, the subscriber disconnected after the subscriptions tree selected that session as a
+            // destination.
+            LOG.debug("PUBLISH to not yet present session. CId: {}, topicFilter: {}, qos: {}", sub.getClientId(),
+                      sub.getTopicFilter(), qos);
         }
     }
 
     /**
      * First phase of a publish QoS2 protocol, sent by publisher to the broker. Publish to all interested
      * subscribers.
+     * @return
      */
-    void receivedPublishQos2(MQTTConnection connection, MqttPublishMessage mqttPublishMessage, String username) {
-        LOG.trace("Processing PUBREL message on connection: {}", connection);
-        final Topic topic = new Topic(mqttPublishMessage.variableHeader().topicName());
-        final ByteBuf payload = mqttPublishMessage.payload();
+    CompletableFuture<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();
 
         final String clientId = connection.getClientId();
         if (!authorizator.canWrite(topic, username, clientId)) {
             LOG.error("MQTT client is not authorized to publish on topic: {}", topic);
-            return;
+            return CompletableFuture.completedFuture(null);
         }
 
-        publish2Subscribers(payload, topic, EXACTLY_ONCE);
+        final int messageID = msg.variableHeader().packetId();
+        final CompletableFuture<RoutingResults> publishFuture;
+        if (msg.fixedHeader().isDup()) {
+            final Set<String> failedClients = failedPublishes.listFailed(clientId, messageID);
+            publishFuture = publish2Subscribers(payload, topic, EXACTLY_ONCE, failedClients);
+        } else {
+            publishFuture = publish2Subscribers(payload, topic, EXACTLY_ONCE);
+        }
 
-        final boolean retained = mqttPublishMessage.fixedHeader().isRetain();
-        if (retained) {
-            if (!payload.isReadable()) {
-                retainedRepository.cleanRetained(topic);
+        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 {
-                // before wasn't stored
-                retainedRepository.retain(topic, mqttPublishMessage);
+                // some session event loop enqueue raised a problem
+                failedPublishes.insertAll(messageID, clientId, routings.failedRoutings);
             }
-        }
+            ReferenceCountUtil.release(msg);
 
-        String clientID = connection.getClientId();
-        interceptor.notifyTopicPublished(mqttPublishMessage, clientID, username);
+            // cleanup success resends from the failed publishes cache
+            failedPublishes.removeAll(messageID, clientId, routings.successedRoutings);
+        });
     }
 
     static MqttQoS lowerQosToTheSubscriptionDesired(Subscription sub, MqttQoS qos) {
@@ -272,24 +541,26 @@ class PostOffice {
      *
      * @param msg
      *            the message to publish
+     * @return
      */
-    public void internalPublish(MqttPublishMessage msg) {
+    public CompletableFuture<Void> 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);
 
-        publish2Subscribers(payload, topic, qos);
-
-        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 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);
+        });
     }
 
     /**
@@ -308,8 +579,43 @@ class PostOffice {
         interceptor.notifyClientConnectionLost(clientId, userName);
     }
 
-//    void flushInFlight(MQTTConnection mqttConnection) {
-//        Session targetSession = sessionRegistry.retrieve(mqttConnection.getClientId());
-//        targetSession.flushAllQueuedMessages();
-//    }
+    /**
+     * Route the command to the owning SessionEventLoop
+     * */
+    public CompletableFuture<RouteResult> routeCommand(String clientId, 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);
+        final FutureTask<String> task = new FutureTask<>(() -> {
+            cmd.execute();
+            cmd.complete();
+            return cmd.getSessionId();
+        });
+        if (Thread.currentThread() == sessionExecutors[targetQueueId]) {
+            SessionEventLoop.executeTask(task);
+            return cmd.completableFuture().thenApply(RouteResult::success);
+        }
+        if (this.sessionQueues[targetQueueId].offer(task)) {
+            return cmd.completableFuture().thenApply(RouteResult::success);
+        } else {
+            LOG.warn("Session command queue {} is full", targetQueueId);
+            final CompletableFuture<RouteResult> failed = new CompletableFuture<>();
+            failed.complete(RouteResult.failed(cmd.getSessionId()));
+            return failed;
+        }
+    }
+
+    public void terminate() {
+        for (Thread processor:sessionExecutors) {
+            processor.interrupt();
+        }
+    }
+
+    /**
+     * Clean up all the data related to the specified client;
+     * */
+    public void clientDisconnected(String clientID, String userName) {
+        dispatchDisconnection(clientID, userName);
+        this.failedPublishes.cleanupForClient(clientID);
+    }
 }
diff --git a/broker/src/main/java/io/moquette/broker/Server.java b/broker/src/main/java/io/moquette/broker/Server.java
index 81d45e07..1fab4260 100644
--- a/broker/src/main/java/io/moquette/broker/Server.java
+++ b/broker/src/main/java/io/moquette/broker/Server.java
@@ -35,6 +35,7 @@ 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;
 
@@ -178,7 +179,9 @@ public class Server {
         subscriptions.init(subscriptionsRepository);
         final Authorizator authorizator = new Authorizator(authorizatorPolicy);
         sessions = new SessionRegistry(subscriptions, queueRepository, authorizator);
-        dispatcher = new PostOffice(subscriptions, retainedRepository, sessions, interceptor, authorizator);
+        final int sessionQueueSize = config.intProp(BrokerConstants.SESSION_QUEUE_SIZE, 1024);
+        dispatcher = new PostOffice(subscriptions, retainedRepository, sessions, interceptor, authorizator,
+                                    sessionQueueSize);
         final BrokerConfiguration brokerConfig = new BrokerConfiguration(config);
         MQTTConnectionFactory connectionFactory = new MQTTConnectionFactory(brokerConfig, authenticator, sessions,
                                                                             dispatcher);
@@ -298,8 +301,9 @@ public class Server {
      * @param msg      the message to forward. The ByteBuf in the message will be released.
      * @param clientId the id of the sending integration.
      * @throws IllegalStateException if the integration is not yet started
+     * @return
      */
-    public void internalPublish(MqttPublishMessage msg, final String clientId) {
+    public CompletableFuture<Void> 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,
@@ -307,8 +311,8 @@ 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);
-        dispatcher.internalPublish(msg);
-        msg.payload().release();
+        return dispatcher.internalPublish(msg)
+            .thenRun(() -> msg.payload().release());
     }
 
     public void stopServer() {
@@ -327,6 +331,7 @@ public class Server {
         }
 
         interceptor.stop();
+        dispatcher.terminate();
         LOG.info("Moquette integration has been stopped.");
     }
 
diff --git a/broker/src/main/java/io/moquette/broker/Session.java b/broker/src/main/java/io/moquette/broker/Session.java
index 1880ff04..e8a55a9b 100644
--- a/broker/src/main/java/io/moquette/broker/Session.java
+++ b/broker/src/main/java/io/moquette/broker/Session.java
@@ -201,6 +201,9 @@ class Session {
             return;
         }
 
+        if (mqttConnection == null) {
+            return;
+        }
         inflightWindow.put(pubRecPacketId, new SessionRegistry.PubRelMarker());
         inflightTimeouts.add(new InFlightPacket(pubRecPacketId, FLIGHT_BEFORE_RESEND_MS));
         MqttMessage pubRel = MQTTConnection.pubrel(pubRecPacketId);
@@ -250,9 +253,10 @@ class Session {
             return;
         }
 
-        if (canSkipQueue()) {
+        final MQTTConnection localMqttConnectionRef = mqttConnection;
+        if (canSkipQueue(localMqttConnectionRef)) {
             inflightSlots.decrementAndGet();
-            int packetId = mqttConnection.nextPacketId();
+            int packetId = localMqttConnectionRef.nextPacketId();
 
             // Adding to a map, retain.
             payload.retain();
@@ -266,10 +270,10 @@ class Session {
 
             MqttPublishMessage publishMsg = MQTTConnection.notRetainedPublishWithMessageId(topic.toString(), qos,
                                                                                            payload, packetId);
-            mqttConnection.sendPublish(publishMsg);
+            localMqttConnectionRef.sendPublish(publishMsg);
             LOG.debug("Write direct to the peer, inflight slots: {}", inflightSlots.get());
             if (inflightSlots.get() == 0) {
-                mqttConnection.flush();
+                localMqttConnectionRef.flush();
             }
 
             // TODO drainQueueToConnection();?
@@ -283,9 +287,10 @@ class Session {
     }
 
     private void sendPublishQos2(Topic topic, MqttQoS qos, ByteBuf payload) {
-        if (canSkipQueue()) {
+        final MQTTConnection localMqttConnectionRef = mqttConnection;
+        if (canSkipQueue(localMqttConnectionRef)) {
             inflightSlots.decrementAndGet();
-            int packetId = mqttConnection.nextPacketId();
+            int packetId = localMqttConnectionRef.nextPacketId();
 
             // Retain before adding to map
             payload.retain();
@@ -299,7 +304,7 @@ class Session {
 
             MqttPublishMessage publishMsg = MQTTConnection.notRetainedPublishWithMessageId(topic.toString(), qos,
                                                                                            payload, packetId);
-            mqttConnection.sendPublish(publishMsg);
+            localMqttConnectionRef.sendPublish(publishMsg);
 
             drainQueueToConnection();
         } else {
@@ -310,11 +315,12 @@ class Session {
         }
     }
 
-    private boolean canSkipQueue() {
-        return sessionQueue.isEmpty() &&
+    private boolean canSkipQueue(MQTTConnection localMqttConnectionRef) {
+        return localMqttConnectionRef != null &&
+            sessionQueue.isEmpty() &&
             inflightSlots.get() > 0 &&
             connected() &&
-            mqttConnection.channel.isWritable();
+            localMqttConnectionRef.channel.isWritable();
     }
 
     private boolean inflighHasSlotsAndConnectionIsUp() {
@@ -361,8 +367,8 @@ class Session {
                 final Topic topic = pubMsg.topic;
                 final MqttQoS qos = pubMsg.publishingQos;
                 final ByteBuf payload = pubMsg.payload;
-                final ByteBuf copiedPayload = payload.retainedDuplicate();
-                MqttPublishMessage publishMsg = publishNotRetainedDuplicated(notAckPacketId, topic, qos, copiedPayload);
+                // message fetched from map, but not removed from map. No need to duplicate or release.
+                MqttPublishMessage publishMsg = publishNotRetainedDuplicated(notAckPacketId, topic, qos, payload);
                 inflightTimeouts.add(new InFlightPacket(notAckPacketId.packetId, FLIGHT_BEFORE_RESEND_MS));
                 mqttConnection.sendPublish(publishMsg);
             }
@@ -443,7 +449,7 @@ class Session {
         // In case of evil client with duplicate msgid.
         ReferenceCountUtil.release(old);
 
-        mqttConnection.sendPublishReceived(messageID);
+//        mqttConnection.sendPublishReceived(messageID);
     }
 
     public void receivedPubRelQos2(int messageID) {
diff --git a/broker/src/main/java/io/moquette/broker/SessionCommand.java b/broker/src/main/java/io/moquette/broker/SessionCommand.java
new file mode 100644
index 00000000..a51a5423
--- /dev/null
+++ b/broker/src/main/java/io/moquette/broker/SessionCommand.java
@@ -0,0 +1,33 @@
+package io.moquette.broker;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+
+final class SessionCommand {
+
+    private final String sessionId;
+    private final Callable<String> action;
+    private final CompletableFuture<String> task;
+
+    public  SessionCommand(String sessionId, Callable<String> action) {
+        this.sessionId = sessionId;
+        this.action = action;
+        this.task = new CompletableFuture<>();
+    }
+
+    public String getSessionId() {
+        return this.sessionId;
+    }
+
+    public void execute() throws Exception {
+        action.call();
+    }
+
+    public void complete() {
+        task.complete(sessionId);
+    }
+
+    public CompletableFuture<String> completableFuture() {
+        return task;
+    }
+}
diff --git a/broker/src/main/java/io/moquette/broker/SessionEventLoop.java b/broker/src/main/java/io/moquette/broker/SessionEventLoop.java
new file mode 100644
index 00000000..a25ca9a1
--- /dev/null
+++ b/broker/src/main/java/io/moquette/broker/SessionEventLoop.java
@@ -0,0 +1,45 @@
+package io.moquette.broker;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.FutureTask;
+
+final class SessionEventLoop implements Runnable {
+
+    private static final Logger LOG = LoggerFactory.getLogger(SessionEventLoop.class);
+
+    private final BlockingQueue<FutureTask<String>> sessionQueue;
+
+    public SessionEventLoop(BlockingQueue<FutureTask<String>> sessionQueue) {
+        this.sessionQueue = sessionQueue;
+    }
+
+    @Override
+    public void run() {
+        while (!Thread.interrupted()) {
+            try {
+                // blocking call
+                final FutureTask<String> task = this.sessionQueue.take();
+                executeTask(task);
+            } catch (InterruptedException e) {
+                LOG.info("SessionEventLoop {} interrupted", Thread.currentThread().getName());
+                Thread.currentThread().interrupt();
+            }
+        }
+    }
+
+    public static void executeTask(final FutureTask<String> task) {
+        if (!task.isCancelled()) {
+            try {
+                task.run();
+
+                // we ran it, but we have to grab the exception if raised
+                task.get();
+            } catch (Throwable th) {
+                LOG.info("SessionEventLoop {} reached exception in processing command", Thread.currentThread().getName(), th);
+            }
+        }
+    }
+}
diff --git a/distribution/src/main/resources/moquette.conf b/distribution/src/main/resources/moquette.conf
index dd83b0da..21991d19 100644
--- a/distribution/src/main/resources/moquette.conf
+++ b/distribution/src/main/resources/moquette.conf
@@ -131,6 +131,14 @@ password_file config/password_file.conf
 # netty.epoll true
 # netty.mqtt.message_size 8092
 
+#*********************************************************************
+# Command session queues
+#
+# session_queue_size:
+#         the size of each session command queue used to Session's Event loops
+#*********************************************************************
+# session_queue_size 1024
+
 #*********************************************************************
 # Metrics Configuration
 #
diff --git a/embedding_moquette/pom.xml b/embedding_moquette/pom.xml
index 70523d47..f8eb1c93 100644
--- a/embedding_moquette/pom.xml
+++ b/embedding_moquette/pom.xml
@@ -30,6 +30,14 @@
                     <skip>true</skip>
                 </configuration>
             </plugin>
+<!--            <plugin>-->
+<!--                <groupId>org.apache.maven.plugins</groupId>-->
+<!--                <artifactId>maven-compiler-plugin</artifactId>-->
+<!--                <configuration>-->
+<!--                    <source>1.8</source>-->
+<!--                    <target>1.8</target>-->
+<!--                </configuration>-->
+<!--            </plugin>-->
         </plugins>
     </build>
 


Test patch:
diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java
index 9eea8954..ca73c87d 100644
--- a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java
+++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java
@@ -26,8 +26,11 @@ import io.netty.handler.codec.mqtt.MqttConnectMessage;
 import io.netty.handler.codec.mqtt.MqttMessageBuilders;
 import io.netty.handler.codec.mqtt.MqttVersion;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
+import java.util.concurrent.ExecutionException;
+
 import static io.moquette.broker.NettyChannelAssertions.assertEqualsConnAck;
 import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
 import static java.util.Collections.singleton;
@@ -69,7 +72,7 @@ public class MQTTConnectionConnectTest {
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
         sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll);
         postOffice = new PostOffice(subscriptions, new MemoryRetainedRepository(), sessionRegistry,
-                                    ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll);
+                                    ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, 1024);
 
         sut = createMQTTConnection(CONFIG);
         channel = (EmbeddedChannel) sut.channel;
@@ -90,7 +93,7 @@ public class MQTTConnectionConnectTest {
     }
 
     @Test
-    public void testZeroByteClientIdWithCleanSession() {
+    public void testZeroByteClientIdWithCleanSession() throws InterruptedException, ExecutionException {
         // Connect message with clean session set to true and client id is null.
         MqttConnectMessage msg = MqttMessageBuilders.connect()
             .protocolVersion(MqttVersion.MQTT_3_1_1)
@@ -98,7 +101,7 @@ public class MQTTConnectionConnectTest {
             .cleanSession(true)
             .build();
 
-        sut.processConnect(msg);
+        sut.processConnect(msg).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");
@@ -121,39 +124,39 @@ public class MQTTConnectionConnectTest {
     }
 
     @Test
-    public void testConnect_badClientID() {
+    public void testConnect_badClientID() throws ExecutionException, InterruptedException {
         connMsg.clientId("extremely_long_clientID_greater_than_23").build();
 
         // Exercise
-        sut.processConnect(connMsg.clientId("extremely_long_clientID_greater_than_23").build());
+        sut.processConnect(connMsg.clientId("extremely_long_clientID_greater_than_23").build()).get();
 
         // Verify
         assertEqualsConnAck(CONNECTION_ACCEPTED, channel.readOutbound());
     }
 
     @Test
-    public void testWillIsAccepted() {
+    public void testWillIsAccepted() throws ExecutionException, InterruptedException {
         MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).willFlag(true)
             .willTopic("topic").willMessage("Topic message").build();
 
         // Exercise
-        // m_handler.setMessaging(mockedMessaging);
-        sut.processConnect(msg);
+        sut.processConnect(msg).get();
 
         // Verify
         assertEqualsConnAck(CONNECTION_ACCEPTED, channel.readOutbound());
         assertTrue(channel.isOpen(), "Connection is accepted and therefore should remain open");
     }
 
+    @Disabled("already covered by testWillMessageIsFiredOnClientKeepAliveExpiry and testWillMessageIsPublishedOnClientBadDisconnection")
     @Test
-    public void testWillIsFired() {
+    public void testWillIsFired() throws ExecutionException, InterruptedException {
         final PostOffice postOfficeMock = mock(PostOffice.class);
         sut = createMQTTConnectionWithPostOffice(CONFIG, postOfficeMock);
         channel = (EmbeddedChannel) sut.channel;
 
         MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).willFlag(true)
             .willTopic("topic").willMessage("Topic message").build();
-        sut.processConnect(msg);
+        sut.processConnect(msg).get();
 
         // Exercise
         sut.handleConnectionLost();
@@ -164,11 +167,11 @@ public class MQTTConnectionConnectTest {
     }
 
     @Test
-    public void acceptAnonymousClient() {
+    public void acceptAnonymousClient() throws ExecutionException, InterruptedException {
         MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build();
 
         // Exercise
-        sut.processConnect(msg);
+        sut.processConnect(msg).get();
 
         // Verify
         assertEqualsConnAck(CONNECTION_ACCEPTED, channel.readOutbound());
@@ -176,12 +179,12 @@ public class MQTTConnectionConnectTest {
     }
 
     @Test
-    public void validAuthentication() {
+    public void validAuthentication() throws ExecutionException, InterruptedException {
         MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
             .username(TEST_USER).password(TEST_PWD).build();
 
         // Exercise
-        sut.processConnect(msg);
+        sut.processConnect(msg).get();
 
         // Verify
         assertEqualsConnAck(CONNECTION_ACCEPTED, channel.readOutbound());
@@ -268,13 +271,13 @@ public class MQTTConnectionConnectTest {
     }
 
     @Test
-    public void testBindWithSameClientIDBadCredentialsDoesntDropExistingClient() {
+    public void testBindWithSameClientIDBadCredentialsDoesntDropExistingClient() throws ExecutionException, InterruptedException {
         // Connect a client1
         MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
             .username(TEST_USER)
             .password(TEST_PWD)
             .build();
-        sut.processConnect(msg);
+        sut.processConnect(msg).get();
         assertEqualsConnAck(CONNECTION_ACCEPTED, channel.readOutbound());
 
         // create another connect same clientID but with bad credentials
@@ -301,17 +304,17 @@ public class MQTTConnectionConnectTest {
     }
 
     @Test
-    public void testForceClientDisconnection_issue116() {
+    public void testForceClientDisconnection_issue116() throws ExecutionException, InterruptedException {
         MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
             .username(TEST_USER)
             .password(TEST_PWD)
             .build();
-        sut.processConnect(msg);
+        sut.processConnect(msg).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);
+        anotherConnection.processConnect(msg).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 f6721c65..9fd9b530 100644
--- a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java
+++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java
@@ -31,6 +31,8 @@ import io.netty.handler.codec.mqtt.MqttVersion;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.util.concurrent.ExecutionException;
+
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Collections.singleton;
 import static java.util.Collections.singletonMap;
@@ -59,6 +61,7 @@ public class MQTTConnectionPublishTest {
 
     private void createMQTTConnection(BrokerConfiguration config) {
         channel = new EmbeddedChannel();
+        NettyUtils.clientID(channel, "test_client");
         sut = createMQTTConnection(config, channel);
     }
 
@@ -75,12 +78,12 @@ public class MQTTConnectionPublishTest {
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
         sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll);
         final PostOffice postOffice = new PostOffice(subscriptions,
-            new MemoryRetainedRepository(), sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll);
+            new MemoryRetainedRepository(), sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, 1024);
         return new MQTTConnection(channel, config, mockAuthenticator, sessionRegistry, postOffice);
     }
 
     @Test
-    public void dropConnectionOnPublishWithInvalidTopicFormat() {
+    public void dropConnectionOnPublishWithInvalidTopicFormat() throws ExecutionException, InterruptedException {
         // Connect message with clean session set to true and client id is null.
         final ByteBuf payload = Unpooled.copiedBuffer("Hello MQTT world!".getBytes(UTF_8));
         MqttPublishMessage publish = MqttMessageBuilders.publish()
@@ -89,7 +92,7 @@ public class MQTTConnectionPublishTest {
             .qos(MqttQoS.AT_MOST_ONCE)
             .payload(payload).build();
 
-        sut.processPublish(publish);
+        sut.processPublish(publish).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 58c205e7..a5b05ffd 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java
@@ -29,6 +29,9 @@ import org.junit.jupiter.api.BeforeEach;
 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.*;
@@ -57,7 +60,7 @@ public class PostOfficeInternalPublishTest {
     private MemoryQueueRepository queueRepository;
 
     @BeforeEach
-    public void setUp() {
+    public void setUp() throws ExecutionException, InterruptedException {
         sessionRegistry = initPostOfficeAndSubsystems();
 
         mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD));
@@ -65,7 +68,7 @@ public class PostOfficeInternalPublishTest {
 
         connectMessage = ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID);
 
-        connection.processConnect(connectMessage);
+        connection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
     }
 
@@ -89,7 +92,7 @@ public class PostOfficeInternalPublishTest {
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
         SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll);
         sut = new PostOffice(subscriptions, retainedRepository, sessionRegistry,
-                             ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll);
+                             ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, 1024);
         return sessionRegistry;
     }
 
@@ -107,7 +110,11 @@ public class PostOfficeInternalPublishTest {
             .retained(retained)
             .qos(qos)
             .payload(Unpooled.copiedBuffer(PAYLOAD.getBytes(UTF_8))).build();
-        sut.internalPublish(publish);
+        try {
+            sut.internalPublish(publish).get(5, TimeUnit.SECONDS);
+        } catch (InterruptedException | ExecutionException | TimeoutException e) {
+            throw new RuntimeException(e);
+        }
     }
 
     @Test
@@ -254,7 +261,7 @@ public class PostOfficeInternalPublishTest {
     }
 
     @Test
-    public void testClientSubscribeWithoutCleanSession() {
+    public void testClientSubscribeWithoutCleanSession() throws ExecutionException, InterruptedException {
         subscribe(AT_MOST_ONCE, "foo", connection);
         connection.processDisconnect(null);
         assertEquals(1, subscriptions.size());
@@ -265,7 +272,7 @@ public class PostOfficeInternalPublishTest {
             .clientId(FAKE_CLIENT_ID)
             .cleanSession(false)
             .build();
-        anotherConn.processConnect(connectMessage);
+        anotherConn.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted((EmbeddedChannel) anotherConn.channel);
 
         assertEquals(1, subscriptions.size());
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java
index 33170cd6..caa02ff8 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java
@@ -33,6 +33,9 @@ import org.junit.jupiter.api.Test;
 
 import java.nio.charset.Charset;
 import java.util.*;
+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.AT_LEAST_ONCE;
@@ -96,13 +99,13 @@ public class PostOfficePublishTest {
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
         SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll);
         sut = new PostOffice(subscriptions, retainedRepository, sessionRegistry,
-                             ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll);
+                             ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, 1024);
         return sessionRegistry;
     }
 
     @Test
-    public void testPublishQoS0ToItself() {
-        connection.processConnect(connectMessage);
+    public void testPublishQoS0ToItself() throws ExecutionException, InterruptedException, TimeoutException {
+        connection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
 
         // subscribe
@@ -115,14 +118,14 @@ public class PostOfficePublishTest {
                 .payload(payload.retainedDuplicate())
                 .qos(MqttQoS.AT_MOST_ONCE)
                 .retained(false)
-                .topicName(NEWS_TOPIC).build());
+                .topicName(NEWS_TOPIC).build()).get(5, TimeUnit.SECONDS);
 
         // Verify
         ConnectionTestUtils.verifyReceivePublish(channel, NEWS_TOPIC, "Hello world!");
     }
 
     @Test
-    public void testForceClientDisconnection_issue116() {
+    public void testForceClientDisconnection_issue116() throws ExecutionException, InterruptedException, TimeoutException {
         final MQTTConnection clientXA = connectAs("subscriber");
         subscribe(clientXA, NEWS_TOPIC, AT_MOST_ONCE);
 
@@ -132,6 +135,7 @@ public class PostOfficePublishTest {
             .payload(anyPayload)
             .qos(MqttQoS.EXACTLY_ONCE)
             .retained(false)
+            .messageId(1)
             .topicName(NEWS_TOPIC).build(), "username");
 
         final MQTTConnection clientYA = connectAs("subscriber");
@@ -143,7 +147,8 @@ public class PostOfficePublishTest {
             .payload(anyPayload2)
             .qos(MqttQoS.EXACTLY_ONCE)
             .retained(true)
-            .topicName(NEWS_TOPIC).build(), "username");
+            .messageId(2)
+            .topicName(NEWS_TOPIC).build(), "username").get(5, TimeUnit.SECONDS);
 
         // Verify
         assertFalse(clientXA.channel.isOpen(), "First 'subscriber' channel MUST be closed by the broker");
@@ -153,7 +158,11 @@ public class PostOfficePublishTest {
     private MQTTConnection connectAs(String clientId) {
         EmbeddedChannel channel = new EmbeddedChannel();
         MQTTConnection connection = createMQTTConnection(CONFIG, channel);
-        connection.processConnect(ConnectionTestUtils.buildConnect(clientId));
+        try {
+            connection.processConnect(ConnectionTestUtils.buildConnect(clientId)).get();
+        } catch (InterruptedException | ExecutionException e) {
+            fail(e);
+        }
         ConnectionTestUtils.assertConnectAccepted(channel);
         return connection;
     }
@@ -190,17 +199,17 @@ public class PostOfficePublishTest {
     }
 
     @Test
-    public void testPublishToMultipleSubscribers() {
+    public void testPublishToMultipleSubscribers() throws ExecutionException, InterruptedException, TimeoutException {
         final Set<String> clientIds = new HashSet<>(Arrays.asList(FAKE_CLIENT_ID, FAKE_CLIENT_ID2));
         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));
+        connection1.processConnect(ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID)).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));
+        connection2.processConnect(ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID2)).get();
         ConnectionTestUtils.assertConnectAccepted(channel2);
 
         // subscribe
@@ -216,7 +225,7 @@ public class PostOfficePublishTest {
                 .payload(payload.retainedDuplicate())
                 .qos(MqttQoS.AT_MOST_ONCE)
                 .retained(false)
-                .topicName(NEWS_TOPIC).build());
+                .topicName(NEWS_TOPIC).build()).get(5, TimeUnit.SECONDS);
 
         // Verify
         ConnectionTestUtils.verifyReceivePublish(channel1, NEWS_TOPIC, "Hello world!");
@@ -224,8 +233,8 @@ public class PostOfficePublishTest {
     }
 
     @Test
-    public void testPublishWithEmptyPayloadClearRetainedStore() {
-        connection.processConnect(connectMessage);
+    public void testPublishWithEmptyPayloadClearRetainedStore() throws ExecutionException, InterruptedException {
+        connection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
 
         final ByteBuf payload1 = ByteBufUtil.writeAscii(UnpooledByteBufAllocator.DEFAULT, "Hello world!");
@@ -252,8 +261,8 @@ public class PostOfficePublishTest {
     }
 
     @Test
-    public void testPublishWithQoS1() {
-        connection.processConnect(connectMessage);
+    public void testPublishWithQoS1() throws ExecutionException, InterruptedException, TimeoutException {
+        connection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         subscribe(connection, NEWS_TOPIC, AT_LEAST_ONCE);
 
@@ -264,15 +273,15 @@ public class PostOfficePublishTest {
                 .payload(Unpooled.copiedBuffer("Any payload", Charset.defaultCharset()))
                 .qos(MqttQoS.AT_LEAST_ONCE)
                 .retained(true)
-                .topicName(NEWS_TOPIC).build());
+                .topicName(NEWS_TOPIC).build()).get(5, TimeUnit.SECONDS);
 
         // Verify
         ConnectionTestUtils.verifyPublishIsReceived(channel, AT_LEAST_ONCE, "Any payload");
     }
 
     @Test
-    public void testPublishWithQoS2() {
-        connection.processConnect(connectMessage);
+    public void testPublishWithQoS2() throws ExecutionException, InterruptedException, TimeoutException {
+        connection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         subscribe(connection, NEWS_TOPIC, EXACTLY_ONCE);
 
@@ -282,7 +291,8 @@ public class PostOfficePublishTest {
                 .payload(anyPayload)
                 .qos(MqttQoS.EXACTLY_ONCE)
                 .retained(true)
-                .topicName(NEWS_TOPIC).build(), "username");
+                .messageId(2)
+                .topicName(NEWS_TOPIC).build(), "username").get(5000, TimeUnit.SECONDS);
 
         // Verify
         ConnectionTestUtils.verifyPublishIsReceived(channel, EXACTLY_ONCE, "Any payload");
@@ -290,8 +300,8 @@ public class PostOfficePublishTest {
 
     // aka testPublishWithQoS1_notCleanSession
     @Test
-    public void forwardQoS1PublishesWhenNotCleanSessionReconnects() {
-        connection.processConnect(ConnectionTestUtils.buildConnectNotClean(FAKE_CLIENT_ID));
+    public void forwardQoS1PublishesWhenNotCleanSessionReconnects() throws ExecutionException, InterruptedException {
+        connection.processConnect(ConnectionTestUtils.buildConnectNotClean(FAKE_CLIENT_ID)).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         subscribe(connection, NEWS_TOPIC, AT_LEAST_ONCE);
         connection.processDisconnect(null);
@@ -299,7 +309,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));
+        pubConn.processConnect(ConnectionTestUtils.buildConnect(PUBLISHER_ID)).get();
         ConnectionTestUtils.assertConnectAccepted(pubChannel);
 
         final ByteBuf anyPayload = Unpooled.copiedBuffer("Any payload", Charset.defaultCharset());
@@ -312,7 +322,7 @@ public class PostOfficePublishTest {
         // simulate a reconnection from the other client
         connection = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID);
         connectMessage = ConnectionTestUtils.buildConnectNotClean(FAKE_CLIENT_ID);
-        connection.processConnect(connectMessage);
+        connection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
 
         // Verify
@@ -320,9 +330,9 @@ public class PostOfficePublishTest {
     }
 
     @Test
-    public void checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession() {
+    public void checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession() throws ExecutionException, InterruptedException, TimeoutException {
         // first connect - subscribe -disconnect
-        connection.processConnect(ConnectionTestUtils.buildConnectNotClean(FAKE_CLIENT_ID));
+        connection.processConnect(ConnectionTestUtils.buildConnectNotClean(FAKE_CLIENT_ID)).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         subscribe(connection, NEWS_TOPIC, AT_LEAST_ONCE);
         connection.processDisconnect(null);
@@ -330,14 +340,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));
+        secondConn.processConnect(ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID)).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));
+        pubConn.processConnect(ConnectionTestUtils.buildConnect(PUBLISHER_ID)).get();
         ConnectionTestUtils.assertConnectAccepted(pubChannel);
 
         final ByteBuf anyPayload = Unpooled.copiedBuffer("Any payload", Charset.defaultCharset());
@@ -346,16 +356,16 @@ public class PostOfficePublishTest {
                 .payload(anyPayload.retainedDuplicate())
                 .qos(MqttQoS.AT_LEAST_ONCE)
                 .retained(true)
-                .topicName(NEWS_TOPIC).build());
+                .topicName(NEWS_TOPIC).build()).get(5, TimeUnit.SECONDS);
 
         // Verify that after a reconnection the client receive the message
         ConnectionTestUtils.verifyPublishIsReceived(secondChannel, AT_LEAST_ONCE, "Any payload");
     }
 
     @Test
-    public void noPublishToInactiveSession() {
+    public void noPublishToInactiveSession() throws ExecutionException, InterruptedException {
         // create an inactive session for Subscriber
-        connection.processConnect(ConnectionTestUtils.buildConnectNotClean(SUBSCRIBER_ID));
+        connection.processConnect(ConnectionTestUtils.buildConnectNotClean(SUBSCRIBER_ID)).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         subscribe(connection, NEWS_TOPIC, AT_LEAST_ONCE);
         connection.processDisconnect(null);
@@ -363,7 +373,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));
+        pubConn.processConnect(ConnectionTestUtils.buildConnect(PUBLISHER_ID)).get();
         ConnectionTestUtils.assertConnectAccepted(pubChannel);
 
         final ByteBuf anyPayload = Unpooled.copiedBuffer("Any payload", Charset.defaultCharset());
@@ -383,8 +393,8 @@ public class PostOfficePublishTest {
     }
 
     @Test
-    public void cleanRetainedMessageStoreWhenPublishWithRetainedQos0IsReceived() {
-        connection.processConnect(connectMessage);
+    public void cleanRetainedMessageStoreWhenPublishWithRetainedQos0IsReceived() throws ExecutionException, InterruptedException {
+        connection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
 
         // publish a QoS1 retained message
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java
index 4bc3abb1..30a638b8 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java
@@ -34,6 +34,9 @@ import org.junit.jupiter.api.Test;
 import java.nio.charset.Charset;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import static io.moquette.broker.PostOfficePublishTest.ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID;
 import static io.moquette.broker.PostOfficePublishTest.SUBSCRIBER_ID;
@@ -94,7 +97,7 @@ public class PostOfficeSubscribeTest {
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
         sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll);
         sut = new PostOffice(subscriptions, new MemoryRetainedRepository(), sessionRegistry,
-                             ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll);
+                             ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, 1024);
     }
 
     private MQTTConnection createMQTTConnection(BrokerConfiguration config, Channel channel) {
@@ -111,8 +114,8 @@ public class PostOfficeSubscribeTest {
     }
 
     @Test
-    public void testSubscribe() {
-        connection.processConnect(connectMessage);
+    public void testSubscribe() throws ExecutionException, InterruptedException {
+        connection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
 
         // Exercise & verify
@@ -159,7 +162,7 @@ public class PostOfficeSubscribeTest {
     }
 
     @Test
-    public void testSubscribedToNotAuthorizedTopic() {
+    public void testSubscribedToNotAuthorizedTopic() throws ExecutionException, InterruptedException {
         NettyUtils.userName(channel, FAKE_USER_NAME);
 
         IAuthorizatorPolicy prohibitReadOnNewsTopic = mock(IAuthorizatorPolicy.class);
@@ -167,9 +170,9 @@ public class PostOfficeSubscribeTest {
             .thenReturn(false);
 
         sut = new PostOffice(subscriptions, new MemoryRetainedRepository(), sessionRegistry,
-                             ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, new Authorizator(prohibitReadOnNewsTopic));
+                             ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, new Authorizator(prohibitReadOnNewsTopic), 1024);
 
-        connection.processConnect(connectMessage);
+        connection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
 
         //Exercise
@@ -191,8 +194,8 @@ public class PostOfficeSubscribeTest {
     }
 
     @Test
-    public void testDoubleSubscribe() {
-        connection.processConnect(connectMessage);
+    public void testDoubleSubscribe() throws ExecutionException, InterruptedException {
+        connection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         assertEquals(0, subscriptions.size(), "After CONNECT subscription MUST be empty");
         subscribe(channel, NEWS_TOPIC, AT_MOST_ONCE);
@@ -203,8 +206,8 @@ public class PostOfficeSubscribeTest {
     }
 
     @Test
-    public void testSubscribeWithBadFormattedTopic() {
-        connection.processConnect(connectMessage);
+    public void testSubscribeWithBadFormattedTopic() throws ExecutionException, InterruptedException {
+        connection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         assertEquals(0, subscriptions.size(), "After CONNECT subscription MUST be empty");
 
@@ -221,8 +224,8 @@ public class PostOfficeSubscribeTest {
     }
 
     @Test
-    public void testCleanSession_maintainClientSubscriptions() {
-        connection.processConnect(connectMessage);
+    public void testCleanSession_maintainClientSubscriptions() throws ExecutionException, InterruptedException, TimeoutException {
+        connection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         assertEquals(0, subscriptions.size(), "After CONNECT subscription MUST be empty");
 
@@ -235,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));
+        anotherConn.processConnect(ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID)).get();
         ConnectionTestUtils.assertConnectAccepted(anotherChannel);
         assertEquals(1, subscriptions.size(), "After a reconnect, subscription MUST be still present");
 
@@ -245,7 +248,7 @@ public class PostOfficeSubscribeTest {
                 .payload(payload.retainedDuplicate())
                 .qos(MqttQoS.AT_MOST_ONCE)
                 .retained(false)
-                .topicName(NEWS_TOPIC).build());
+                .topicName(NEWS_TOPIC).build()).get(5, TimeUnit.SECONDS);
 
         ConnectionTestUtils.verifyPublishIsReceived(anotherChannel, AT_MOST_ONCE, "Hello world!");
     }
@@ -256,8 +259,8 @@ public class PostOfficeSubscribeTest {
      * previous subscription
      */
     @Test
-    public void testCleanSession_correctlyClientSubscriptions() {
-        connection.processConnect(connectMessage);
+    public void testCleanSession_correctlyClientSubscriptions() throws ExecutionException, InterruptedException, TimeoutException {
+        connection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         assertEquals(0, subscriptions.size(), "After CONNECT subscription MUST be empty");
 
@@ -267,10 +270,10 @@ public class PostOfficeSubscribeTest {
             .addSubscription(AT_MOST_ONCE, NEWS_TOPIC)
             .messageId(1)
             .build();
-        connection.processSubscribe(subscribeMsg);
+        connection.processSubscribe(subscribeMsg).get(5, TimeUnit.SECONDS);
         assertEquals(1, subscriptions.size(), "Subscribe MUST contain one subscription");
 
-        connection.processDisconnect(null);
+        connection.processDisconnect(null).get(5, TimeUnit.SECONDS);
         assertEquals(1, subscriptions.size(), "Disconnection MUSTN'T clear subscriptions");
 
         connectMessage = MqttMessageBuilders.connect()
@@ -279,7 +282,7 @@ public class PostOfficeSubscribeTest {
             .build();
         channel = new EmbeddedChannel();
         connection = createMQTTConnection(CONFIG, channel);
-        connection.processConnect(connectMessage);
+        connection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         assertEquals(0, subscriptions.size(), "After CONNECT with clean, subscription MUST be empty");
 
@@ -297,9 +300,9 @@ public class PostOfficeSubscribeTest {
     }
 
     @Test
-    public void testReceiveRetainedPublishRespectingSubscriptionQoSAndNotPublisher() {
+    public void testReceiveRetainedPublishRespectingSubscriptionQoSAndNotPublisher() throws ExecutionException, InterruptedException {
         // publisher publish a retained message on topic /news
-        connection.processConnect(connectMessage);
+        connection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(channel);
         final ByteBuf payload = Unpooled.copiedBuffer("Hello world!", Charset.defaultCharset());
         final MqttPublishMessage retainedPubQoS1Msg = MqttMessageBuilders.publish()
@@ -313,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));
+        subConn.processConnect(ConnectionTestUtils.buildConnect(SUBSCRIBER_ID)).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 e988bace..177117fc 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java
@@ -27,12 +27,16 @@ import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.codec.mqtt.*;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import java.nio.charset.Charset;
 import java.util.Collections;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import static io.moquette.broker.PostOfficePublishTest.PUBLISHER_ID;
 import static io.netty.handler.codec.mqtt.MqttQoS.*;
@@ -85,7 +89,7 @@ public class PostOfficeUnsubscribeTest {
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
         sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll);
         sut = new PostOffice(subscriptions, new MemoryRetainedRepository(), sessionRegistry,
-                             ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll);
+                             ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, 1024);
     }
 
     private MQTTConnection createMQTTConnection(BrokerConfiguration config, Channel channel) {
@@ -98,7 +102,11 @@ public class PostOfficeUnsubscribeTest {
     }
 
     protected void connect(MQTTConnection connection, MqttConnectMessage connectMessage) {
-        connection.processConnect(connectMessage);
+        try {
+            connection.processConnect(connectMessage).get();
+        } catch (InterruptedException | ExecutionException e) {
+            Assertions.fail(e);
+        }
         ConnectionTestUtils.assertConnectAccepted((EmbeddedChannel) connection.channel);
     }
 
@@ -137,7 +145,7 @@ public class PostOfficeUnsubscribeTest {
     }
 
     @Test
-    public void testDontNotifyClientSubscribedToTopicAfterDisconnectedAndReconnectOnSameChannel() {
+    public void testDontNotifyClientSubscribedToTopicAfterDisconnectedAndReconnectOnSameChannel() throws ExecutionException, InterruptedException, TimeoutException {
         connect(this.connection, FAKE_CLIENT_ID);
         subscribe(connection, NEWS_TOPIC, AT_MOST_ONCE);
 
@@ -148,7 +156,7 @@ public class PostOfficeUnsubscribeTest {
                 .payload(payload.retainedDuplicate())
                 .qos(MqttQoS.AT_MOST_ONCE)
                 .retained(false)
-                .topicName(NEWS_TOPIC).build());
+                .topicName(NEWS_TOPIC).build()).get(5, TimeUnit.SECONDS);
 
         ConnectionTestUtils.verifyPublishIsReceived(channel, AT_MOST_ONCE, "Hello world!");
 
@@ -176,7 +184,7 @@ public class PostOfficeUnsubscribeTest {
     }
 
     @Test
-    public void testDontNotifyClientSubscribedToTopicAfterDisconnectedAndReconnectOnNewChannel() {
+    public void testDontNotifyClientSubscribedToTopicAfterDisconnectedAndReconnectOnNewChannel() throws ExecutionException, InterruptedException, TimeoutException {
         connect(this.connection, FAKE_CLIENT_ID);
         subscribe(connection, NEWS_TOPIC, AT_MOST_ONCE);
         // publish on /news
@@ -186,7 +194,7 @@ public class PostOfficeUnsubscribeTest {
                 .payload(payload.retainedDuplicate())
                 .qos(MqttQoS.AT_MOST_ONCE)
                 .retained(false)
-                .topicName(NEWS_TOPIC).build());
+                .topicName(NEWS_TOPIC).build()).get(5, TimeUnit.SECONDS);
 
         ConnectionTestUtils.verifyPublishIsReceived(channel, AT_MOST_ONCE, "Hello world!");
 
@@ -196,7 +204,7 @@ public class PostOfficeUnsubscribeTest {
         // connect on another channel
         EmbeddedChannel anotherChannel = new EmbeddedChannel();
         MQTTConnection anotherConnection = createMQTTConnection(CONFIG, anotherChannel);
-        anotherConnection.processConnect(connectMessage);
+        anotherConnection.processConnect(connectMessage).get();
         ConnectionTestUtils.assertConnectAccepted(anotherChannel);
 
         // publish on /news
@@ -212,7 +220,7 @@ public class PostOfficeUnsubscribeTest {
     }
 
     @Test
-    public void avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1() {
+    public void avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1() throws ExecutionException, InterruptedException {
         final MqttConnectMessage notCleanConnect = ConnectionTestUtils.buildConnectNotClean(FAKE_CLIENT_ID);
         connect(connection, notCleanConnect);
         subscribe(connection, NEWS_TOPIC, AT_LEAST_ONCE);
@@ -225,7 +233,7 @@ public class PostOfficeUnsubscribeTest {
         // reconnect FAKE_CLIENT on another channel
         EmbeddedChannel anotherChannel2 = new EmbeddedChannel();
         MQTTConnection anotherConnection2 = createMQTTConnection(CONFIG, anotherChannel2);
-        anotherConnection2.processConnect(notCleanConnect);
+        anotherConnection2.processConnect(notCleanConnect).get();
         ConnectionTestUtils.assertConnectAccepted(anotherChannel2);
 
         ConnectionTestUtils.verifyPublishIsReceived(anotherChannel2, MqttQoS.AT_LEAST_ONCE, firstPayload);
@@ -237,7 +245,7 @@ public class PostOfficeUnsubscribeTest {
 
         EmbeddedChannel anotherChannel3 = new EmbeddedChannel();
         MQTTConnection anotherConnection3 = createMQTTConnection(CONFIG, anotherChannel3);
-        anotherConnection3.processConnect(notCleanConnect);
+        anotherConnection3.processConnect(notCleanConnect).get();
         ConnectionTestUtils.assertConnectAccepted(anotherChannel3);
 
         ConnectionTestUtils.verifyPublishIsReceived(anotherChannel3, MqttQoS.AT_LEAST_ONCE, secondPayload);
@@ -262,7 +270,11 @@ public class PostOfficeUnsubscribeTest {
     private MQTTConnection connectNotCleanAs(String clientId) {
         EmbeddedChannel channel = new EmbeddedChannel();
         MQTTConnection connection = createMQTTConnection(CONFIG, channel);
-        connection.processConnect(ConnectionTestUtils.buildConnectNotClean(clientId));
+        try {
+            connection.processConnect(ConnectionTestUtils.buildConnectNotClean(clientId)).get();
+        } catch (InterruptedException | ExecutionException e) {
+            Assertions.fail(e);
+        }
         ConnectionTestUtils.assertConnectAccepted(channel);
         return connection;
     }
@@ -270,13 +282,17 @@ public class PostOfficeUnsubscribeTest {
     private MQTTConnection connectAs(String clientId) {
         EmbeddedChannel channel = new EmbeddedChannel();
         MQTTConnection connection = createMQTTConnection(CONFIG, channel);
-        connection.processConnect(ConnectionTestUtils.buildConnect(clientId));
+        try {
+            connection.processConnect(ConnectionTestUtils.buildConnect(clientId)).get();
+        } catch (InterruptedException | ExecutionException e) {
+            Assertions.fail(e);
+        }
         ConnectionTestUtils.assertConnectAccepted(channel);
         return connection;
     }
 
     @Test
-    public void testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142() {
+    public void testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142() throws ExecutionException, InterruptedException, TimeoutException {
         connect(connection, FAKE_CLIENT_ID);
         subscribe(connection, NEWS_TOPIC, AT_MOST_ONCE);
         // publish on /news
@@ -286,7 +302,7 @@ public class PostOfficeUnsubscribeTest {
                 .payload(payload.retainedDuplicate())
                 .qos(MqttQoS.AT_MOST_ONCE)
                 .retained(false)
-                .topicName(NEWS_TOPIC).build());
+                .topicName(NEWS_TOPIC).build()).get(5, TimeUnit.SECONDS);
 
         ConnectionTestUtils.verifyPublishIsReceived((EmbeddedChannel) connection.channel, AT_MOST_ONCE, "Hello world!");
 
@@ -295,7 +311,7 @@ public class PostOfficeUnsubscribeTest {
         final MqttConnectMessage notCleanConnect = ConnectionTestUtils.buildConnect(FAKE_CLIENT_ID);
         EmbeddedChannel subscriberChannel = new EmbeddedChannel();
         MQTTConnection subscriberConnection = createMQTTConnection(CONFIG, subscriberChannel);
-        subscriberConnection.processConnect(notCleanConnect);
+        subscriberConnection.processConnect(notCleanConnect).get();
         ConnectionTestUtils.assertConnectAccepted(subscriberChannel);
 
         subscribe(subscriberConnection, NEWS_TOPIC, AT_MOST_ONCE);
@@ -306,17 +322,17 @@ public class PostOfficeUnsubscribeTest {
                 .payload(payload2.retainedDuplicate())
                 .qos(MqttQoS.AT_MOST_ONCE)
                 .retained(false)
-                .topicName(NEWS_TOPIC).build());
+                .topicName(NEWS_TOPIC).build()).get(5, TimeUnit.SECONDS);
 
         ConnectionTestUtils.verifyPublishIsReceived(subscriberChannel, AT_MOST_ONCE, "Hello world2!");
 
-        subscriberConnection.processDisconnect(null);
+        subscriberConnection.processDisconnect(null).get();
 
         assertFalse(subscriberChannel.isOpen(), "after a disconnect the client should be disconnected");
     }
 
     @Test
-    public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() {
+    public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() throws ExecutionException, InterruptedException {
         final MQTTConnection publisher = connectAs("Publisher");
 
         connect(this.connection, FAKE_CLIENT_ID);
@@ -325,7 +341,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);
+        connection.processDisconnect(null).get();
 
         publishQos1(publisher, NEWS_TOPIC, "Hello world MQTT!!-2", 100);
         publishQos1(publisher, NEWS_TOPIC, "Hello world MQTT!!-3", 101);
@@ -338,12 +354,16 @@ public class PostOfficeUnsubscribeTest {
 
     private void publishQos1(MQTTConnection publisher, String topic, String payload, int messageID) {
         final ByteBuf bytePayload = Unpooled.copiedBuffer(payload, Charset.defaultCharset());
-        sut.receivedPublishQos1(publisher, new Topic(topic), TEST_USER, messageID,
-            MqttMessageBuilders.publish()
-                .payload(bytePayload)
-                .qos(MqttQoS.AT_LEAST_ONCE)
-                .retained(false)
-                .topicName(NEWS_TOPIC).build());
+        try {
+            sut.receivedPublishQos1(publisher, new Topic(topic), TEST_USER, messageID,
+                MqttMessageBuilders.publish()
+                    .payload(bytePayload)
+                    .qos(MqttQoS.AT_LEAST_ONCE)
+                    .retained(false)
+                    .topicName(NEWS_TOPIC).build()).get(5, TimeUnit.SECONDS);
+        } catch (InterruptedException | ExecutionException | TimeoutException e) {
+            throw new RuntimeException(e);
+        }
     }
 
     private void publishQos2(MQTTConnection connection, String topic, String payload) {
diff --git a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
index 7793ddf1..0f6eb5ba 100644
--- a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
+++ b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
@@ -38,6 +38,7 @@ import org.junit.jupiter.api.Test;
 
 import java.io.File;
 import java.nio.charset.StandardCharsets;
+import java.util.concurrent.ExecutionException;
 
 import static io.moquette.broker.NettyChannelAssertions.assertEqualsConnAck;
 import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED;
@@ -86,7 +87,7 @@ public class SessionRegistryTest {
         final Authorizator permitAll = new Authorizator(authorizatorPolicy);
         sut = new SessionRegistry(subscriptions, queueRepository, permitAll);
         final PostOffice postOffice = new PostOffice(subscriptions,
-            new MemoryRetainedRepository(), sut, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll);
+            new MemoryRetainedRepository(), sut, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll, 1024);
         return new MQTTConnection(channel, config, mockAuthenticator, sut, postOffice);
     }
 
@@ -116,12 +117,12 @@ public class SessionRegistryTest {
     }
 
     @Test
-    public void connectWithCleanSessionUpdateClientSession() {
+    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);
+        connection.processConnect(msg).get();
         assertEqualsConnAck(CONNECTION_ACCEPTED, channel.readOutbound());
-        connection.processDisconnect(null);
+        connection.processDisconnect(null).get();
         assertFalse(channel.isOpen());
 
         // second connect with clean session false
@@ -133,7 +134,7 @@ public class SessionRegistryTest {
             .protocolVersion(MqttVersion.MQTT_3_1)
             .build();
 
-        anotherConnection.processConnect(secondConnMsg);
+        anotherConnection.processConnect(secondConnMsg).get();
         assertEqualsConnAck(CONNECTION_ACCEPTED, anotherChannel.readOutbound());
 
         // Verify client session is clean false
diff --git a/broker/src/test/java/io/moquette/integration/IntegrationUtils.java b/broker/src/test/java/io/moquette/integration/IntegrationUtils.java
index ca062d64..7f4cff61 100644
--- a/broker/src/test/java/io/moquette/integration/IntegrationUtils.java
+++ b/broker/src/test/java/io/moquette/integration/IntegrationUtils.java
@@ -16,10 +16,16 @@
 
 package io.moquette.integration;
 
+import org.eclipse.paho.client.mqttv3.IMqttActionListener;
+import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
+import org.eclipse.paho.client.mqttv3.IMqttClient;
+import org.eclipse.paho.client.mqttv3.MqttException;
+
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Path;
 import java.util.Properties;
+
 import static io.moquette.BrokerConstants.DEFAULT_MOQUETTE_STORE_H2_DB_FILENAME;
 import static io.moquette.BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME;
 import static io.moquette.BrokerConstants.PORT_PROPERTY_NAME;
@@ -57,4 +63,22 @@ public final class IntegrationUtils {
 
     private IntegrationUtils() {
     }
+
+    static void disconnectClient(IMqttClient client) throws MqttException {
+        if (client != null && client.isConnected()) {
+            client.disconnect();
+        }
+    }
+
+    static void disconnectClient(IMqttAsyncClient client, IMqttActionListener callback) throws MqttException {
+        if (client != null && client.isConnected()) {
+            client.disconnect(null, callback);
+        }
+    }
+
+    static void disconnectClient(IMqttAsyncClient client) throws MqttException {
+        if (client != null && client.isConnected()) {
+            client.disconnect();
+        }
+    }
 }
diff --git a/broker/src/test/java/io/moquette/integration/PublishToManySubscribersUseCaseTest.java b/broker/src/test/java/io/moquette/integration/PublishToManySubscribersUseCaseTest.java
new file mode 100644
index 00000000..dc0ae70a
--- /dev/null
+++ b/broker/src/test/java/io/moquette/integration/PublishToManySubscribersUseCaseTest.java
@@ -0,0 +1,188 @@
+package io.moquette.integration;
+
+import io.moquette.BrokerConstants;
+import io.moquette.broker.Server;
+import io.moquette.broker.config.IConfig;
+import io.moquette.broker.config.MemoryConfig;
+import org.awaitility.Awaitility;
+import org.awaitility.Durations;
+import org.eclipse.paho.client.mqttv3.IMqttActionListener;
+import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
+import org.eclipse.paho.client.mqttv3.IMqttToken;
+import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
+import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+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;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.function.BiConsumer;
+
+import static io.moquette.BrokerConstants.FLIGHT_BEFORE_RESEND_MS;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import java.util.concurrent.Semaphore;
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+// inspired by ServerIntegrationPahoTest
+public class PublishToManySubscribersUseCaseTest {
+
+    private static final Logger LOG = LoggerFactory.getLogger(PublishToManySubscribersUseCaseTest.class);
+
+    private static final int COMMAND_QUEUE_SIZE = 32;
+
+    private static final int EVENT_LOOPS = Runtime.getRuntime().availableProcessors();
+    public static final int NUM_SUBSCRIBERS = COMMAND_QUEUE_SIZE * EVENT_LOOPS * 4;
+    private Server broker;
+    private IConfig brokerConfig;
+
+    @TempDir
+    Path tempFolder;
+    private String dbPath;
+    private MqttAsyncClient publisher;
+    private List<IMqttAsyncClient> subscribers;
+
+    protected void startServer(String dbPath) throws IOException {
+        broker = new Server();
+        final Properties configProps = IntegrationUtils.prepareTestProperties(dbPath);
+        configProps.put(BrokerConstants.SESSION_QUEUE_SIZE, Integer.toString(COMMAND_QUEUE_SIZE));
+        brokerConfig = new MemoryConfig(configProps);
+        broker.startServer(brokerConfig);
+    }
+
+    @BeforeAll
+    public static void beforeTests() {
+        Awaitility.setDefaultTimeout(Durations.ONE_SECOND);
+    }
+
+    @BeforeEach
+    public void setUp() throws Exception {
+        dbPath = IntegrationUtils.tempH2Path(tempFolder);
+        startServer(dbPath);
+
+        publisher = createClient("publisher");
+        publisher.connect().waitForCompletion(1_000);
+
+        subscribers = createSubscribers(NUM_SUBSCRIBERS);
+
+        CountDownLatch latch = new CountDownLatch(NUM_SUBSCRIBERS);
+
+        final IMqttActionListener callback = createMqttCallback(latch);
+        for (IMqttAsyncClient client : subscribers) {
+            client.connect(null, callback);
+        }
+
+        latch.await();
+        LOG.debug("After all subscriptions completed");
+    }
+
+    private IMqttActionListener createMqttCallback(CountDownLatch latch) {
+        return new IMqttActionListener() {
+            @Override
+            public void onSuccess(IMqttToken asyncActionToken) {
+                latch.countDown();
+            }
+
+            @Override
+            public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
+                latch.countDown();
+            }
+        };
+    }
+
+    private IMqttActionListener createMqttCallback(Semaphore openSlots) {
+        return new IMqttActionListener() {
+            @Override
+            public void onSuccess(IMqttToken asyncActionToken) {
+                openSlots.release();
+            }
+
+            @Override
+            public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
+                openSlots.release();
+            }
+        };
+    }
+
+    private List<IMqttAsyncClient> createSubscribers(int numSubscribers) throws MqttException, IOException {
+        List<IMqttAsyncClient> clients = new ArrayList<>(numSubscribers);
+        for (int i = 0; i < numSubscribers; i++) {
+            clients.add(createClient("subscriber_" + i));
+        }
+        return clients;
+    }
+
+    private MqttAsyncClient createClient(String clientName) throws IOException, MqttException {
+        final String dataPath = IntegrationUtils.newFolder(tempFolder, clientName).getAbsolutePath();
+        MqttClientPersistence clientDataStore = new MqttDefaultFilePersistence(dataPath);
+        return new MqttAsyncClient("tcp://localhost:1883", clientName, clientDataStore);
+    }
+
+    @AfterEach
+    public void tearDown() throws Exception {
+        IntegrationUtils.disconnectClient(this.publisher);
+        disconnectAsyncClients(this.subscribers);
+
+        this.broker.stopServer();
+    }
+
+    private void disconnectAsyncClients(List<IMqttAsyncClient> clients) throws MqttException, InterruptedException {
+        CountDownLatch latch = new CountDownLatch(clients.size());
+        final IMqttActionListener completionCallback = createMqttCallback(latch);
+        for (IMqttAsyncClient client : clients) {
+            IntegrationUtils.disconnectClient(client, completionCallback);
+        }
+        latch.await();
+    }
+
+    @Test
+    void onePublishTriggerManySubscriptionsNotifications() throws MqttException, InterruptedException {
+        final LongAdder receivedPublish = new LongAdder();
+
+        //subscribe all
+        segmentedParallelSubscriptions((subscriber, completionCallback) -> {
+            try {
+                subscriber.subscribe("/temperature", 1, null, completionCallback,
+                    (String topic1, org.eclipse.paho.client.mqttv3.MqttMessage message) -> {
+                        receivedPublish.increment();
+                    });
+            } catch (MqttException e) {
+                throw new RuntimeException(e);
+            }
+        });
+
+        this.publisher.publish("/temperature", "15°C".getBytes(UTF_8), 0, false);
+
+        Awaitility.await("Waiting for resend.")
+            .atMost(FLIGHT_BEFORE_RESEND_MS * 3, TimeUnit.MILLISECONDS)
+            .pollInterval(100, TimeUnit.MILLISECONDS)
+            .untilAdder(receivedPublish, equalTo((long) NUM_SUBSCRIBERS));
+    }
+
+    private void segmentedParallelSubscriptions(BiConsumer<IMqttAsyncClient, IMqttActionListener> biConsumer) throws InterruptedException {
+        int openSlotCount = COMMAND_QUEUE_SIZE;
+        Semaphore openSlots = new Semaphore(openSlotCount);
+        IMqttActionListener completionCallback = createMqttCallback(openSlots);
+        for (IMqttAsyncClient subscriber : this.subscribers) {
+            boolean haveSlot = openSlots.tryAcquire(5, TimeUnit.SECONDS);
+            assertTrue(haveSlot, "Subscription ACK semaphore expired");
+            biConsumer.accept(subscriber, completionCallback);
+        }
+        // Wait until all subscribes are completed (all slots are free again)
+        boolean haveSlot = openSlots.tryAcquire(openSlotCount, 5, TimeUnit.SECONDS);
+        assertTrue(haveSlot, "Subscription ACK semaphore expired");
+    }
+}
diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationDBAuthenticatorTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationDBAuthenticatorTest.java
index dcb90d22..fe4f0251 100644
--- a/broker/src/test/java/io/moquette/integration/ServerIntegrationDBAuthenticatorTest.java
+++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationDBAuthenticatorTest.java
@@ -32,7 +32,7 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import java.io.File;
+
 import java.io.IOException;
 import java.nio.file.Path;
 import java.security.NoSuchAlgorithmException;
@@ -102,13 +102,8 @@ public class ServerIntegrationDBAuthenticatorTest {
 
     @AfterEach
     public void tearDown() throws Exception {
-        if (m_client != null && m_client.isConnected()) {
-            m_client.disconnect();
-        }
-
-        if (m_publisher != null && m_publisher.isConnected()) {
-            m_publisher.disconnect();
-        }
+        IntegrationUtils.disconnectClient(m_client);
+        IntegrationUtils.disconnectClient(m_publisher);
 
         stopServer();
     }
diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationMultiConnectTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationMultiConnectTest.java
index bb0d3aab..d87378ee 100644
--- a/broker/src/test/java/io/moquette/integration/ServerIntegrationMultiConnectTest.java
+++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationMultiConnectTest.java
@@ -36,7 +36,6 @@ import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.PrintStream;
 import java.nio.file.Path;
-import java.security.SecureRandom;
 import java.util.Properties;
 
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -101,10 +100,7 @@ public class ServerIntegrationMultiConnectTest {
 
     @AfterEach
     public void tearDown() throws Exception {
-        if (client != null && client.isConnected()) {
-            client.disconnect();
-        }
-
+        IntegrationUtils.disconnectClient(client);
         server.stopServer();
     }
 
diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java
index 76727677..337ee504 100644
--- a/broker/src/test/java/io/moquette/integration/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java
+++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java
@@ -113,13 +113,8 @@ public class ServerIntegrationPahoCanPublishOnReadBlockedTopicTest {
 
     @AfterEach
     public void tearDown() throws Exception {
-        if (m_client != null && m_client.isConnected()) {
-            m_client.disconnect();
-        }
-
-        if (m_publisher != null && m_publisher.isConnected()) {
-            m_publisher.disconnect();
-        }
+        IntegrationUtils.disconnectClient(m_client);
+        IntegrationUtils.disconnectClient(m_publisher);
 
         stopServer();
     }
diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationPahoTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationPahoTest.java
index 9b5013eb..7fbcd566 100644
--- a/broker/src/test/java/io/moquette/integration/ServerIntegrationPahoTest.java
+++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationPahoTest.java
@@ -82,13 +82,8 @@ public class ServerIntegrationPahoTest {
 
     @AfterEach
     public void tearDown() throws Exception {
-        if (m_client != null && m_client.isConnected()) {
-            m_client.disconnect();
-        }
-
-        if (m_publisher != null && m_publisher.isConnected()) {
-            m_publisher.disconnect();
-        }
+        IntegrationUtils.disconnectClient(m_client);
+        IntegrationUtils.disconnectClient(m_publisher);
 
         stopServer();
     }
diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationQoSValidationTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationQoSValidationTest.java
index 4000c598..870ff96a 100644
--- a/broker/src/test/java/io/moquette/integration/ServerIntegrationQoSValidationTest.java
+++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationQoSValidationTest.java
@@ -36,6 +36,7 @@ import org.slf4j.LoggerFactory;
 import java.io.IOException;
 import java.nio.file.Path;
 import java.util.Properties;
+import java.util.concurrent.TimeUnit;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -147,7 +148,8 @@ public class ServerIntegrationQoSValidationTest {
         m_subscriber.subscribe("/topic", 1);
 
         m_publisher.publish("/topic", "Hello world MQTT QoS1".getBytes(UTF_8), 1, false);
-        Awaitility.await().until(m_callback::isMessageReceived);
+        Awaitility.await().atMost(5, TimeUnit.SECONDS)
+            .until(m_callback::isMessageReceived);
         MqttMessage message = m_callback.retrieveMessage();
         assertEquals("Hello world MQTT QoS1", message.toString());
         assertEquals(1, message.getQos());
diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationRestartTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationRestartTest.java
index 795be19e..d8dd23fc 100644
--- a/broker/src/test/java/io/moquette/integration/ServerIntegrationRestartTest.java
+++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationRestartTest.java
@@ -98,7 +98,7 @@ public class ServerIntegrationRestartTest {
         m_server.stopServer();
     }
 
-    @DisplayName("given not clean session then after a server restart the session should be present")
+    @DisplayName("given not clean session after a server restart, the session is still present")
     @Test
     public void testNotCleanSessionIsVisibleAfterServerRestart() throws Exception {
         m_subscriber.connect(CLEAN_SESSION_OPT);
@@ -114,6 +114,7 @@ public class ServerIntegrationRestartTest {
 
         //reconnect subscriber and topic should be sent
         m_subscriber.connect(CLEAN_SESSION_OPT);
+
         // verify the sent message while offline is read
         Awaitility.await().until(m_messageCollector::isMessageReceived);
         MqttMessage msg = m_messageCollector.retrieveMessage();
diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLClientAuthTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLClientAuthTest.java
index ab8dcc19..3d0054c1 100644
--- a/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLClientAuthTest.java
+++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLClientAuthTest.java
@@ -204,9 +204,7 @@ public class ServerIntegrationSSLClientAuthTest {
 
     @AfterEach
     public void tearDown() throws Exception {
-        if (m_client != null && m_client.isConnected()) {
-            m_client.disconnect();
-        }
+        IntegrationUtils.disconnectClient(m_client);
 
         if (m_server != null) {
             m_server.stopServer();
diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLTest.java
index f01c1048..99ffa335 100644
--- a/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLTest.java
+++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLTest.java
@@ -135,9 +135,7 @@ public class ServerIntegrationSSLTest {
 
     @AfterEach
     public void tearDown() throws Exception {
-        if (m_client != null && m_client.isConnected()) {
-            m_client.disconnect();
-        }
+        IntegrationUtils.disconnectClient(m_client);
 
         if (m_server != null) {
             m_server.stopServer();
diff --git a/broker/src/test/java/io/moquette/integration/ServerLowlevelMessagesIntegrationTest.java b/broker/src/test/java/io/moquette/integration/ServerLowlevelMessagesIntegrationTest.java
index 1433139b..08b07d7e 100644
--- a/broker/src/test/java/io/moquette/integration/ServerLowlevelMessagesIntegrationTest.java
+++ b/broker/src/test/java/io/moquette/integration/ServerLowlevelMessagesIntegrationTest.java
@@ -116,8 +116,8 @@ public class ServerLowlevelMessagesIntegrationTest {
     }
 
     @Test
-    public void testWillMessageIsWiredOnClientKeepAliveExpiry() throws Exception {
-        LOG.info("*** testWillMessageIsWiredOnClientKeepAliveExpiry ***");
+    public void testWillMessageIsFiredOnClientKeepAliveExpiry() throws Exception {
+        LOG.info("*** testWillMessageIsFiredOnClientKeepAliveExpiry ***");
         String willTestamentTopic = "/will/test";
         String willTestamentMsg = "Bye bye";