Keep consistent type for message id adding to and removing from the U…

…nAckedMessageTracker
apache · Jul 16, 2023 · 305ac2e · 305ac2e
1 parent 4a6a1ef
commit 305ac2e
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 3 deletions.
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -757,7 +757,7 @@ public void negativeAcknowledge(Message<?> message) {
         negativeAcksTracker.add(message);
 
         // Ensure the message is not redelivered for ack-timeout, since we did receive an "ack"
-        unAckedMessageTracker.remove(message.getMessageId());
+        unAckedMessageTracker.remove(MessageIdAdvUtils.discardBatch(message.getMessageId()));
     }
 
     @Override

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java
@@ -192,7 +192,7 @@ public boolean add(MessageId messageId) {
         if (messageId == null) {
             return false;
         }
-        messageId = MessageIdAdvUtils.discardBatch(messageId);
+
         writeLock.lock();
         try {
             HashSet<MessageId> partition = timePartitions.peekLast();
@@ -224,7 +224,7 @@ public boolean remove(MessageId messageId) {
         if (messageId == null) {
             return false;
         }
-        messageId = MessageIdAdvUtils.discardBatch(messageId);
+
         writeLock.lock();
         try {
             boolean removed = false;