[ISSUE #10225] Avoid waiting on expired receipt handle locks

proxy/src/main/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroup.java

@@ -113,34 +113,48 @@ public HandleData(MessageReceiptHandle messageReceiptHandle) {
 
         public Long lock(long timeoutMs) {
             try {
-                boolean result = this.semaphore.tryAcquire(timeoutMs, TimeUnit.MILLISECONDS);
+                // Try non-blocking acquire first, preserving interruption handling.
+                boolean result = this.semaphore.tryAcquire(0, TimeUnit.MILLISECONDS);
                 long 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;
-                            }
-                        }
-                    }
                 }
-                return null;
+                // Check expiration before waiting for the semaphore.
+                long expiredTimeMs = ConfigurationManager.getProxyConfig().getLockExpiredTimeMsInHandleGroup();
+                Long lockTimeMs = tryAcquireExpiredLock(currentTimeMs, expiredTimeMs);
+                if (lockTimeMs != null) {
+                    return lockTimeMs;
+                }
+                result = this.semaphore.tryAcquire(timeoutMs, TimeUnit.MILLISECONDS);
+                currentTimeMs = System.currentTimeMillis();
+                if (result) {
+                    this.lastLockTimeMs.set(currentTimeMs);
+                    return currentTimeMs;
+                }
+                return tryAcquireExpiredLock(currentTimeMs, expiredTimeMs);
             } catch (InterruptedException e) {
                 return null;
             }
         }
 
+        private Long tryAcquireExpiredLock(long currentTimeMs, long expiredTimeMs) {
+            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;
+                    }
+                }
+            }
+            return null;
+        }
+
         public void unlock(long lockTimeMs) {
             // if the lock is expired, we don't need to unlock it
-            if (System.currentTimeMillis() - lockTimeMs > ConfigurationManager.getProxyConfig().getLockTimeoutMsInHandleGroup() * 2) {
+            if (System.currentTimeMillis() - lockTimeMs > ConfigurationManager.getProxyConfig().getLockExpiredTimeMsInHandleGroup()) {
                 log.warn("HandleData lock expired, unlock fail. MessageReceiptHandle={}, lockTime={}, now={}",
                     messageReceiptHandle, lockTimeMs, System.currentTimeMillis());
                 return;

proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java

@@ -249,6 +249,7 @@ public class ProxyConfig implements ConfigFile {
     private int renewMaxThreadPoolNums = 4;
     private int renewThreadPoolQueueCapacity = 300;
     private long lockTimeoutMsInHandleGroup = TimeUnit.SECONDS.toMillis(3);
+    private long lockExpiredTimeMsInHandleGroup = TimeUnit.SECONDS.toMillis(4);
     private long renewAheadTimeMillis = TimeUnit.SECONDS.toMillis(10);
     private long renewMaxTimeMillis = TimeUnit.HOURS.toMillis(3);
     private long renewSchedulePeriodMillis = TimeUnit.SECONDS.toMillis(5);
@@ -1223,6 +1224,14 @@ public void setLockTimeoutMsInHandleGroup(long lockTimeoutMsInHandleGroup) {
         this.lockTimeoutMsInHandleGroup = lockTimeoutMsInHandleGroup;
     }
 
+    public long getLockExpiredTimeMsInHandleGroup() {
+        return lockExpiredTimeMsInHandleGroup;
+    }
+
+    public void setLockExpiredTimeMsInHandleGroup(long lockExpiredTimeMsInHandleGroup) {
+        this.lockExpiredTimeMsInHandleGroup = lockExpiredTimeMsInHandleGroup;
+    }
+
     public long getRenewAheadTimeMillis() {
         return renewAheadTimeMillis;
     }

proxy/src/test/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroupTest.java

@@ -16,24 +16,28 @@
  */
 package org.apache.rocketmq.proxy.common;
 
+import java.lang.reflect.Field;
 import java.time.Duration;
 import java.util.Random;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 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;
 
 import static org.awaitility.Awaitility.await;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
@@ -312,4 +316,43 @@ public void testRemoveOne() {
     private MessageReceiptHandle createMessageReceiptHandle(String handle, String msgID) {
         return new MessageReceiptHandle(GROUP, TOPIC, 0, handle, msgID, 0, 0);
     }
-}
+
+    @Test
+    public void testInterruptedLockAcquisition() {
+        ReceiptHandleGroup.HandleData data = new ReceiptHandleGroup.HandleData(null);
+        Thread.currentThread().interrupt();
+        try {
+            assertNull(data.lock(0));
+            assertFalse(Thread.currentThread().isInterrupted());
+        } finally {
+            Thread.interrupted();
+        }
+        assertNotNull(data.lock(0));
+    }
+
+    @Test
+    public void testAcquireExpiredLockWithoutWaiting() throws Exception {
+        assertEquals(4000, ConfigurationManager.getProxyConfig().getLockExpiredTimeMsInHandleGroup());
+        ConfigurationManager.getProxyConfig().setLockExpiredTimeMsInHandleGroup(8000);
+        ReceiptHandleGroup.HandleData data = new ReceiptHandleGroup.HandleData(null);
+        assertNotNull(data.lock(0));
+        assertNull(data.lock(0));
+        Field field = ReceiptHandleGroup.HandleData.class.getDeclaredField("lastLockTimeMs");
+        field.setAccessible(true);
+        AtomicLong lastLockTimeMs = (AtomicLong) field.get(data);
+        lastLockTimeMs.set(System.currentTimeMillis() - 7000);
+        assertNull(data.lock(0));
+        long expiredLockTime = System.currentTimeMillis() - 8100;
+        lastLockTimeMs.set(expiredLockTime);
+
+        long start = System.nanoTime();
+        Long lockTime = data.lock(3000);
+        assertNotNull(lockTime);
+        assertTrue(Duration.ofNanos(System.nanoTime() - start).toMillis() < 1000);
+        // The old owner must not release the semaphore after the configured expiration.
+        data.unlock(expiredLockTime);
+        assertNull(data.lock(0));
+        data.unlock(lockTime);
+        assertNotNull(data.lock(0));
+    }
+}