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 651c8d881f103cd637a9dba6e69a1661fee63278 Author: AloysZhang <[email protected]> AuthorDate: Mon Jul 17 15:43:52 2023 +0800 [fix][client] fix negative message re-delivery twice issue (#20750) (cherry picked from commit ecd16d68e29700cc9521e6d77d124a403c3486c9) (cherry picked from commit cbc3ea169367e7e14cae3682f36e0da4fc9b587f) --- .../java/org/apache/pulsar/client/impl/NegativeAcksTest.java | 3 +++ .../java/org/apache/pulsar/client/impl/ConsumerBase.java | 4 ++++ .../java/org/apache/pulsar/client/impl/ConsumerImpl.java | 3 ++- .../apache/pulsar/client/impl/MultiTopicsConsumerImpl.java | 3 +++ .../org/apache/pulsar/client/impl/NegativeAcksTracker.java | 12 +++++++++--- .../org/apache/pulsar/client/impl/UnAckedMessageTracker.java | 8 ++++++++ 6 files changed, 29 insertions(+), 4 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 30aefd90b80..576b8778251 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 @@ -150,6 +150,9 @@ public class NegativeAcksTest extends ProducerConsumerBase { consumer.negativeAcknowledge(msg); } + assertTrue(consumer instanceof ConsumerBase<String>); + assertEquals(((ConsumerBase<String>) consumer).getUnAckedMessageTracker().size(), 0); + Set<String> receivedMessages = new HashSet<>(); // All the messages should be received again diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java index 5d77da8a046..7233480bfd5 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java @@ -174,6 +174,10 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T initReceiverQueueSize(); } + protected UnAckedMessageTracker getUnAckedMessageTracker() { + return unAckedMessageTracker; + } + protected void triggerBatchReceiveTimeoutTask() { if (!hasBatchReceiveTimeout() && batchReceivePolicy.getTimeoutMs() > 0) { batchReceiveTimeout = client.timer().newTimeout(this::pendingBatchReceiveTask, 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 35f47076a9b..85cafe644fb 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 @@ -394,6 +394,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle return connectionHandler; } + @Override public UnAckedMessageTracker getUnAckedMessageTracker() { return unAckedMessageTracker; } @@ -805,7 +806,7 @@ 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(message.getMessageId()); + unAckedMessageTracker.remove(MessageIdAdvUtils.discardBatch(message.getMessageId())); } @Override diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java index 52d8f56a812..36fdf4427ea 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java @@ -597,6 +597,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> { ConsumerImpl<T> consumer = consumers.get(topicMessageId.getTopicPartitionName()); consumer.negativeAcknowledge(topicMessageId.getInnerMessageId()); + unAckedMessageTracker.remove(messageId); } @Override @@ -607,6 +608,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> { ConsumerImpl<T> consumer = consumers.get(topicMessageId.getTopicPartitionName()); consumer.negativeAcknowledge(message); + unAckedMessageTracker.remove(messageId); } @Override @@ -883,6 +885,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> { return stats; } + @Override public UnAckedMessageTracker getUnAckedMessageTracker() { return unAckedMessageTracker; } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NegativeAcksTracker.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NegativeAcksTracker.java index 6273f4d582e..f24872fb619 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NegativeAcksTracker.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NegativeAcksTracker.java @@ -30,8 +30,11 @@ import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.RedeliveryBackoff; import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; class NegativeAcksTracker implements Closeable { + private static final Logger log = LoggerFactory.getLogger(NegativeAcksTracker.class); private HashMap<MessageId, Long> nackedMessages = null; @@ -77,9 +80,12 @@ class NegativeAcksTracker implements Closeable { } }); - messagesToRedeliver.forEach(nackedMessages::remove); - consumer.onNegativeAcksSend(messagesToRedeliver); - consumer.redeliverUnacknowledgedMessages(messagesToRedeliver); + if (!messagesToRedeliver.isEmpty()) { + messagesToRedeliver.forEach(nackedMessages::remove); + consumer.onNegativeAcksSend(messagesToRedeliver); + log.info("[{}] {} messages will be re-delivered", consumer, messagesToRedeliver.size()); + consumer.redeliverUnacknowledgedMessages(messagesToRedeliver); + } this.timeout = timer.newTimeout(this::triggerRedelivery, timerIntervalNanos, TimeUnit.NANOSECONDS); } 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 index 5a706720dc5..5386f6d4fe5 100644 --- 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,6 +192,10 @@ public class UnAckedMessageTracker implements Closeable { } public boolean add(MessageId messageId) { + if (messageId == null) { + return false; + } + writeLock.lock(); try { HashSet<MessageId> partition = timePartitions.peekLast(); @@ -220,6 +224,10 @@ public class UnAckedMessageTracker implements Closeable { } public boolean remove(MessageId messageId) { + if (messageId == null) { + return false; + } + writeLock.lock(); try { boolean removed = false;
