diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java
index 500a8a88..39e7c493 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java
@@ -26,6 +26,8 @@ public interface IStorageService {
     void storePublishForFuture(PublishEvent evt);
 
     List<PublishEvent> retrivePersistedPublishes(String clientID);
+    
+    void cleanPersistedPublishMessage(String clientID, int messageID);
 
     void cleanPersistedPublishes(String clientID);
 
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
index f0349be8..c2df8cb1 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
@@ -55,6 +55,7 @@ public class HawtDBStorageService implements IStorageService {
     private PageFileFactory pageFactory;
 
     //maps clientID to the list of pending messages stored
+    //TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map)
     private SortedIndex<String, List<StoredPublishEvent>> m_persistentMessageStore;
     private SortedIndex<String, StoredMessage> m_retainedStore;
     //bind clientID+MsgID -> evt message published
@@ -185,6 +186,21 @@ public class HawtDBStorageService implements IStorageService {
         }
         return liveEvts;
     }
+    
+    public void cleanPersistedPublishMessage(String clientID, int messageID) {
+        List<StoredPublishEvent> events = m_persistentMessageStore.get(clientID);
+        if (events == null) {
+            return;
+        }
+        StoredPublishEvent toRemoveEvt = null;
+        for (StoredPublishEvent evt : events) {
+            if (evt.getMessageID() == messageID) {
+                toRemoveEvt = evt;
+            }
+        }
+        events.remove(toRemoveEvt);
+        m_persistentMessageStore.put(clientID, events);
+    }
 
     public void cleanPersistedPublishes(String clientID) {
         m_persistentMessageStore.remove(clientID);
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java
index 7570832e..79ca189e 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java
@@ -23,6 +23,7 @@ public class MemoryStorageService implements IStorageService {
     
     private Map<String, Set<Subscription>> m_persistentSubscriptions = new HashMap<String, Set<Subscription>>();
     private Map<String, HawtDBStorageService.StoredMessage> m_retainedStore = new HashMap<String, HawtDBStorageService.StoredMessage>();
+    //TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map)
     private Map<String, List<PublishEvent>> m_persistentMessageStore = new HashMap<String, List<PublishEvent>>();
     private Map<String, PublishEvent> m_inflightStore = new HashMap<String, PublishEvent>();
     private Map<String, PublishEvent> m_qos2Store = new HashMap<String, PublishEvent>();
@@ -76,6 +77,18 @@ public class MemoryStorageService implements IStorageService {
     public List<PublishEvent> retrivePersistedPublishes(String clientID) {
         return m_persistentMessageStore.get(clientID);
     }
+    
+    public void cleanPersistedPublishMessage(String clientID, int messageID) {
+        List<PublishEvent> events = m_persistentMessageStore.get(clientID);
+        PublishEvent toRemoveEvt = null;
+        for (PublishEvent evt : events) {
+            if (evt.getMessageID() == messageID) {
+                toRemoveEvt = evt;
+            }
+        }
+        events.remove(toRemoveEvt);
+        m_persistentMessageStore.put(clientID, events);
+    }
 
     public void cleanPersistedPublishes(String clientID) {
         m_persistentMessageStore.remove(clientID);
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
index ae1eb3ad..7c0562ea 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
@@ -167,7 +167,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
     }
     
     private void republishStored(String clientID) {
-        LOG.debug("republishStored invoked");
+        LOG.trace("republishStored invoked");
         List<PublishEvent> publishedEvents = m_storageService.retrivePersistedPublishes(clientID);
         if (publishedEvents == null) {
             LOG.info("No stored messages for client <{}>", clientID);
@@ -181,6 +181,11 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         }
     }
     
+    void processPubAck(String clientID, int messageID) {
+        //Remove the message from message store
+        m_storageService.cleanPersistedPublishMessage(clientID, messageID);
+    }
+    
     private void processRemoveAllSubscriptions(String clientID) {
         LOG.info("cleaning old saved subscriptions for client <{}>", clientID);
         subscriptions.removeForClient(clientID);
@@ -190,7 +195,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
     }
     
     protected void processPublish(PublishEvent evt) {
-        LOG.debug("processPublish invoked with " + evt);
+        LOG.trace("processPublish invoked with {}", evt);
         final String topic = evt.getTopic();
         final AbstractMessage.QOSType qos = evt.getQos();
         final ByteBuffer message = evt.getMessage();
@@ -287,13 +292,13 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
             LOG.debug("Session for clientId {} is {}", clientId, m_clientIDs.get(clientId).getSession());
 //            m_clientIDs.get(clientId).getSession().write(pubMessage);
             disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), pubMessage));
-        }catch(Throwable t) {
+        } catch(Throwable t) {
             LOG.error(null, t);
         }
     }
     
     private void sendPubRec(String clientID, int messageID) {
-        LOG.debug("sendPubRec invoked for clientID {} with messageID {}", clientID, messageID);
+        LOG.trace("sendPubRec invoked for clientID {} with messageID {}", clientID, messageID);
         PubRecMessage pubRecMessage = new PubRecMessage();
         pubRecMessage.setMessageID(messageID);
 
@@ -302,7 +307,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
     }
     
     private void sendPubAck(PubAckEvent evt) {
-        LOG.debug("sendPubAck invoked");
+        LOG.trace("sendPubAck invoked");
 
         String clientId = evt.getClientID();
 
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
index 8025246c..c0e39d87 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
@@ -148,6 +148,10 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
                 String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
                 int messageID = ((PubCompMessage) message).getMessageID();
                 m_processor.processPubComp(clientID, messageID);
+            } else if (message instanceof PubAckMessage) {
+                String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
+                int messageID = ((PubAckMessage) message).getMessageID();
+                m_processor.processPubAck(clientID, messageID);
             } else {
                 throw new RuntimeException("Illegal message received " + message);
             }
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java
index 4e6d1aa0..b7146549 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java
@@ -23,7 +23,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
     
     private static final Logger LOG = LoggerFactory.getLogger(NettyMQTTHandler.class);
     private IMessaging m_messaging;
-    private Map<ChannelHandlerContext, NettyChannel> m_channelMapper = new HashMap<ChannelHandlerContext, NettyChannel>();
+    private final Map<ChannelHandlerContext, NettyChannel> m_channelMapper = new HashMap<ChannelHandlerContext, NettyChannel>();
     
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object message) {
@@ -39,6 +39,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
                 case PUBCOMP:
                 case PUBREL:
                 case DISCONNECT:
+                case PUBACK:    
                     NettyChannel channel;
                     synchronized(m_channelMapper) {
                         if (!m_channelMapper.containsKey(ctx)) {