[fix][build] Fix compatibility issue introduced by #20750

(cherry picked from commit 05ac1f9)
apache · Aug 9, 2024 · 581ad60 · 581ad60
1 parent 651c8d8
commit 581ad60
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 2 deletions.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java
@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.client.impl;
 
+import static org.junit.Assert.assertTrue;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNull;
 
@@ -150,7 +151,7 @@ public void testNegativeAcks(boolean batching, boolean usePartitions, Subscripti
             consumer.negativeAcknowledge(msg);
         }
 
-        assertTrue(consumer instanceof ConsumerBase<String>);
+        assertTrue(consumer instanceof ConsumerBase);
         assertEquals(((ConsumerBase<String>) consumer).getUnAckedMessageTracker().size(), 0);
 
         Set<String> receivedMessages = new HashSet<>();

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
@@ -806,7 +806,20 @@ 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(MessageIdAdvUtils.discardBatch(message.getMessageId()));
+        unAckedMessageTracker.remove(discardBatch(message.getMessageId()));
+    }
+
+    static MessageIdImpl discardBatch(MessageId messageId) {
+        if (messageId instanceof ChunkMessageIdImpl) {
+            return (MessageIdImpl) messageId;
+        }
+        MessageIdImpl msgId;
+        if (messageId instanceof TopicMessageIdImpl) {
+           msgId = (MessageIdImpl) ((TopicMessageIdImpl) messageId).getInnerMessageId();
+        } else {
+            msgId = (MessageIdImpl) messageId;
+        }
+        return new MessageIdImpl(msgId.getLedgerId(), msgId.getEntryId(), msgId.getPartitionIndex());
     }
 
     @Override