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 2026/07/22 10:02:39 UTC
(pulsar) 04/15: [fix][client] Fix UnAckedMessageRedeliveryTracker to skip cancelled timeouts (#26043)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4c47857b0160eb1775ea505ee298ee453fe4aa83
Author: Dream95 <86...@qq.com>
AuthorDate: Fri Jul 3 21:35:56 2026 +0800
[fix][client] Fix UnAckedMessageRedeliveryTracker to skip cancelled timeouts (#26043)
Signed-off-by: Dream95 <zh...@163.com>
(cherry picked from commit e85b1f7d3b69fb418afbf467f53ac58ce876d07c)
---
.../pulsar/client/impl/UnAckedMessageRedeliveryTracker.java | 11 +++++++++--
1 file changed, 9 insertions(+), 2 deletions(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageRedeliveryTracker.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageRedeliveryTracker.java
index 1405e279f8fe..076616d87d40 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageRedeliveryTracker.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageRedeliveryTracker.java
@@ -59,6 +59,10 @@ public class UnAckedMessageRedeliveryTracker extends UnAckedMessageTracker {
timeout = client.timer().newTimeout(new TimerTask() {
@Override
public void run(Timeout t) throws Exception {
+ if (t.isCancelled()) {
+ return;
+ }
+
writeLock.lock();
try {
HashSet<UnackMessageIdWrapper> headPartition = redeliveryTimePartitions.removeFirst();
@@ -73,8 +77,11 @@ public class UnAckedMessageRedeliveryTracker extends UnAckedMessageTracker {
redeliveryTimePartitions.addLast(headPartition);
triggerRedelivery(consumerBase);
} finally {
- writeLock.unlock();
- timeout = client.timer().newTimeout(this, tickDurationInMs, TimeUnit.MILLISECONDS);
+ try {
+ timeout = client.timer().newTimeout(this, tickDurationInMs, TimeUnit.MILLISECONDS);
+ } finally {
+ writeLock.unlock();
+ }
}
}
}, this.tickDurationInMs, TimeUnit.MILLISECONDS);