diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
index f0c799c8..0bcdf9f4 100755
--- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
@@ -608,8 +608,9 @@ public class ProtocolProcessor {
         channel.closeFuture();
 
         //disconnect the session
-
-        m_sessionsStore.sessionForClient(clientID).disconnect(session.getUsername(), isDup, isRetain);
+        if(session != null) {
+            m_sessionsStore.sessionForClient(clientID).disconnect(session.getUsername(), isDup, isRetain);
+        }
     }
 
 
diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
index b28fd33d..7065bbdd 100755
--- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
+++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
@@ -252,6 +252,13 @@ public class Qos1PublishHandler extends QosPublishHandler {
         
         MemorySessionStore.Session session = m_sessionStore.getSession(clientID);
         payloadContent = AES.AESDecrypt(payloadContent, session.getSecret(), true);
+        if(payloadContent == null) {
+            ByteBuf ackPayload = Unpooled.buffer();
+            ackPayload.ensureWritable(1).writeByte(ERROR_CODE_INVALID_DATA.getCode());
+            sendPubAck(clientID, messageID, ackPayload, ERROR_CODE_INVALID_DATA);
+            return;
+        }
+
         imHandler(clientID, username, imtopic, payloadContent, (errorCode, ackPayload) -> sendPubAck(clientID, messageID, ackPayload, errorCode), ProtoConstants.RequestSourceType.Request_From_User);
     }