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 <[email protected]> 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
