Fix qos1 pubAck action is not executed forever when log level is not …

…debug (#69) (#83) (#84)
streamnative · Aug 16, 2021 · 7cd2fd3 · 7cd2fd3
1 parent de2e003
commit 7cd2fd3
Showing 1 changed file with 9 additions and 9 deletions.
diff --git a/...c/main/java/io/streamnative/pulsar/handlers/mqtt/support/ProtocolMethodProcessorImpl.java b/...c/main/java/io/streamnative/pulsar/handlers/mqtt/support/ProtocolMethodProcessorImpl.java
@@ -186,15 +186,15 @@ public void processConnect(Channel channel, MqttConnectMessage msg) {
     public void processPubAck(Channel channel, MqttPubAckMessage msg) {
         if (log.isDebugEnabled()) {
             log.debug("[PubAck] [{}] msg: {}", channel, msg);
-            int packetId = msg.variableHeader().messageId();
-            OutstandingPacket packet = outstandingPacketContainer.remove(packetId);
-            if (packet != null) {
-                packet.getConsumer().getSubscription().acknowledgeMessage(
-                        Collections.singletonList(PositionImpl.get(packet.getLedgerId(), packet.getEntryId())),
-                        CommandAck.AckType.Individual, Collections.emptyMap());
-                packet.getConsumer().getPendingAcks().remove(packet.getLedgerId(), packet.getEntryId());
-                packet.getConsumer().incrementPermits();
-            }
+        }
+        int packetId = msg.variableHeader().messageId();
+        OutstandingPacket packet = outstandingPacketContainer.remove(packetId);
+        if (packet != null) {
+            packet.getConsumer().getSubscription().acknowledgeMessage(
+                    Collections.singletonList(PositionImpl.get(packet.getLedgerId(), packet.getEntryId())),
+                    CommandAck.AckType.Individual, Collections.emptyMap());
+            packet.getConsumer().getPendingAcks().remove(packet.getLedgerId(), packet.getEntryId());
+            packet.getConsumer().incrementPermits();
         }
     }