[ISSUE #3949] Support autoRenew

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

@@ -22,4 +22,5 @@ public class ContextVariable {
     public final static String LOCAL_ADDRESS = "local-address";
     public static final String CLIENT_ID = "client-id";
     public static final String LANGUAGE = "language";
+    public final static String CHANNEL_KEY = "channel-key";
 }

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

@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.proxy.common;
+
+import org.apache.rocketmq.common.message.MessageQueue;
+
+public class MessageReceiptHandle {
+    private final String group;
+    private final MessageQueue messageQueue;
+    private final String messageId;
+    private final long queueOffset;
+    private final String originalReceiptHandle;
+    private final long timestamp;
+    private final int reconsumeTimes;
+    private final long expectInvisibleTime;
+
+    private String receiptHandle;
+
+    public MessageReceiptHandle(String group, MessageQueue messageQueue, String receiptHandle, String messageId,
+        long queueOffset, int reconsumeTimes, long expectInvisibleTime) {
+        this.group = group;
+        this.messageQueue = messageQueue;
+        this.receiptHandle = receiptHandle;
+        this.originalReceiptHandle = receiptHandle;
+        this.messageId = messageId;
+        this.queueOffset = queueOffset;
+        this.reconsumeTimes = reconsumeTimes;
+        this.expectInvisibleTime = expectInvisibleTime;
+        this.timestamp = System.currentTimeMillis();
+    }
+
+    public String getGroup() {
+        return group;
+    }
+
+    public MessageQueue getMessageQueue() {
+        return messageQueue;
+    }
+
+    public String getReceiptHandle() {
+        return receiptHandle;
+    }
+
+    public String getOriginalReceiptHandle() {
+        return originalReceiptHandle;
+    }
+
+    public String getMessageId() {
+        return messageId;
+    }
+
+    public long getQueueOffset() {
+        return queueOffset;
+    }
+
+    public int getReconsumeTimes() {
+        return reconsumeTimes;
+    }
+
+    public long getTimestamp() {
+        return timestamp;
+    }
+
+    public long getExpectInvisibleTime() {
+        return expectInvisibleTime;
+    }
+
+    public void update(String receiptHandle) {
+        this.receiptHandle = receiptHandle;
+    }
+}

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

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.proxy.common;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class ReceiptHandleGroup {
+    private final Map<String, MessageReceiptHandle> receiptHandleMap = new ConcurrentHashMap<>();
+
+    public void put(String key, MessageReceiptHandle value) {
+        receiptHandleMap.put(key, value);
+    }
+
+    public void remove(String key) {
+        receiptHandleMap.remove(key);
+    }
+
+    public Map<String, MessageReceiptHandle> all() {
+        return ImmutableMap.copyOf(receiptHandleMap);
+    }
+}

proxy/src/main/java/org/apache/rocketmq/proxy/common/utils/ChannelUtils.java

@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.proxy.common.utils;
+
+import io.netty.channel.Channel;
+
+public class ChannelUtils {
+    public static String buildChannelKey(Channel channel, String groupName) {
+        return channel.id().asLongText() + "%" + groupName;
+    }
+}

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

@@ -18,6 +18,7 @@
 package org.apache.rocketmq.proxy.config;
 
 import java.time.Duration;
+import java.util.concurrent.TimeUnit;
 import org.apache.rocketmq.proxy.ProxyMode;
 
 public class ProxyConfig {
@@ -97,6 +98,12 @@ public class ProxyConfig {
 
     private long longPollingReserveTimeInMillis = 100;
 
+    private long invisibleTimeMillisWhenClear = 1000L;
+    private boolean enableProxyAutoRenew = true;
+    private long renewAheadTimeMillis = TimeUnit.SECONDS.toMillis(10);
+    private long renewSliceTimeMillis = TimeUnit.SECONDS.toMillis(60);
+    private long renewSchedulePeriodMillis = TimeUnit.SECONDS.toMillis(5);
+
     private boolean enableACL = false;
 
     private boolean enableTopicMessageTypeCheck = true;
@@ -532,4 +539,44 @@ public class ProxyConfig {
     public void setEnableTopicMessageTypeCheck(boolean enableTopicMessageTypeCheck) {
         this.enableTopicMessageTypeCheck = enableTopicMessageTypeCheck;
     }
+
+    public long getInvisibleTimeMillisWhenClear() {
+        return invisibleTimeMillisWhenClear;
+    }
+
+    public void setInvisibleTimeMillisWhenClear(long invisibleTimeMillisWhenClear) {
+        this.invisibleTimeMillisWhenClear = invisibleTimeMillisWhenClear;
+    }
+
+    public boolean isEnableProxyAutoRenew() {
+        return enableProxyAutoRenew;
+    }
+
+    public void setEnableProxyAutoRenew(boolean enableProxyAutoRenew) {
+        this.enableProxyAutoRenew = enableProxyAutoRenew;
+    }
+
+    public long getRenewAheadTimeMillis() {
+        return renewAheadTimeMillis;
+    }
+
+    public void setRenewAheadTimeMillis(long renewAheadTimeMillis) {
+        this.renewAheadTimeMillis = renewAheadTimeMillis;
+    }
+
+    public long getRenewSliceTimeMillis() {
+        return renewSliceTimeMillis;
+    }
+
+    public void setRenewSliceTimeMillis(long renewSliceTimeMillis) {
+        this.renewSliceTimeMillis = renewSliceTimeMillis;
+    }
+
+    public long getRenewSchedulePeriodMillis() {
+        return renewSchedulePeriodMillis;
+    }
+
+    public void setRenewSchedulePeriodMillis(long renewSchedulePeriodMillis) {
+        this.renewSchedulePeriodMillis = renewSchedulePeriodMillis;
+    }
 }

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/AbstractMessingActivity.java

@@ -17,9 +17,12 @@
 package org.apache.rocketmq.proxy.grpc.v2;
 
 import io.grpc.Context;
+import io.netty.channel.Channel;
 import org.apache.rocketmq.proxy.common.ContextVariable;
 import org.apache.rocketmq.proxy.common.ProxyContext;
+import org.apache.rocketmq.proxy.common.utils.ChannelUtils;
 import org.apache.rocketmq.proxy.grpc.interceptor.InterceptorConstants;
+import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
 import org.apache.rocketmq.proxy.processor.MessagingProcessor;
 import org.apache.rocketmq.remoting.protocol.LanguageCode;
@@ -28,16 +31,26 @@ public abstract class AbstractMessingActivity {
 
     protected final MessagingProcessor messagingProcessor;
     protected final GrpcClientSettingsManager grpcClientSettingsManager;
+    protected final GrpcChannelManager grpcChannelManager;
 
     public AbstractMessingActivity(MessagingProcessor messagingProcessor,
-        GrpcClientSettingsManager grpcClientSettingsManager) {
+        GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) {
         this.messagingProcessor = messagingProcessor;
         this.grpcClientSettingsManager = grpcClientSettingsManager;
+        this.grpcChannelManager = grpcChannelManager;
     }
 
     protected ProxyContext createContext(Context ctx) {
         return ProxyContext.create()
+            .withVal(ContextVariable.LOCAL_ADDRESS, InterceptorConstants.METADATA.get(ctx).get(InterceptorConstants.LOCAL_ADDRESS))
+            .withVal(ContextVariable.REMOTE_ADDRESS, InterceptorConstants.METADATA.get(ctx).get(InterceptorConstants.REMOTE_ADDRESS))
             .withVal(ContextVariable.CLIENT_ID, InterceptorConstants.METADATA.get(ctx).get(InterceptorConstants.CLIENT_ID))
             .withVal(ContextVariable.LANGUAGE, LanguageCode.valueOf(InterceptorConstants.METADATA.get(ctx).get(InterceptorConstants.LANGUAGE)));
     }
+
+    protected void attachChannelId(Context ctx, ProxyContext context, String groupName) {
+        String clientId = context.getVal(ContextVariable.CLIENT_ID);
+        Channel channel = grpcChannelManager.createChannel(ctx, groupName, clientId);
+        context.withVal(ContextVariable.CHANNEL_KEY, ChannelUtils.buildChannelKey(channel, groupName));
+    }
 }

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/DefaultGrpcMessingActivity.java

@@ -44,6 +44,7 @@ import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.proxy.common.AbstractStartAndShutdown;
+import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager;
 import org.apache.rocketmq.proxy.grpc.v2.client.ClientActivity;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
 import org.apache.rocketmq.proxy.grpc.v2.consumer.AckMessageActivity;
@@ -54,12 +55,11 @@ import org.apache.rocketmq.proxy.grpc.v2.producer.SendMessageActivity;
 import org.apache.rocketmq.proxy.grpc.v2.route.RouteActivity;
 import org.apache.rocketmq.proxy.grpc.v2.transaction.EndTransactionActivity;
 import org.apache.rocketmq.proxy.processor.MessagingProcessor;
+import org.apache.rocketmq.proxy.processor.ReceiptHandleProcessor;
 
 public class DefaultGrpcMessingActivity extends AbstractStartAndShutdown implements GrpcMessingActivity {
     private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
 
-    private final GrpcClientSettingsManager grpcClientSettingsManager;
-
     private final ReceiveMessageActivity receiveMessageActivity;
     private final AckMessageActivity ackMessageActivity;
     private final ChangeInvisibleDurationActivity changeInvisibleDurationActivity;
@@ -70,16 +70,18 @@ public class DefaultGrpcMessingActivity extends AbstractStartAndShutdown impleme
     private final ClientActivity clientActivity;
 
     protected DefaultGrpcMessingActivity(MessagingProcessor messagingProcessor) {
-        this.grpcClientSettingsManager = new GrpcClientSettingsManager(messagingProcessor);
+        GrpcClientSettingsManager grpcClientSettingsManager = new GrpcClientSettingsManager(messagingProcessor);
+        GrpcChannelManager grpcChannelManager = new GrpcChannelManager(messagingProcessor.getProxyOutService());
+        ReceiptHandleProcessor receiptHandleProcessor = new ReceiptHandleProcessor(messagingProcessor);
 
-        this.receiveMessageActivity = new ReceiveMessageActivity(messagingProcessor, this.grpcClientSettingsManager);
-        this.ackMessageActivity = new AckMessageActivity(messagingProcessor, this.grpcClientSettingsManager);
-        this.changeInvisibleDurationActivity = new ChangeInvisibleDurationActivity(messagingProcessor, this.grpcClientSettingsManager);
-        this.sendMessageActivity = new SendMessageActivity(messagingProcessor, this.grpcClientSettingsManager);
-        this.forwardMessageToDLQActivity = new ForwardMessageToDLQActivity(messagingProcessor, this.grpcClientSettingsManager);
-        this.endTransactionActivity = new EndTransactionActivity(messagingProcessor, this.grpcClientSettingsManager);
-        this.routeActivity = new RouteActivity(messagingProcessor, this.grpcClientSettingsManager);
-        this.clientActivity = new ClientActivity(messagingProcessor, this.grpcClientSettingsManager);
+        this.receiveMessageActivity = new ReceiveMessageActivity(messagingProcessor, receiptHandleProcessor, grpcClientSettingsManager, grpcChannelManager);
+        this.ackMessageActivity = new AckMessageActivity(messagingProcessor, receiptHandleProcessor, grpcClientSettingsManager, grpcChannelManager);
+        this.changeInvisibleDurationActivity = new ChangeInvisibleDurationActivity(messagingProcessor, receiptHandleProcessor, grpcClientSettingsManager, grpcChannelManager);
+        this.sendMessageActivity = new SendMessageActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
+        this.forwardMessageToDLQActivity = new ForwardMessageToDLQActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
+        this.endTransactionActivity = new EndTransactionActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
+        this.routeActivity = new RouteActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
+        this.clientActivity = new ClientActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
     }
 
     @Override

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java

@@ -71,22 +71,10 @@ public class ClientActivity extends AbstractMessingActivity {
 
     private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
 
-    private final GrpcChannelManager grpcChannelManager;
-
     public ClientActivity(MessagingProcessor messagingProcessor,
-        GrpcClientSettingsManager grpcClientSettingsManager) {
-        super(messagingProcessor, grpcClientSettingsManager);
-
-        this.grpcChannelManager = new GrpcChannelManager(messagingProcessor.getProxyOutService());
-
-        this.init();
-    }
-
-    ClientActivity(MessagingProcessor messagingProcessor,
         GrpcClientSettingsManager grpcClientSettingsManager,
         GrpcChannelManager grpcChannelManager) {
-        super(messagingProcessor, grpcClientSettingsManager);
-        this.grpcChannelManager = grpcChannelManager;
+        super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
         this.init();
     }
 

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/AckMessageActivity.java

@@ -30,22 +30,29 @@ import java.util.concurrent.CompletableFuture;
 import org.apache.rocketmq.client.consumer.AckResult;
 import org.apache.rocketmq.client.consumer.AckStatus;
 import org.apache.rocketmq.common.consumer.ReceiptHandle;
+import org.apache.rocketmq.proxy.common.ContextVariable;
 import org.apache.rocketmq.proxy.common.ProxyContext;
 import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity;
+import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter;
 import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder;
 import org.apache.rocketmq.proxy.processor.MessagingProcessor;
+import org.apache.rocketmq.proxy.processor.ReceiptHandleProcessor;
 
 public class AckMessageActivity extends AbstractMessingActivity {
-
-    public AckMessageActivity(MessagingProcessor messagingProcessor,
-        GrpcClientSettingsManager grpcClientSettingsManager) {
-        super(messagingProcessor, grpcClientSettingsManager);
+    protected ReceiptHandleProcessor receiptHandleProcessor;
+    public AckMessageActivity(MessagingProcessor messagingProcessor, ReceiptHandleProcessor receiptHandleProcessor,
+        GrpcClientSettingsManager grpcClientSettingsManager,
+        GrpcChannelManager grpcChannelManager) {
+        super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
+        this.receiptHandleProcessor = receiptHandleProcessor;
     }
 
     public CompletableFuture<AckMessageResponse> ackMessage(Context ctx, AckMessageRequest request) {
         ProxyContext proxyContext = createContext(ctx);
+        String groupName = GrpcConverter.wrapResourceWithNamespace(request.getGroup());
+        attachChannelId(ctx, proxyContext, groupName);
         CompletableFuture<AckMessageResponse> future = new CompletableFuture<>();
 
         try {
@@ -101,12 +108,16 @@ public class AckMessageActivity extends AbstractMessingActivity {
                 ackMessageEntry.getMessageId(),
                 GrpcConverter.wrapResourceWithNamespace(request.getGroup()),
                 GrpcConverter.wrapResourceWithNamespace(request.getTopic()));
-            ackResultFuture
-                .thenAccept(result -> future.complete(convertToAckMessageResultEntry(ctx, ackMessageEntry, result)))
-                .exceptionally(throwable -> {
-                    future.complete(failResult.setStatus(ResponseBuilder.buildStatus(throwable)).build());
-                    return null;
-                });
+            ackResultFuture.thenAccept(result -> {
+                if (AckStatus.OK.equals(result.getStatus())) {
+                    String channelId = ctx.getVal(ContextVariable.CHANNEL_KEY);
+                    receiptHandleProcessor.removeReceiptHandle(channelId, ackMessageEntry.getReceiptHandle());
+                }
+                future.complete(convertToAckMessageResultEntry(ctx, ackMessageEntry, result));
+            }).exceptionally(throwable -> {
+                future.complete(failResult.setStatus(ResponseBuilder.buildStatus(throwable)).build());
+                return null;
+            });
         } catch (Throwable t) {
             future.complete(failResult.setStatus(ResponseBuilder.buildStatus(t)).build());
         }

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivity.java

@@ -25,23 +25,30 @@ import java.util.concurrent.CompletableFuture;
 import org.apache.rocketmq.client.consumer.AckResult;
 import org.apache.rocketmq.client.consumer.AckStatus;
 import org.apache.rocketmq.common.consumer.ReceiptHandle;
+import org.apache.rocketmq.proxy.common.ContextVariable;
 import org.apache.rocketmq.proxy.common.ProxyContext;
 import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity;
+import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter;
 import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder;
 import org.apache.rocketmq.proxy.processor.MessagingProcessor;
+import org.apache.rocketmq.proxy.processor.ReceiptHandleProcessor;
 
 public class ChangeInvisibleDurationActivity extends AbstractMessingActivity {
+    protected ReceiptHandleProcessor receiptHandleProcessor;
 
-    public ChangeInvisibleDurationActivity(MessagingProcessor messagingProcessor,
-        GrpcClientSettingsManager grpcClientSettingsManager) {
-        super(messagingProcessor, grpcClientSettingsManager);
+    public ChangeInvisibleDurationActivity(MessagingProcessor messagingProcessor, ReceiptHandleProcessor receiptHandleProcessor,
+        GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) {
+        super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
+        this.receiptHandleProcessor = receiptHandleProcessor;
     }
 
     public CompletableFuture<ChangeInvisibleDurationResponse> changeInvisibleDuration(Context ctx,
         ChangeInvisibleDurationRequest request) {
         ProxyContext context = createContext(ctx);
+        String groupName = GrpcConverter.wrapResourceWithNamespace(request.getGroup());
+        attachChannelId(ctx, context, groupName);
         CompletableFuture<ChangeInvisibleDurationResponse> future = new CompletableFuture<>();
 
         try {
@@ -54,7 +61,13 @@ public class ChangeInvisibleDurationActivity extends AbstractMessingActivity {
                 GrpcConverter.wrapResourceWithNamespace(request.getGroup()),
                 GrpcConverter.wrapResourceWithNamespace(request.getTopic()),
                 Durations.toMillis(request.getInvisibleDuration())
-            ).thenApply(ackResult -> convertToChangeInvisibleDurationResponse(context, request, ackResult));
+            ).thenApply(ackResult -> {
+                if (AckStatus.OK.equals(ackResult.getStatus())) {
+                    String channelId = context.getVal(ContextVariable.CHANNEL_KEY);
+                    receiptHandleProcessor.removeReceiptHandle(channelId, receiptHandle.getReceiptHandle());
+                }
+                return convertToChangeInvisibleDurationResponse(context, request, ackResult);
+            });
         } catch (Throwable t) {
             future.completeExceptionally(t);
         }

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java

@@ -25,33 +25,46 @@ import apache.rocketmq.v2.Subscription;
 import com.google.protobuf.util.Durations;
 import io.grpc.Context;
 import io.grpc.stub.StreamObserver;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.common.constant.ConsumeInitMode;
 import org.apache.rocketmq.common.filter.FilterAPI;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
 import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
+import org.apache.rocketmq.proxy.common.ContextVariable;
+import org.apache.rocketmq.proxy.common.MessageReceiptHandle;
 import org.apache.rocketmq.proxy.common.ProxyContext;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
+import org.apache.rocketmq.proxy.config.ProxyConfig;
 import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity;
+import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter;
 import org.apache.rocketmq.proxy.processor.MessagingProcessor;
 import org.apache.rocketmq.proxy.processor.QueueSelector;
+import org.apache.rocketmq.proxy.processor.ReceiptHandleProcessor;
 import org.apache.rocketmq.proxy.service.route.MessageQueueSelector;
 import org.apache.rocketmq.proxy.service.route.MessageQueueView;
 import org.apache.rocketmq.proxy.service.route.SelectableMessageQueue;
 
 public class ReceiveMessageActivity extends AbstractMessingActivity {
+    protected ReceiptHandleProcessor receiptHandleProcessor;
 
-    public ReceiveMessageActivity(MessagingProcessor messagingProcessor,
-        GrpcClientSettingsManager grpcClientSettingsManager) {
-        super(messagingProcessor, grpcClientSettingsManager);
+    public ReceiveMessageActivity(MessagingProcessor messagingProcessor, ReceiptHandleProcessor receiptHandleProcessor,
+        GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) {
+        super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
+        this.receiptHandleProcessor = receiptHandleProcessor;
     }
 
     public void receiveMessage(Context ctx, ReceiveMessageRequest request,
         StreamObserver<ReceiveMessageResponse> responseObserver) {
         ProxyContext proxyContext = createContext(ctx);
         ReceiveMessageResponseStreamWriter writer = createWriter(proxyContext, responseObserver);
+        String groupName = GrpcConverter.wrapResourceWithNamespace(request.getGroup());
+        attachChannelId(ctx, proxyContext, groupName);
 
         try {
             Settings settings = this.grpcClientSettingsManager.getClientSettings(proxyContext);
@@ -68,10 +81,11 @@ public class ReceiveMessageActivity extends AbstractMessingActivity {
                 return;
             }
 
-            long invisibleTime = Durations.toMillis(request.getInvisibleDuration());
-            if (request.getAutoRenew()) {
-                invisibleTime = Durations.toMillis(subscription.getLongPollingTimeout()
-                );
+            final long requestInvisibleTime = Durations.toMillis(request.getInvisibleDuration());
+            long actualInvisibleTime = requestInvisibleTime;
+            ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
+            if (proxyConfig.isEnableProxyAutoRenew() && request.getAutoRenew()) {
+                actualInvisibleTime = Math.min(actualInvisibleTime, proxyConfig.getRenewSliceTimeMillis());
             }
 
             String topic = GrpcConverter.wrapResourceWithNamespace(request.getMessageQueue().getTopic());
@@ -94,14 +108,30 @@ public class ReceiveMessageActivity extends AbstractMessingActivity {
                 group,
                 topic,
                 request.getBatchSize(),
-                invisibleTime,
+                actualInvisibleTime,
                 pollTime,
                 ConsumeInitMode.MAX,
                 subscriptionData,
                 fifo,
                 new PopMessageResultFilterImpl(grpcClientSettingsManager),
                 timeRemaining
-            ).thenAccept(popResult -> writer.writeAndComplete(proxyContext, request, popResult))
+            ).thenAccept(popResult -> {
+                if (proxyConfig.isEnableProxyAutoRenew() && request.getAutoRenew()) {
+                    List<MessageExt> messageExtList = popResult.getMsgFoundList();
+                    for (MessageExt messageExt : messageExtList) {
+                        String receiptHandle = messageExt.getProperty(MessageConst.PROPERTY_POP_CK);
+                        if (receiptHandle != null) {
+                            MessageQueue messageQueue = new MessageQueue(topic, messageExt.getBrokerName(), messageExt.getQueueId());
+                            MessageReceiptHandle messageReceiptHandle =
+                                new MessageReceiptHandle(group, messageQueue, receiptHandle, messageExt.getMsgId(),
+                                    messageExt.getQueueOffset(), messageExt.getReconsumeTimes(), requestInvisibleTime);
+                            String channelId = proxyContext.getVal(ContextVariable.CHANNEL_KEY);
+                            receiptHandleProcessor.addReceiptHandle(channelId, receiptHandle, messageReceiptHandle);
+                        }
+                    }
+                }
+                writer.writeAndComplete(proxyContext, request, popResult);
+            })
                 .exceptionally(t -> {
                     writer.writeAndComplete(proxyContext, request, t);
                     return null;

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/ForwardMessageToDLQActivity.java

@@ -23,6 +23,7 @@ import java.util.concurrent.CompletableFuture;
 import org.apache.rocketmq.common.consumer.ReceiptHandle;
 import org.apache.rocketmq.proxy.common.ProxyContext;
 import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity;
+import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter;
 import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder;
@@ -32,8 +33,8 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 public class ForwardMessageToDLQActivity extends AbstractMessingActivity {
 
     public ForwardMessageToDLQActivity(MessagingProcessor messagingProcessor,
-        GrpcClientSettingsManager grpcClientSettingsManager) {
-        super(messagingProcessor, grpcClientSettingsManager);
+        GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) {
+        super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
     }
 
     public CompletableFuture<ForwardMessageToDeadLetterQueueResponse> forwardMessageToDeadLetterQueue(Context ctx,

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java

@@ -44,6 +44,7 @@ import org.apache.rocketmq.common.message.MessageConst;
 import org.apache.rocketmq.common.sysflag.MessageSysFlag;
 import org.apache.rocketmq.proxy.common.ProxyContext;
 import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity;
+import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcProxyException;
@@ -56,8 +57,8 @@ import org.apache.rocketmq.proxy.service.route.SelectableMessageQueue;
 public class SendMessageActivity extends AbstractMessingActivity {
 
     public SendMessageActivity(MessagingProcessor messagingProcessor,
-        GrpcClientSettingsManager grpcClientSettingsManager) {
-        super(messagingProcessor, grpcClientSettingsManager);
+        GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) {
+        super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
     }
 
     public CompletableFuture<SendMessageResponse> sendMessage(Context ctx, SendMessageRequest request) {

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/route/RouteActivity.java

@@ -41,6 +41,7 @@ import org.apache.rocketmq.common.constant.PermName;
 import org.apache.rocketmq.common.protocol.route.QueueData;
 import org.apache.rocketmq.proxy.common.ProxyContext;
 import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity;
+import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter;
 import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder;
@@ -51,8 +52,8 @@ import org.apache.rocketmq.proxy.service.route.TopicRouteHelper;
 public class RouteActivity extends AbstractMessingActivity {
 
     public RouteActivity(MessagingProcessor messagingProcessor,
-        GrpcClientSettingsManager grpcClientSettingsManager) {
-        super(messagingProcessor, grpcClientSettingsManager);
+        GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) {
+        super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
     }
 
     public CompletableFuture<QueryRouteResponse> queryRoute(Context ctx, QueryRouteRequest request) {

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/transaction/EndTransactionActivity.java

@@ -25,6 +25,7 @@ import io.grpc.Context;
 import java.util.concurrent.CompletableFuture;
 import org.apache.rocketmq.proxy.common.ProxyContext;
 import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity;
+import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter;
 import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder;
@@ -35,8 +36,8 @@ import org.apache.rocketmq.proxy.service.transaction.TransactionId;
 public class EndTransactionActivity extends AbstractMessingActivity {
 
     public EndTransactionActivity(MessagingProcessor messagingProcessor,
-        GrpcClientSettingsManager grpcClientSettingsManager) {
-        super(messagingProcessor, grpcClientSettingsManager);
+        GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) {
+        super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
     }
 
     public CompletableFuture<EndTransactionResponse> endTransaction(Context ctx, EndTransactionRequest request) {

proxy/src/main/java/org/apache/rocketmq/proxy/processor/AbstractProcessor.java

@@ -20,11 +20,12 @@ import org.apache.rocketmq.common.attribute.TopicMessageType;
 import org.apache.rocketmq.common.consumer.ReceiptHandle;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.proxy.common.AbstractStartAndShutdown;
 import org.apache.rocketmq.proxy.common.ProxyException;
 import org.apache.rocketmq.proxy.common.ProxyExceptionCode;
 import org.apache.rocketmq.proxy.service.ServiceManager;
 
-public abstract class AbstractProcessor {
+public abstract class AbstractProcessor extends AbstractStartAndShutdown {
 
     protected MessagingProcessor messagingProcessor;
     protected ServiceManager serviceManager;

proxy/src/main/java/org/apache/rocketmq/proxy/processor/ConsumerProcessor.java

@@ -46,8 +46,8 @@ public class ConsumerProcessor extends AbstractProcessor {
 
     private final ExecutorService executor;
 
-    public ConsumerProcessor(MessagingProcessor messagingProcessor,
-        ServiceManager serviceManager, ExecutorService executor) {
+    public ConsumerProcessor(MessagingProcessor messagingProcessor, ServiceManager serviceManager,
+        ExecutorService executor) {
         super(messagingProcessor, serviceManager);
         this.executor = executor;
     }

proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java

@@ -45,6 +45,7 @@ import org.apache.rocketmq.proxy.config.ConfigurationManager;
 import org.apache.rocketmq.proxy.config.ProxyConfig;
 import org.apache.rocketmq.proxy.service.ServiceManager;
 import org.apache.rocketmq.proxy.service.ServiceManagerFactory;
+import org.apache.rocketmq.proxy.service.metadata.MetadataService;
 import org.apache.rocketmq.proxy.service.relay.ProxyRelayService;
 import org.apache.rocketmq.proxy.service.route.ProxyTopicRouteData;
 import org.apache.rocketmq.proxy.service.transaction.TransactionId;
@@ -226,4 +227,8 @@ public class DefaultMessagingProcessor extends AbstractStartAndShutdown implemen
         return this.serviceManager.getProxyRelayService();
     }
 
+    @Override
+    public MetadataService getMetadataService() {
+        return this.serviceManager.getMetadataService();
+    }
 }

proxy/src/main/java/org/apache/rocketmq/proxy/processor/MessagingProcessor.java

@@ -38,6 +38,7 @@ import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
 import org.apache.rocketmq.proxy.common.Address;
 import org.apache.rocketmq.proxy.common.ProxyContext;
 import org.apache.rocketmq.proxy.common.StartAndShutdown;
+import org.apache.rocketmq.proxy.service.metadata.MetadataService;
 import org.apache.rocketmq.proxy.service.relay.ProxyRelayService;
 import org.apache.rocketmq.proxy.service.route.ProxyTopicRouteData;
 import org.apache.rocketmq.proxy.service.transaction.TransactionId;
@@ -228,4 +229,6 @@ public interface MessagingProcessor extends StartAndShutdown {
     );
 
     ProxyRelayService getProxyOutService();
+
+    MetadataService getMetadataService();
 }

proxy/src/main/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessor.java

@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.proxy.processor;
+
+import io.netty.channel.Channel;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import org.apache.rocketmq.broker.client.ClientChannelInfo;
+import org.apache.rocketmq.broker.client.ConsumerGroupEvent;
+import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener;
+import org.apache.rocketmq.client.consumer.AckResult;
+import org.apache.rocketmq.client.consumer.AckStatus;
+import org.apache.rocketmq.common.ThreadFactoryImpl;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.consumer.ReceiptHandle;
+import org.apache.rocketmq.common.subscription.RetryPolicy;
+import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.common.thread.ThreadPoolMonitor;
+import org.apache.rocketmq.proxy.common.AbstractStartAndShutdown;
+import org.apache.rocketmq.proxy.common.MessageReceiptHandle;
+import org.apache.rocketmq.proxy.common.ProxyContext;
+import org.apache.rocketmq.proxy.common.ReceiptHandleGroup;
+import org.apache.rocketmq.proxy.common.StartAndShutdown;
+import org.apache.rocketmq.proxy.common.utils.ChannelUtils;
+import org.apache.rocketmq.proxy.config.ConfigurationManager;
+import org.apache.rocketmq.proxy.config.ProxyConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ReceiptHandleProcessor extends AbstractStartAndShutdown {
+    protected final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
+    protected final ConcurrentMap<String, ReceiptHandleGroup> receiptHandleGroupMap;
+    protected final ScheduledExecutorService scheduledExecutorService =
+        Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("RenewalScheduledThread_"));
+    protected final ExecutorService renewalWorkerService = ThreadPoolMonitor.createAndMonitor(
+        2, 4, 0, TimeUnit.MILLISECONDS,
+        "RenewalWorkerThread_", 10000
+    );
+    protected final MessagingProcessor messagingProcessor;
+
+    public ReceiptHandleProcessor(MessagingProcessor messagingProcessor) {
+        this.messagingProcessor = messagingProcessor;
+        this.messagingProcessor.registerConsumerListener(new ConsumerIdsChangeListener() {
+            @Override
+            public void handle(ConsumerGroupEvent event, String group, Object... args) {
+                if (ConsumerGroupEvent.CLIENT_UNREGISTER.equals(event)) {
+                    if (args == null || args.length < 1) {
+                        return;
+                    }
+                    if (args[0] instanceof ClientChannelInfo) {
+                        ClientChannelInfo clientChannelInfo = (ClientChannelInfo) args[0];
+                        Channel channel = clientChannelInfo.getChannel();
+                        clearGroup(ChannelUtils.buildChannelKey(channel, group));
+                    }
+                }
+            }
+
+            @Override
+            public void shutdown() {
+
+            }
+        });
+        this.receiptHandleGroupMap = new ConcurrentHashMap<>();
+        this.init();
+    }
+
+    protected void init() {
+        this.appendStartAndShutdown(new StartAndShutdown() {
+            @Override
+            public void start() throws Exception {
+                log.info("scan for renewal start.");
+                scheduledExecutorService.scheduleAtFixedRate(() -> scheduleRenewTask(), 0,
+                    ConfigurationManager.getProxyConfig().getRenewSchedulePeriodMillis(), TimeUnit.MILLISECONDS);
+                log.info("renewal queue has started");
+            }
+
+            @Override
+            public void shutdown() throws Exception {
+                scheduledExecutorService.shutdown();
+            }
+        });
+    }
+
+    protected void scheduleRenewTask() {
+        ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
+        for (Map.Entry<String, ReceiptHandleGroup> entry : receiptHandleGroupMap.entrySet()) {
+            String key = entry.getKey();
+            ReceiptHandleGroup group = entry.getValue();
+            group.all().forEach((k, v) -> {
+                ReceiptHandle handle = ReceiptHandle.decode(v.getReceiptHandle());
+                long now = System.currentTimeMillis();
+                if (handle.getNextVisibleTime() - now > proxyConfig.getRenewAheadTimeMillis()) {
+                    return;
+                }
+                SubscriptionGroupConfig subscriptionGroupConfig =
+                    messagingProcessor.getMetadataService().getSubscriptionGroupConfig(v.getGroup());
+                if (subscriptionGroupConfig == null) {
+                    log.error("Group's subscriptionGroupConfig is null, group = {}", v.getGroup());
+                    return;
+                }
+                RetryPolicy retryPolicy = subscriptionGroupConfig.getGroupRetryPolicy().getRetryPolicy();
+                renewalWorkerService.submit(() -> renewMessage(key, v, handle, retryPolicy));
+            });
+        }
+
+        log.info("scan for renewal done.");
+    }
+
+    protected void renewMessage(String key, MessageReceiptHandle messageReceiptHandle,
+        ReceiptHandle handle, RetryPolicy retryPolicy) {
+        ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
+        long current = System.currentTimeMillis();
+        if (current - messageReceiptHandle.getTimestamp() < messageReceiptHandle.getExpectInvisibleTime()) {
+            CompletableFuture<AckResult> future =
+                messagingProcessor.changeInvisibleTime(ProxyContext.create(), handle, messageReceiptHandle.getMessageId(),
+                    messageReceiptHandle.getGroup(), messageReceiptHandle.getMessageQueue().getTopic(), proxyConfig.getRenewSliceTimeMillis());
+            future.thenAccept(ackResult -> {
+                if (AckStatus.OK.equals(ackResult.getStatus())) {
+                    messageReceiptHandle.update(ackResult.getExtraInfo());
+                    addReceiptHandle(key, messageReceiptHandle.getOriginalReceiptHandle(), messageReceiptHandle);
+                }
+            });
+        } else {
+            CompletableFuture<AckResult> future = messagingProcessor.changeInvisibleTime(ProxyContext.create(),
+                handle, messageReceiptHandle.getMessageId(), messageReceiptHandle.getGroup(),
+                messageReceiptHandle.getMessageQueue().getTopic(),
+                retryPolicy.nextDelayDuration(messageReceiptHandle.getReconsumeTimes(), TimeUnit.MILLISECONDS));
+            future.thenAccept(ackResult -> {
+                if (AckStatus.OK.equals(ackResult.getStatus())) {
+                    removeReceiptHandle(key, messageReceiptHandle.getOriginalReceiptHandle());
+                }
+            });
+        }
+    }
+
+    public void addReceiptHandle(String key, String receiptHandle,
+        MessageReceiptHandle messageReceiptHandle) {
+        if (key == null) {
+            return;
+        }
+        receiptHandleGroupMap.computeIfAbsent(key,
+            k -> new ReceiptHandleGroup()).put(receiptHandle, messageReceiptHandle);
+    }
+
+    public void removeReceiptHandle(String key, String receiptHandle) {
+        if (key == null) {
+            return;
+        }
+        receiptHandleGroupMap.computeIfPresent(key, (k, v) -> {
+            v.remove(receiptHandle);
+            return v;
+        });
+    }
+
+    public void clearGroup(String key) {
+        if (key == null) {
+            return;
+        }
+        ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
+        receiptHandleGroupMap.computeIfPresent(key, (k, v) -> {
+                Map<String, MessageReceiptHandle> all = v.all();
+                all.forEach((key0, value0) -> {
+                    ReceiptHandle receiptHandle = ReceiptHandle.decode(value0.getReceiptHandle());
+                    messagingProcessor.changeInvisibleTime(
+                        ProxyContext.create(),
+                        receiptHandle,
+                        value0.getMessageId(),
+                        value0.getGroup(),
+                        value0.getMessageQueue().getTopic(),
+                        proxyConfig.getInvisibleTimeMillisWhenClear()
+                    );
+                });
+                return null;
+            }
+        );
+    }
+
+    protected String keyWithChannelId(String channelId, String groupName) {
+        return channelId + "%" + groupName;
+    }
+}

proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java

@@ -23,6 +23,7 @@ import io.netty.channel.Channel;
 import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelId;
 import io.netty.channel.ChannelMetadata;
 import io.netty.channel.ChannelOutboundBuffer;
 import io.netty.channel.DefaultChannelPromise;
@@ -57,7 +58,11 @@ public class SimpleChannel extends AbstractChannel {
      * @param localAddress  Local address
      */
     public SimpleChannel(Channel parent, String remoteAddress, String localAddress) {
-        super(parent);
+        this(parent, null, remoteAddress, localAddress);
+    }
+
+    public SimpleChannel(Channel parent, ChannelId id, String remoteAddress, String localAddress) {
+        super(parent, id);
         lastAccessTime = System.currentTimeMillis();
         this.remoteAddress = remoteAddress;
         this.localAddress = localAddress;

proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java

@@ -17,7 +17,6 @@
 
 package org.apache.rocketmq.proxy.service.relay;
 
-import io.netty.channel.AbstractChannel;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelFuture;
@@ -43,36 +42,29 @@ import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.proxy.common.ContextVariable;
 import org.apache.rocketmq.proxy.common.ProxyContext;
+import org.apache.rocketmq.proxy.service.channel.SimpleChannel;
 import org.apache.rocketmq.proxy.service.transaction.TransactionId;
 import org.apache.rocketmq.remoting.common.RemotingUtil;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 
-public abstract class ProxyChannel extends AbstractChannel {
+public abstract class ProxyChannel extends SimpleChannel {
     private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
-    protected final String remoteAddress;
     protected final SocketAddress remoteSocketAddress;
-    protected final String localAddress;
     protected final SocketAddress localSocketAddress;
 
     protected final ProxyRelayService proxyRelayService;
 
-    protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, String remoteAddress,
-        String localAddress) {
-        super(parent);
+    protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, String remoteAddress, String localAddress) {
+        super(parent, remoteAddress, localAddress);
         this.proxyRelayService = proxyRelayService;
-        this.remoteAddress = remoteAddress;
         this.remoteSocketAddress = RemotingUtil.string2SocketAddress(remoteAddress);
-        this.localAddress = localAddress;
         this.localSocketAddress = RemotingUtil.string2SocketAddress(localAddress);
     }
 
-    protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, ChannelId id, String remoteAddress,
-        String localAddress) {
-        super(parent, id);
+    protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, ChannelId id, String remoteAddress, String localAddress) {
+        super(parent, id, remoteAddress, localAddress);
         this.proxyRelayService = proxyRelayService;
-        this.remoteAddress = remoteAddress;
         this.remoteSocketAddress = RemotingUtil.string2SocketAddress(remoteAddress);
-        this.localAddress = localAddress;
         this.localSocketAddress = RemotingUtil.string2SocketAddress(localAddress);
     }
 

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/BaseActivityTest.java

@@ -26,8 +26,10 @@ import java.util.concurrent.TimeUnit;
 import org.apache.rocketmq.common.protocol.header.ExtraInfoUtil;
 import org.apache.rocketmq.proxy.config.InitConfigAndLoggerTest;
 import org.apache.rocketmq.proxy.grpc.interceptor.InterceptorConstants;
+import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager;
 import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
 import org.apache.rocketmq.proxy.processor.MessagingProcessor;
+import org.apache.rocketmq.proxy.processor.ReceiptHandleProcessor;
 import org.apache.rocketmq.proxy.service.relay.ProxyRelayService;
 import org.junit.Ignore;
 import org.junit.runner.RunWith;
@@ -42,7 +44,9 @@ public class BaseActivityTest extends InitConfigAndLoggerTest {
     protected static final Random RANDOM = new Random();
     protected MessagingProcessor messagingProcessor;
     protected GrpcClientSettingsManager grpcClientSettingsManager;
+    protected GrpcChannelManager grpcChannelManager;
     protected ProxyRelayService proxyRelayService;
+    protected ReceiptHandleProcessor receiptHandleProcessor;
 
     protected static final String REMOTE_ADDR = "192.168.0.1:8080";
     protected static final String LOCAL_ADDR = "127.0.0.1:8080";
@@ -55,12 +59,14 @@ public class BaseActivityTest extends InitConfigAndLoggerTest {
         messagingProcessor = mock(MessagingProcessor.class);
         grpcClientSettingsManager = mock(GrpcClientSettingsManager.class);
         proxyRelayService = mock(ProxyRelayService.class);
+        receiptHandleProcessor = mock(ReceiptHandleProcessor.class);
 
         metadata.put(InterceptorConstants.CLIENT_ID, CLIENT_ID);
         metadata.put(InterceptorConstants.LANGUAGE, "JAVA");
         metadata.put(InterceptorConstants.REMOTE_ADDRESS, REMOTE_ADDR);
         metadata.put(InterceptorConstants.LOCAL_ADDRESS, LOCAL_ADDR);
         when(messagingProcessor.getProxyOutService()).thenReturn(proxyRelayService);
+        grpcChannelManager = new GrpcChannelManager(messagingProcessor.getProxyOutService());
     }
 
     protected Context createContext() {

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java

@@ -89,7 +89,7 @@ public class ClientActivityTest extends BaseActivityTest {
     @Before
     public void before() throws Throwable {
         super.before();
-        this.clientActivity = new ClientActivity(this.messagingProcessor, this.grpcClientSettingsManager);
+        this.clientActivity = new ClientActivity(this.messagingProcessor, this.grpcClientSettingsManager, grpcChannelManager);
     }
 
     protected TelemetryCommand sendProducerTelemetry(Context context) throws Throwable {

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/AckMessageActivityTest.java

@@ -45,7 +45,7 @@ public class AckMessageActivityTest extends BaseActivityTest {
     @Before
     public void before() throws Throwable {
         super.before();
-        this.ackMessageActivity = new AckMessageActivity(this.messagingProcessor, this.grpcClientSettingsManager);
+        this.ackMessageActivity = new AckMessageActivity(messagingProcessor, receiptHandleProcessor, grpcClientSettingsManager, grpcChannelManager);
     }
 
     @Test

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivityTest.java

@@ -45,7 +45,8 @@ public class ChangeInvisibleDurationActivityTest extends BaseActivityTest {
     @Before
     public void before() throws Throwable {
         super.before();
-        this.changeInvisibleDurationActivity = new ChangeInvisibleDurationActivity(this.messagingProcessor, this.grpcClientSettingsManager);
+        this.changeInvisibleDurationActivity = new ChangeInvisibleDurationActivity(messagingProcessor, receiptHandleProcessor,
+            grpcClientSettingsManager, grpcChannelManager);
     }
 
     @Test

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java

@@ -69,7 +69,8 @@ public class ReceiveMessageActivityTest extends BaseActivityTest {
     @Before
     public void before() throws Throwable {
         super.before();
-        this.receiveMessageActivity = new ReceiveMessageActivity(this.messagingProcessor, this.grpcClientSettingsManager);
+        this.receiveMessageActivity = new ReceiveMessageActivity(messagingProcessor, receiptHandleProcessor,
+            grpcClientSettingsManager, grpcChannelManager);
     }
 
     @Test
@@ -117,8 +118,7 @@ public class ReceiveMessageActivityTest extends BaseActivityTest {
             any(),
             anyBoolean(),
             any(),
-            anyLong()
-        )).thenReturn(CompletableFuture.completedFuture(popResult));
+            anyLong())).thenReturn(CompletableFuture.completedFuture(popResult));
 
         this.receiveMessageActivity.receiveMessage(
             createContext(),

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/ForwardMessageToDLQActivityTest.java

@@ -41,7 +41,7 @@ public class ForwardMessageToDLQActivityTest extends BaseActivityTest {
     @Before
     public void before() throws Throwable {
         super.before();
-        this.forwardMessageToDLQActivity = new ForwardMessageToDLQActivity(this.messagingProcessor, this.grpcClientSettingsManager);
+        this.forwardMessageToDLQActivity = new ForwardMessageToDLQActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
     }
 
     @Test

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java

@@ -71,7 +71,7 @@ public class SendMessageActivityTest extends BaseActivityTest {
     @Before
     public void before() throws Throwable {
         super.before();
-        this.sendMessageActivity = new SendMessageActivity(this.messagingProcessor, this.grpcClientSettingsManager);
+        this.sendMessageActivity = new SendMessageActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
     }
 
     @Test

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/route/RouteActivityTest.java

@@ -77,7 +77,7 @@ public class RouteActivityTest extends BaseActivityTest {
     @Before
     public void before() throws Throwable {
         super.before();
-        this.routeActivity = new RouteActivity(this.messagingProcessor, this.grpcClientSettingsManager);
+        this.routeActivity = new RouteActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
     }
 
     @Test

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/transaction/EndTransactionActivityTest.java

@@ -60,7 +60,7 @@ public class EndTransactionActivityTest extends BaseActivityTest {
     @Before
     public void before() throws Throwable {
         super.before();
-        this.endTransactionActivity = new EndTransactionActivity(this.messagingProcessor, this.grpcClientSettingsManager);
+        this.endTransactionActivity = new EndTransactionActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
     }
 
     @Test

proxy/src/test/java/org/apache/rocketmq/proxy/processor/BaseProcessorTest.java

@@ -83,6 +83,7 @@ public class BaseProcessorTest extends InitConfigAndLoggerTest {
         when(serviceManager.getTransactionService()).thenReturn(transactionService);
         when(serviceManager.getProxyRelayService()).thenReturn(proxyRelayService);
         when(serviceManager.getMetadataService()).thenReturn(metadataService);
+        when(messagingProcessor.getMetadataService()).thenReturn(metadataService);
     }
 
     protected static ProxyContext createContext() {

proxy/src/test/java/org/apache/rocketmq/proxy/processor/ConsumerProcessorTest.java

@@ -64,7 +64,8 @@ public class ConsumerProcessorTest extends BaseProcessorTest {
     @Before
     public void before() throws Throwable {
         super.before();
-        this.consumerProcessor = new ConsumerProcessor(this.messagingProcessor, this.serviceManager, Executors.newCachedThreadPool());
+        ReceiptHandleProcessor receiptHandleProcessor = new ReceiptHandleProcessor(messagingProcessor);
+        this.consumerProcessor = new ConsumerProcessor(messagingProcessor, serviceManager, Executors.newCachedThreadPool());
     }
 
     @Test

proxy/src/test/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessorTest.java

@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.proxy.processor;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener;
+import org.apache.rocketmq.client.consumer.AckResult;
+import org.apache.rocketmq.client.consumer.AckStatus;
+import org.apache.rocketmq.common.consumer.ReceiptHandle;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.proxy.common.ContextVariable;
+import org.apache.rocketmq.proxy.common.MessageReceiptHandle;
+import org.apache.rocketmq.proxy.common.ProxyContext;
+import org.apache.rocketmq.proxy.config.ConfigurationManager;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class ReceiptHandleProcessorTest extends BaseProcessorTest {
+    ReceiptHandleProcessor receiptHandleProcessor;
+
+    ProxyContext context = ProxyContext.create();
+    String group = "group";
+    MessageQueue messageQueue = new MessageQueue("topic", "broker", 1);
+    String messageId = "messageId";
+    long offset = 123L;
+    long invisibleTime = 100000L;
+    int reconsumeTimes = 1;
+    MessageReceiptHandle messageReceiptHandle;
+
+    String receiptHandle = ReceiptHandle.builder()
+        .startOffset(0L)
+        .retrieveTime(0)
+        .invisibleTime(invisibleTime)
+        .reviveQueueId(1)
+        .topicType(ReceiptHandle.NORMAL_TOPIC)
+        .brokerName(messageQueue.getBrokerName())
+        .queueId(messageQueue.getQueueId())
+        .offset(offset)
+        .commitLogOffset(0L)
+        .build().encode();
+
+    @Before
+    public void setup() {
+        context.withVal(ContextVariable.CHANNEL_KEY, "channel-id");
+        receiptHandleProcessor = new ReceiptHandleProcessor(messagingProcessor);
+        Mockito.doNothing().when(messagingProcessor).registerConsumerListener(Mockito.any(ConsumerIdsChangeListener.class));
+        messageReceiptHandle = new MessageReceiptHandle(group, messageQueue, receiptHandle, messageId, offset,
+            reconsumeTimes, invisibleTime);
+    }
+
+    @Test
+    public void testAddReceiptHandle() {
+        String channelId = context.getVal(ContextVariable.CHANNEL_KEY);
+        receiptHandleProcessor.addReceiptHandle(channelId, receiptHandle, messageReceiptHandle);
+        Mockito.when(metadataService.getSubscriptionGroupConfig(Mockito.eq(group))).thenReturn(new SubscriptionGroupConfig());
+        receiptHandleProcessor.scheduleRenewTask();
+        Mockito.verify(messagingProcessor, Mockito.timeout(1000).times(1))
+            .changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.any(ReceiptHandle.class), Mockito.eq(messageId),
+                Mockito.eq(group), Mockito.eq(messageQueue.getTopic()), Mockito.eq(ConfigurationManager.getProxyConfig().getRenewSliceTimeMillis()));
+    }
+
+    @Test
+    public void testRenewReceiptHandle() {
+        String channelId = context.getVal(ContextVariable.CHANNEL_KEY);
+        receiptHandleProcessor.addReceiptHandle(channelId, receiptHandle, messageReceiptHandle);
+        SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+        Mockito.when(metadataService.getSubscriptionGroupConfig(Mockito.eq(group))).thenReturn(groupConfig);
+        long newInvisibleTime = 2000L;
+        ReceiptHandle newReceiptHandleClass = ReceiptHandle.builder()
+            .startOffset(0L)
+            .retrieveTime(0)
+            .invisibleTime(newInvisibleTime)
+            .reviveQueueId(1)
+            .topicType(ReceiptHandle.NORMAL_TOPIC)
+            .brokerName(messageQueue.getBrokerName())
+            .queueId(messageQueue.getQueueId())
+            .offset(offset)
+            .commitLogOffset(0L)
+            .build();
+        String newReceiptHandle = newReceiptHandleClass.encode();
+        AckResult ackResult = new AckResult();
+        ackResult.setStatus(AckStatus.OK);
+        ackResult.setExtraInfo(newReceiptHandle);
+        Mockito.when(messagingProcessor.changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.any(ReceiptHandle.class), Mockito.eq(messageId),
+            Mockito.eq(group), Mockito.eq(messageQueue.getTopic()), Mockito.eq(ConfigurationManager.getProxyConfig().getRenewSliceTimeMillis())))
+            .thenReturn(CompletableFuture.completedFuture(ackResult));
+        receiptHandleProcessor.scheduleRenewTask();
+        Mockito.verify(messagingProcessor, Mockito.timeout(1000).times(1))
+            .changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.argThat((r) -> r.getInvisibleTime() == invisibleTime), Mockito.eq(messageId),
+                Mockito.eq(group), Mockito.eq(messageQueue.getTopic()), Mockito.eq(ConfigurationManager.getProxyConfig().getRenewSliceTimeMillis()));
+        receiptHandleProcessor.scheduleRenewTask();
+        Mockito.verify(messagingProcessor, Mockito.timeout(1000).times(1))
+            .changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.argThat((r) -> r.getInvisibleTime() == newInvisibleTime), Mockito.eq(messageId),
+                Mockito.eq(group), Mockito.eq(messageQueue.getTopic()), Mockito.eq(ConfigurationManager.getProxyConfig().getRenewSliceTimeMillis()));
+    }
+
+    @Test
+    public void testRenewReceiptHandleWhenTimeout() {
+        long newInvisibleTime = 0L;
+        String newReceiptHandle = ReceiptHandle.builder()
+            .startOffset(0L)
+            .retrieveTime(0)
+            .invisibleTime(newInvisibleTime)
+            .reviveQueueId(1)
+            .topicType(ReceiptHandle.NORMAL_TOPIC)
+            .brokerName(messageQueue.getBrokerName())
+            .queueId(messageQueue.getQueueId())
+            .offset(offset)
+            .commitLogOffset(0L)
+            .build().encode();
+        messageReceiptHandle = new MessageReceiptHandle(group, messageQueue, receiptHandle, messageId, offset,
+            reconsumeTimes, newInvisibleTime);
+        String channelId = context.getVal(ContextVariable.CHANNEL_KEY);
+        receiptHandleProcessor.addReceiptHandle(channelId, newReceiptHandle, messageReceiptHandle);
+        SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+        Mockito.when(metadataService.getSubscriptionGroupConfig(Mockito.eq(group))).thenReturn(groupConfig);
+        receiptHandleProcessor.scheduleRenewTask();
+        Mockito.verify(messagingProcessor, Mockito.timeout(1000).times(1))
+            .changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.any(ReceiptHandle.class), Mockito.eq(messageId),
+                Mockito.eq(group), Mockito.eq(messageQueue.getTopic()), Mockito.eq(groupConfig.getGroupRetryPolicy().getRetryPolicy().nextDelayDuration(reconsumeTimes, TimeUnit.MILLISECONDS)));
+    }
+
+
+    @Test
+    public void testRenewReceiptHandleWhenNotArrivingTime() {
+        String newReceiptHandle = ReceiptHandle.builder()
+            .startOffset(0L)
+            .retrieveTime(System.currentTimeMillis())
+            .invisibleTime(invisibleTime)
+            .reviveQueueId(1)
+            .topicType(ReceiptHandle.NORMAL_TOPIC)
+            .brokerName(messageQueue.getBrokerName())
+            .queueId(messageQueue.getQueueId())
+            .offset(offset)
+            .commitLogOffset(0L)
+            .build().encode();
+        messageReceiptHandle = new MessageReceiptHandle(group, messageQueue, newReceiptHandle, messageId, offset,
+            reconsumeTimes, invisibleTime);
+        String channelId = context.getVal(ContextVariable.CHANNEL_KEY);
+        receiptHandleProcessor.addReceiptHandle(channelId, newReceiptHandle, messageReceiptHandle);
+        SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+        Mockito.when(metadataService.getSubscriptionGroupConfig(Mockito.eq(group))).thenReturn(groupConfig);
+        receiptHandleProcessor.scheduleRenewTask();
+        Mockito.verify(messagingProcessor, Mockito.timeout(1000).times(0))
+            .changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.any(ReceiptHandle.class), Mockito.anyString(),
+                Mockito.anyString(), Mockito.anyString(), Mockito.anyLong());
+    }
+
+    @Test
+    public void testRemoveReceiptHandle() {
+        String channelId = context.getVal(ContextVariable.CHANNEL_KEY);
+        receiptHandleProcessor.addReceiptHandle(channelId, receiptHandle, messageReceiptHandle);
+        receiptHandleProcessor.removeReceiptHandle(channelId, receiptHandle);
+        SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+        Mockito.when(metadataService.getSubscriptionGroupConfig(Mockito.eq(group))).thenReturn(groupConfig);
+        receiptHandleProcessor.scheduleRenewTask();
+        Mockito.verify(messagingProcessor, Mockito.timeout(1000).times(0))
+            .changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.any(ReceiptHandle.class), Mockito.anyString(),
+                Mockito.anyString(), Mockito.anyString(), Mockito.anyLong());
+    }
+
+    @Test
+    public void testClearGroup() {
+        String channelId = context.getVal(ContextVariable.CHANNEL_KEY);
+        receiptHandleProcessor.addReceiptHandle(channelId, receiptHandle, messageReceiptHandle);
+        receiptHandleProcessor.clearGroup(channelId);
+        SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+        Mockito.when(metadataService.getSubscriptionGroupConfig(Mockito.eq(group))).thenReturn(groupConfig);
+        receiptHandleProcessor.scheduleRenewTask();
+        Mockito.verify(messagingProcessor, Mockito.timeout(1000).times(1))
+            .changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.any(ReceiptHandle.class), Mockito.eq(messageId),
+                Mockito.eq(group), Mockito.eq(messageQueue.getTopic()), Mockito.eq(ConfigurationManager.getProxyConfig().getInvisibleTimeMillisWhenClear()));
+    }
+}