fix

Change-Id: I9f233816eb3720505f0389369da0267c2d47beed

Author: qianye1001

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

@@ -268,13 +268,18 @@ public MessageReceiptHandle removeOne(String msgID) {
 
     public void computeIfPresent(String msgID, String handle,
         Function<MessageReceiptHandle, CompletableFuture<MessageReceiptHandle>> function) {
+        long timeout = ConfigurationManager.getProxyConfig().getLockTimeoutMsInHandleGroup();
+        computeIfPresent(msgID, handle, function, timeout);
+    }
+
+    public void computeIfPresent(String msgID, String handle,
+        Function<MessageReceiptHandle, CompletableFuture<MessageReceiptHandle>> function, long lockTimeout) {
         Map<HandleKey, HandleData> handleMap = this.receiptHandleMap.get(msgID);
         if (handleMap == null) {
             return;
         }
-        long timeout = ConfigurationManager.getProxyConfig().getLockTimeoutMsInHandleGroup();
         handleMap.computeIfPresent(new HandleKey(handle), (handleKey, handleData) -> {
-            Long lockTimeMs = handleData.lock(timeout);
+            Long lockTimeMs = handleData.lock(lockTimeout);
             if (lockTimeMs == null) {
                 throw new ProxyException(ProxyExceptionCode.INTERNAL_SERVER_ERROR, "try to compute failed");
             }

proxy/src/main/java/org/apache/rocketmq/proxy/service/receipt/DefaultReceiptHandleManager.java

@@ -40,6 +40,7 @@
 import org.apache.rocketmq.common.thread.ThreadPoolMonitor;
 import org.apache.rocketmq.common.utils.AbstractStartAndShutdown;
 import org.apache.rocketmq.common.utils.ConcurrentHashMapUtils;
+import org.apache.rocketmq.common.utils.ExceptionUtils;
 import org.apache.rocketmq.common.utils.StartAndShutdown;
 import org.apache.rocketmq.common.utils.ThreadUtils;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
@@ -53,7 +54,6 @@
 import org.apache.rocketmq.proxy.common.RenewEvent;
 import org.apache.rocketmq.proxy.common.RenewStrategyPolicy;
 import org.apache.rocketmq.proxy.common.channel.ChannelHelper;
-import org.apache.rocketmq.common.utils.ExceptionUtils;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
 import org.apache.rocketmq.proxy.config.ProxyConfig;
 import org.apache.rocketmq.proxy.service.metadata.MetadataService;
@@ -172,7 +172,7 @@ protected void scheduleRenewTask() {
 
     protected void renewMessage(ProxyContext context, ReceiptHandleGroupKey key, ReceiptHandleGroup group, String msgID, String handleStr) {
         try {
-            group.computeIfPresent(msgID, handleStr, messageReceiptHandle -> startRenewMessage(context, key, messageReceiptHandle));
+            group.computeIfPresent(msgID, handleStr, messageReceiptHandle -> startRenewMessage(context, key, messageReceiptHandle), 0);
         } catch (Exception e) {
             log.error("error when renew message. msgID:{}, handleStr:{}", msgID, handleStr, e);
         }