You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/08/09 14:11:59 UTC
(pulsar) 02/03: [fix][build] Fix compatibility issue introduced by #20750
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 581ad607ced40e7a157b6b38bfe6f3914984581a
Author: xymeng <19...@qq.com>
AuthorDate: Tue Jan 23 12:46:02 2024 +0800
[fix][build] Fix compatibility issue introduced by #20750
(cherry picked from commit 05ac1f9dda575e668627df966504f2ba422354ea)
---
.../org/apache/pulsar/client/impl/NegativeAcksTest.java | 3 ++-
.../java/org/apache/pulsar/client/impl/ConsumerImpl.java | 15 ++++++++++++++-
2 files changed, 16 insertions(+), 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
index 576b8778251..a25c4dee242 100644
--- 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 class NegativeAcksTest extends ProducerConsumerBase {
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
index 85cafe644fb..0c3b8da4396 100644
--- 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 class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
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