From 06bc7a0484da9627ed242103c151f8232a176515 Mon Sep 17 00:00:00 2001 From: redlsz Date: Fri, 27 Mar 2026 19:04:59 +0800 Subject: [PATCH 1/3] fix #ISSUE 10225 --- .../proxy/common/ReceiptHandleGroup.java | 42 +++++++++++++------ .../proxy/common/ReceiptHandleGroupTest.java | 35 +++++++++++++++- 2 files changed, 64 insertions(+), 13 deletions(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroup.java b/proxy/src/main/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroup.java index 005f18025dd..781c9970e57 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroup.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroup.java @@ -113,22 +113,40 @@ public HandleData(MessageReceiptHandle messageReceiptHandle) { public Long lock(long timeoutMs) { try { - boolean result = this.semaphore.tryAcquire(timeoutMs, TimeUnit.MILLISECONDS); + long expiredTimeMs = ConfigurationManager.getProxyConfig().getLockTimeoutMsInHandleGroup() * 3; long currentTimeMs = System.currentTimeMillis(); + // Try non-blocking acquire first + if (this.semaphore.tryAcquire()) { + this.lastLockTimeMs.set(currentTimeMs); + return currentTimeMs; + } + // Check if lock is already expired before blocking + if (currentTimeMs - this.lastLockTimeMs.get() > expiredTimeMs) { + synchronized (this) { + if (currentTimeMs - this.lastLockTimeMs.get() > expiredTimeMs) { + log.warn("HandleData lock expired, acquire lock success without blocking " + + "and reset lock time. MessageReceiptHandle={}, lockTime={}", + messageReceiptHandle, currentTimeMs); + this.lastLockTimeMs.set(currentTimeMs); + return currentTimeMs; + } + } + } + // Try blocking acquire with timeout + boolean result = this.semaphore.tryAcquire(timeoutMs, TimeUnit.MILLISECONDS); + currentTimeMs = System.currentTimeMillis(); if (result) { this.lastLockTimeMs.set(currentTimeMs); return currentTimeMs; - } else { - // if the lock is expired, can be acquired again - long expiredTimeMs = ConfigurationManager.getProxyConfig().getLockTimeoutMsInHandleGroup() * 3; - if (currentTimeMs - this.lastLockTimeMs.get() > expiredTimeMs) { - synchronized (this) { - if (currentTimeMs - this.lastLockTimeMs.get() > expiredTimeMs) { - log.warn("HandleData lock expired, acquire lock success and reset lock time. " + - "MessageReceiptHandle={}, lockTime={}", messageReceiptHandle, currentTimeMs); - this.lastLockTimeMs.set(currentTimeMs); - return currentTimeMs; - } + } + // Recheck expiration after timeout + if (currentTimeMs - this.lastLockTimeMs.get() > expiredTimeMs) { + synchronized (this) { + if (currentTimeMs - this.lastLockTimeMs.get() > expiredTimeMs) { + log.warn("HandleData lock expired, acquire lock success and reset lock time. " + + "MessageReceiptHandle={}, lockTime={}", messageReceiptHandle, currentTimeMs); + this.lastLockTimeMs.set(currentTimeMs); + return currentTimeMs; } } } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroupTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroupTest.java index fdc1c5a3965..5472d18247c 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroupTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroupTest.java @@ -21,12 +21,14 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.apache.rocketmq.common.consumer.ReceiptHandle; import org.apache.rocketmq.common.message.MessageClientIDSetter; import org.apache.rocketmq.common.utils.FutureUtils; +import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.proxy.config.InitConfigTest; import org.junit.Before; import org.junit.Test; @@ -312,4 +314,35 @@ public void testRemoveOne() { private MessageReceiptHandle createMessageReceiptHandle(String handle, String msgID) { return new MessageReceiptHandle(GROUP, TOPIC, 0, handle, msgID, 0, 0); } -} \ No newline at end of file + + @Test + public void testAcquireExpiredLock() throws InterruptedException { + final long lockTimeoutMs = 100L; + ConfigurationManager.getProxyConfig().setLockTimeoutMsInHandleGroup(lockTimeoutMs); + + String handle = ReceiptHandle.builder() + .startOffset(0L) + .retrieveTime(System.currentTimeMillis()) + .invisibleTime(3000) + .reviveQueueId(1) + .topicType(ReceiptHandle.NORMAL_TOPIC) + .brokerName("brokerName") + .queueId(1) + .offset(1L) + .commitLogOffset(0L) + .build().encode(); + receiptHandleGroup.put(msgID, createMessageReceiptHandle(handle, msgID)); + + // Simulate a slow task via a never-completing future + CompletableFuture slowFuture = new CompletableFuture<>(); + receiptHandleGroup.computeIfPresent(msgID, handle, h -> slowFuture); + // Then make the lock expired + TimeUnit.MILLISECONDS.sleep(3 * lockTimeoutMs); + + // Another caller can acquire this expired lock without blocking + long startTime = System.currentTimeMillis(); + receiptHandleGroup.remove(msgID, handle); + long elapsed = System.currentTimeMillis() - startTime; + assertTrue(elapsed < lockTimeoutMs); + } +} From e3d8ef3b32a5cf7974f1784cd9bf64fccc74b7cf Mon Sep 17 00:00:00 2001 From: redlsz Date: Fri, 27 Mar 2026 21:07:03 +0800 Subject: [PATCH 2/3] polish test --- .../proxy/common/ReceiptHandleGroupTest.java | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroupTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroupTest.java index 5472d18247c..9e3edc60dfe 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroupTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroupTest.java @@ -320,17 +320,7 @@ public void testAcquireExpiredLock() throws InterruptedException { final long lockTimeoutMs = 100L; ConfigurationManager.getProxyConfig().setLockTimeoutMsInHandleGroup(lockTimeoutMs); - String handle = ReceiptHandle.builder() - .startOffset(0L) - .retrieveTime(System.currentTimeMillis()) - .invisibleTime(3000) - .reviveQueueId(1) - .topicType(ReceiptHandle.NORMAL_TOPIC) - .brokerName("brokerName") - .queueId(1) - .offset(1L) - .commitLogOffset(0L) - .build().encode(); + String handle = createHandle(); receiptHandleGroup.put(msgID, createMessageReceiptHandle(handle, msgID)); // Simulate a slow task via a never-completing future From 2c3fd5fbb91099b8692f856350df57b4de1c18f7 Mon Sep 17 00:00:00 2001 From: redlsz Date: Wed, 1 Apr 2026 10:22:18 +0800 Subject: [PATCH 3/3] polish --- .../org/apache/rocketmq/proxy/common/ReceiptHandleGroup.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroup.java b/proxy/src/main/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroup.java index 781c9970e57..82ff25c534b 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroup.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroup.java @@ -113,7 +113,6 @@ public HandleData(MessageReceiptHandle messageReceiptHandle) { public Long lock(long timeoutMs) { try { - long expiredTimeMs = ConfigurationManager.getProxyConfig().getLockTimeoutMsInHandleGroup() * 3; long currentTimeMs = System.currentTimeMillis(); // Try non-blocking acquire first if (this.semaphore.tryAcquire()) { @@ -121,6 +120,7 @@ public Long lock(long timeoutMs) { return currentTimeMs; } // Check if lock is already expired before blocking + long expiredTimeMs = ConfigurationManager.getProxyConfig().getLockTimeoutMsInHandleGroup() * 3; if (currentTimeMs - this.lastLockTimeMs.get() > expiredTimeMs) { synchronized (this) { if (currentTimeMs - this.lastLockTimeMs.get() > expiredTimeMs) {