[ISSUE #3949] add test cases

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

@@ -129,7 +129,10 @@ public class ClientActivity extends AbstractMessingActivity {
                     break;
                 }
                 default: {
-                    throw new IllegalArgumentException("ClientType not exist " + clientSettings.getClientType());
+                    future.complete(HeartbeatResponse.newBuilder()
+                        .setStatus(ResponseBuilder.buildStatus(Code.UNRECOGNIZED_CLIENT_TYPE, clientSettings.getClientType().name()))
+                        .build());
+                    return future;
                 }
             }
             future.complete(HeartbeatResponse.newBuilder()
@@ -150,7 +153,7 @@ public class ClientActivity extends AbstractMessingActivity {
             ProxyContext context = createContext(ctx);
             String clientId = context.getVal(GrpcContextConstants.CLIENT_ID);
             LanguageCode languageCode = context.getVal(GrpcContextConstants.LANGUAGE);
-            Settings clientSettings = grpcClientSettingsManager.getClientSettings(context);
+            Settings clientSettings = grpcClientSettingsManager.removeClientSettings(clientId);
 
             switch (clientSettings.getClientType()) {
                 case PRODUCER:
@@ -158,8 +161,8 @@ public class ClientActivity extends AbstractMessingActivity {
                         String topicName = GrpcConverter.wrapResourceWithNamespace(topic);
                         // user topic name as producer group
                         GrpcClientChannel channel = this.grpcChannelManager.removeChannel(topicName, clientId);
-                        ClientChannelInfo clientChannelInfo = new ClientChannelInfo(channel, clientId, languageCode, MQVersion.Version.V5_0_0.ordinal());
                         if (channel != null) {
+                            ClientChannelInfo clientChannelInfo = new ClientChannelInfo(channel, clientId, languageCode, MQVersion.Version.V5_0_0.ordinal());
                             this.messagingProcessor.unRegisterProducer(context, topicName, clientChannelInfo);
                         }
                     }
@@ -171,13 +174,16 @@ public class ClientActivity extends AbstractMessingActivity {
                     }
                     String consumerGroup = GrpcConverter.wrapResourceWithNamespace(request.getGroup());
                     GrpcClientChannel channel = this.grpcChannelManager.removeChannel(consumerGroup, clientId);
-                    ClientChannelInfo clientChannelInfo = new ClientChannelInfo(channel, clientId, languageCode, MQVersion.Version.V5_0_0.ordinal());
                     if (channel != null) {
+                        ClientChannelInfo clientChannelInfo = new ClientChannelInfo(channel, clientId, languageCode, MQVersion.Version.V5_0_0.ordinal());
                         this.messagingProcessor.unRegisterConsumer(context, consumerGroup, clientChannelInfo);
                     }
                     break;
                 default:
-                    break;
+                    future.complete(NotifyClientTerminationResponse.newBuilder()
+                        .setStatus(ResponseBuilder.buildStatus(Code.UNRECOGNIZED_CLIENT_TYPE, clientSettings.getClientType().name()))
+                        .build());
+                    return future;
             }
             future.complete(NotifyClientTerminationResponse.newBuilder()
                 .setStatus(ResponseBuilder.buildStatus(Code.OK, Code.OK.name()))

proxy/src/main/java/org/apache/rocketmq/proxy/service/ClusterServiceManager.java

@@ -37,6 +37,7 @@ import org.apache.rocketmq.proxy.service.message.MessageService;
 import org.apache.rocketmq.proxy.service.metadata.ClusterMetadataService;
 import org.apache.rocketmq.proxy.service.metadata.MetadataService;
 import org.apache.rocketmq.proxy.service.mqclient.DoNothingClientRemotingProcessor;
+import org.apache.rocketmq.proxy.service.mqclient.ProxyClientRemotingProcessor;
 import org.apache.rocketmq.proxy.service.relay.ClusterProxyRelayService;
 import org.apache.rocketmq.proxy.service.relay.ProxyRelayService;
 import org.apache.rocketmq.proxy.service.mqclient.MQClientAPIFactory;
@@ -60,6 +61,7 @@ public class ClusterServiceManager extends AbstractStartAndShutdown implements S
     private final ScheduledExecutorService scheduledExecutorService;
     private final MQClientAPIFactory messagingClientAPIFactory;
     private final MQClientAPIFactory operationClientAPIFactory;
+    private final MQClientAPIFactory transactionClientAPIFactory;
 
     public ClusterServiceManager(RPCHook rpcHook) {
         this.scheduledExecutorService = Executors.newScheduledThreadPool(3);
@@ -68,7 +70,7 @@ public class ClusterServiceManager extends AbstractStartAndShutdown implements S
 
         ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
         this.messagingClientAPIFactory = new MQClientAPIFactory(
-            "CLUSTER_MQ_CLIENT_",
+            "ClusterMQClient_",
             proxyConfig.getRocketmqMQClientNum(),
             new DoNothingClientRemotingProcessor(null),
             rpcHook,
@@ -80,10 +82,17 @@ public class ClusterServiceManager extends AbstractStartAndShutdown implements S
             rpcHook,
             this.scheduledExecutorService
         );
+        this.transactionClientAPIFactory = new MQClientAPIFactory(
+            "ClusterTransaction_",
+            1,
+            new ProxyClientRemotingProcessor(producerManager),
+            rpcHook,
+            scheduledExecutorService);
 
         this.topicRouteService = new ClusterTopicRouteService(operationClientAPIFactory);
         this.messageService = new ClusterMessageService(this.topicRouteService, this.messagingClientAPIFactory);
-        this.clusterTransactionService = new ClusterTransactionService(this.topicRouteService, this.producerManager, rpcHook);
+        this.clusterTransactionService = new ClusterTransactionService(this.topicRouteService, this.producerManager, rpcHook,
+            this.transactionClientAPIFactory);
         this.proxyRelayService = new ClusterProxyRelayService();
         this.metadataService = new ClusterMetadataService(topicRouteService, operationClientAPIFactory);
 
@@ -105,6 +114,7 @@ public class ClusterServiceManager extends AbstractStartAndShutdown implements S
         this.appendShutdown(scheduledExecutorService::shutdown);
         this.appendStartAndShutdown(this.messagingClientAPIFactory);
         this.appendStartAndShutdown(this.operationClientAPIFactory);
+        this.appendStartAndShutdown(this.transactionClientAPIFactory);
         this.appendStartAndShutdown(this.topicRouteService);
         this.appendStartAndShutdown(this.clusterTransactionService);
         this.appendStartAndShutdown(this.metadataService);

proxy/src/main/java/org/apache/rocketmq/proxy/service/metadata/ClusterMetadataService.java

@@ -22,7 +22,6 @@ import com.google.common.cache.LoadingCache;
 import java.util.Optional;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
-import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.attribute.TopicMessageType;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.protocol.route.BrokerData;
@@ -43,15 +42,16 @@ public class ClusterMetadataService extends AbstractStartAndShutdown implements
     protected static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
     private static final long DEFAULT_TIMEOUT = 3000;
 
-    private final ThreadPoolExecutor cacheRefreshExecutor;
     private final TopicRouteService topicRouteService;
     private final MQClientAPIFactory mqClientAPIFactory;
 
-    private final LoadingCache<String, TopicConfigAndQueueMapping> topicCache;
-    private final static TopicConfigAndQueueMapping EMPTY_TOPIC_CONFIG = new TopicConfigAndQueueMapping();
+    protected final ThreadPoolExecutor cacheRefreshExecutor;
 
-    private final LoadingCache<String, SubscriptionGroupConfig> subscriptionGroupConfigCache;
-    private final static SubscriptionGroupConfig EMPTY_SUBSCRIPTION_GROUP_CONFIG = new SubscriptionGroupConfig();
+    protected final LoadingCache<String, TopicConfigAndQueueMapping> topicConfigCache;
+    protected final static TopicConfigAndQueueMapping EMPTY_TOPIC_CONFIG = new TopicConfigAndQueueMapping();
+
+    protected final LoadingCache<String, SubscriptionGroupConfig> subscriptionGroupConfigCache;
+    protected final static SubscriptionGroupConfig EMPTY_SUBSCRIPTION_GROUP_CONFIG = new SubscriptionGroupConfig();
 
     public ClusterMetadataService(TopicRouteService topicRouteService, MQClientAPIFactory mqClientAPIFactory) {
         this.topicRouteService = topicRouteService;
@@ -66,7 +66,7 @@ public class ClusterMetadataService extends AbstractStartAndShutdown implements
             "MetadataCacheRefresh",
             config.getMetadataThreadPoolQueueCapacity()
         );
-        this.topicCache = CacheBuilder.newBuilder()
+        this.topicConfigCache = CacheBuilder.newBuilder()
             .maximumSize(config.getTopicConfigCacheMaxNum())
             .refreshAfterWrite(config.getTopicConfigCacheExpiredInSeconds(), TimeUnit.SECONDS)
             .build(new ClusterTopicConfigCacheLoader());
@@ -86,7 +86,7 @@ public class ClusterMetadataService extends AbstractStartAndShutdown implements
     public TopicMessageType getTopicMessageType(String topic) {
         TopicConfigAndQueueMapping topicConfigAndQueueMapping;
         try {
-            topicConfigAndQueueMapping = topicCache.get(topic);
+            topicConfigAndQueueMapping = topicConfigCache.get(topic);
         } catch (Exception e) {
             return TopicMessageType.UNSPECIFIED;
         }
@@ -159,7 +159,7 @@ public class ClusterMetadataService extends AbstractStartAndShutdown implements
     protected Optional<BrokerData> findOneBroker(String topic) throws Exception {
         try {
             return topicRouteService.getAllMessageQueueView(topic).getTopicRouteData().getBrokerDatas().stream().findAny();
-        } catch (MQClientException e) {
+        } catch (Exception e) {
             if (TopicRouteHelper.isTopicNotExistError(e)) {
                 return Optional.empty();
             }

proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java

@@ -152,16 +152,6 @@ public class MessageQueueSelector {
         return selectOneByIndex(nextIndex, onlyBroker);
     }
 
-    public final SelectableMessageQueue selectOne(String brokerName, int queueId) {
-        for (SelectableMessageQueue targetMessageQueue : queues) {
-            String queueBrokerName = targetMessageQueue.getBrokerName();
-            if (queueBrokerName.equals(brokerName) && targetMessageQueue.getQueueId() == queueId) {
-                return targetMessageQueue;
-            }
-        }
-        return null;
-    }
-
     public final SelectableMessageQueue selectOneByIndex(int index, boolean onlyBroker) {
         if (onlyBroker) {
             if (brokerActingQueues.isEmpty()) {
@@ -176,22 +166,6 @@ public class MessageQueueSelector {
         return queues.get(Math.abs(index) % queues.size());
     }
 
-    // find next same type(but different) queue with last(normal queue or broker acting queue).
-    public final SelectableMessageQueue selectNextQueue(SelectableMessageQueue last) {
-        boolean onlyBroker = last.getQueueId() < 0;
-        SelectableMessageQueue newOne = last;
-        int count = onlyBroker ? brokerActingQueues.size() : queues.size();
-
-        for (int i = 0; i < count; i++) {
-            newOne = selectOne(onlyBroker);
-            if (!newOne.getBrokerName().equals(last.getBrokerName()) || newOne.getQueueId() != last.getQueueId()) {
-                break;
-            }
-        }
-
-        return newOne;
-    }
-
     public List<SelectableMessageQueue> getQueues() {
         return queues;
     }

proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteService.java

@@ -42,11 +42,11 @@ import org.apache.rocketmq.proxy.service.mqclient.MQClientAPIFactory;
 public abstract class TopicRouteService extends AbstractStartAndShutdown {
     private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
 
-    private final LoadingCache<String /* topicName */, MessageQueueView> topicCache;
-
     private final MQClientAPIFactory mqClientAPIFactory;
-    private final ScheduledExecutorService scheduledExecutorService;
-    private final ThreadPoolExecutor cacheRefreshExecutor;
+
+    protected final LoadingCache<String /* topicName */, MessageQueueView> topicCache;
+    protected final ScheduledExecutorService scheduledExecutorService;
+    protected final ThreadPoolExecutor cacheRefreshExecutor;
 
     public TopicRouteService(MQClientAPIFactory mqClientAPIFactory) {
         ProxyConfig config = ConfigurationManager.getProxyConfig();

proxy/src/main/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionService.java

@@ -26,13 +26,10 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import org.apache.rocketmq.broker.client.ProducerManager;
 import org.apache.rocketmq.common.ServiceThread;
-import org.apache.rocketmq.common.ThreadFactoryImpl;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData;
 import org.apache.rocketmq.common.protocol.heartbeat.ProducerData;
@@ -44,7 +41,6 @@ import org.apache.rocketmq.proxy.common.StartAndShutdown;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
 import org.apache.rocketmq.proxy.config.ProxyConfig;
 import org.apache.rocketmq.proxy.service.mqclient.MQClientAPIFactory;
-import org.apache.rocketmq.proxy.service.mqclient.ProxyClientRemotingProcessor;
 import org.apache.rocketmq.proxy.service.route.MessageQueueView;
 import org.apache.rocketmq.proxy.service.route.TopicRouteService;
 import org.apache.rocketmq.remoting.RPCHook;
@@ -57,21 +53,14 @@ public class ClusterTransactionService implements StartAndShutdown, TransactionS
     private final MQClientAPIFactory mqClientAPIFactory;
     private final TopicRouteService topicRouteService;
 
-    private final ScheduledExecutorService scheduledExecutorService;
     private ThreadPoolExecutor heartbeatExecutors;
     private final Map<String /* group */, Set<ClusterData>/* cluster list */> groupClusterData = new ConcurrentHashMap<>();
     private TxHeartbeatServiceThread txHeartbeatServiceThread;
 
-    public ClusterTransactionService(TopicRouteService topicRouteService, ProducerManager producerManager, RPCHook rpcHook) {
-        this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(
-            new ThreadFactoryImpl("ClusterTransactionScheduledThread_"));
+    public ClusterTransactionService(TopicRouteService topicRouteService, ProducerManager producerManager, RPCHook rpcHook,
+        MQClientAPIFactory mqClientAPIFactory) {
         this.topicRouteService = topicRouteService;
-        this.mqClientAPIFactory = new MQClientAPIFactory(
-            "ClusterTransaction_",
-            1,
-            new ProxyClientRemotingProcessor(producerManager),
-            rpcHook,
-            scheduledExecutorService);
+        this.mqClientAPIFactory = mqClientAPIFactory;
     }
 
     @Override
@@ -182,6 +171,10 @@ public class ClusterTransactionService implements StartAndShutdown, TransactionS
         }
     }
 
+    public Map<String, Set<ClusterData>> getGroupClusterData() {
+        return groupClusterData;
+    }
+
     protected void sendHeartBeatToCluster(String clusterName, List<HeartbeatData> heartbeatDataList) {
         if (heartbeatDataList == null) {
             return;
@@ -220,6 +213,10 @@ public class ClusterTransactionService implements StartAndShutdown, TransactionS
             this.cluster = cluster;
         }
 
+        public String getCluster() {
+            return cluster;
+        }
+
         @Override
         public boolean equals(Object obj) {
             if (obj == this) {
@@ -264,7 +261,6 @@ public class ClusterTransactionService implements StartAndShutdown, TransactionS
         ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
         txHeartbeatServiceThread = new TxHeartbeatServiceThread();
 
-        mqClientAPIFactory.start();
         txHeartbeatServiceThread.start();
         heartbeatExecutors = ThreadPoolMonitor.createAndMonitor(
             proxyConfig.getTransactionHeartbeatThreadPoolNums(),
@@ -279,6 +275,5 @@ public class ClusterTransactionService implements StartAndShutdown, TransactionS
     public void shutdown() throws Exception {
         txHeartbeatServiceThread.shutdown();
         heartbeatExecutors.shutdown();
-        mqClientAPIFactory.shutdown();
     }
 }

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

@@ -34,17 +34,15 @@ import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 @Ignore
 @RunWith(MockitoJUnitRunner.Silent.class)
 public class BaseActivityTest extends InitConfigAndLoggerTest {
     protected static final Random RANDOM = new Random();
-    @Mock
     protected MessagingProcessor messagingProcessor;
-    @Mock
     protected GrpcClientSettingsManager grpcClientSettingsManager;
-    @Mock
     protected ProxyRelayService proxyRelayService;
 
     protected static final String REMOTE_ADDR = "192.168.0.1:8080";
@@ -55,6 +53,10 @@ public class BaseActivityTest extends InitConfigAndLoggerTest {
 
     public void before() throws Throwable {
         super.before();
+        messagingProcessor = mock(MessagingProcessor.class);
+        grpcClientSettingsManager = mock(GrpcClientSettingsManager.class);
+        proxyRelayService = mock(ProxyRelayService.class);
+
         metadata.put(InterceptorConstants.CLIENT_ID, CLIENT_ID);
         metadata.put(InterceptorConstants.LANGUAGE, "JAVA");
         metadata.put(InterceptorConstants.REMOTE_ADDRESS, REMOTE_ADDR);

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

@@ -0,0 +1,259 @@
+/*
+ * 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.grpc.v2.client;
+
+import apache.rocketmq.v2.ClientType;
+import apache.rocketmq.v2.Code;
+import apache.rocketmq.v2.FilterExpression;
+import apache.rocketmq.v2.FilterType;
+import apache.rocketmq.v2.HeartbeatRequest;
+import apache.rocketmq.v2.HeartbeatResponse;
+import apache.rocketmq.v2.NotifyClientTerminationRequest;
+import apache.rocketmq.v2.NotifyClientTerminationResponse;
+import apache.rocketmq.v2.Publishing;
+import apache.rocketmq.v2.Resource;
+import apache.rocketmq.v2.Settings;
+import apache.rocketmq.v2.Subscription;
+import apache.rocketmq.v2.SubscriptionEntry;
+import apache.rocketmq.v2.TelemetryCommand;
+import io.grpc.Context;
+import io.grpc.stub.StreamObserver;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import org.apache.rocketmq.broker.client.ClientChannelInfo;
+import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
+import org.apache.rocketmq.proxy.grpc.v2.BaseActivityTest;
+import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcClientChannel;
+import org.apache.rocketmq.remoting.protocol.LanguageCode;
+import org.assertj.core.util.Lists;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.when;
+
+public class ClientActivityTest extends BaseActivityTest {
+
+    private static final String TOPIC = "topic";
+    private static final String CONSUMER_GROUP = "consumerGroup";
+
+    private ClientActivity clientActivity;
+
+    @Before
+    public void before() throws Throwable {
+        super.before();
+        this.clientActivity = new ClientActivity(this.messagingProcessor, this.grpcClientSettingsManager);
+    }
+
+    protected TelemetryCommand sendProducerTelemetry(Context context) throws Throwable {
+        return this.sendClientTelemetry(
+            context,
+            Settings.newBuilder()
+                .setClientType(ClientType.PRODUCER)
+                .setPublishing(Publishing.newBuilder()
+                    .addTopics(Resource.newBuilder().setName(TOPIC).build())
+                    .build())
+                .build()).get();
+    }
+
+    protected HeartbeatResponse sendProducerHeartbeat(Context context) throws Throwable {
+        return this.clientActivity.heartbeat(context, HeartbeatRequest.newBuilder()
+            .setClientType(ClientType.PRODUCER)
+            .build()).get();
+    }
+
+    @Test
+    public void testProducerHeartbeat() throws Throwable {
+        Context context = createContext();
+
+        this.sendProducerTelemetry(context);
+
+        ArgumentCaptor<String> registerProducerGroupArgumentCaptor = ArgumentCaptor.forClass(String.class);
+        ArgumentCaptor<ClientChannelInfo> channelInfoArgumentCaptor = ArgumentCaptor.forClass(ClientChannelInfo.class);
+        doNothing().when(this.messagingProcessor).registerProducer(any(),
+            registerProducerGroupArgumentCaptor.capture(),
+            channelInfoArgumentCaptor.capture());
+
+        ArgumentCaptor<String> txProducerGroupArgumentCaptor = ArgumentCaptor.forClass(String.class);
+        ArgumentCaptor<String> txProducerTopicArgumentCaptor = ArgumentCaptor.forClass(String.class);
+        doNothing().when(this.messagingProcessor).addTransactionSubscription(any(),
+            txProducerGroupArgumentCaptor.capture(),
+            txProducerTopicArgumentCaptor.capture()
+        );
+
+        HeartbeatResponse response = this.sendProducerHeartbeat(context);
+
+        assertEquals(Code.OK, response.getStatus().getCode());
+
+        assertEquals(Lists.newArrayList(TOPIC), registerProducerGroupArgumentCaptor.getAllValues());
+        ClientChannelInfo clientChannelInfo = channelInfoArgumentCaptor.getValue();
+        assertClientChannelInfo(clientChannelInfo, TOPIC);
+
+        assertEquals(Lists.newArrayList(TOPIC), txProducerGroupArgumentCaptor.getAllValues());
+        assertEquals(Lists.newArrayList(TOPIC), txProducerTopicArgumentCaptor.getAllValues());
+    }
+
+    protected TelemetryCommand sendConsumerTelemetry(Context context) throws Throwable {
+        return this.sendClientTelemetry(
+            context,
+            Settings.newBuilder()
+                .setClientType(ClientType.PUSH_CONSUMER)
+                .setSubscription(Subscription.newBuilder()
+                    .addSubscriptions(SubscriptionEntry.newBuilder()
+                        .setExpression(FilterExpression.newBuilder()
+                            .setExpression("tag")
+                            .setType(FilterType.TAG)
+                            .build())
+                        .setTopic(Resource.newBuilder().setName(TOPIC).build())
+                        .build())
+                    .build())
+                .build()).get();
+    }
+
+    protected HeartbeatResponse sendConsumerHeartbeat(Context context) throws Throwable {
+        return this.clientActivity.heartbeat(context, HeartbeatRequest.newBuilder()
+            .setClientType(ClientType.PUSH_CONSUMER)
+            .setGroup(Resource.newBuilder().setName(CONSUMER_GROUP).build())
+            .build()).get();
+    }
+
+    @Test
+    public void testConsumerHeartbeat() throws Throwable {
+        Context context = createContext();
+        this.sendConsumerTelemetry(context);
+
+        ArgumentCaptor<Set<SubscriptionData>> subscriptionDatasArgumentCaptor = ArgumentCaptor.forClass(Set.class);
+        ArgumentCaptor<ClientChannelInfo> channelInfoArgumentCaptor = ArgumentCaptor.forClass(ClientChannelInfo.class);
+        doNothing().when(this.messagingProcessor).registerConsumer(any(),
+            anyString(),
+            channelInfoArgumentCaptor.capture(),
+            any(),
+            any(),
+            any(),
+            subscriptionDatasArgumentCaptor.capture()
+        );
+
+
+        HeartbeatResponse response = this.sendConsumerHeartbeat(context);
+        assertEquals(Code.OK, response.getStatus().getCode());
+
+        ClientChannelInfo clientChannelInfo = channelInfoArgumentCaptor.getValue();
+        assertClientChannelInfo(clientChannelInfo, CONSUMER_GROUP);
+
+        SubscriptionData data = subscriptionDatasArgumentCaptor.getValue().stream().findAny().get();
+        assertEquals("TAG", data.getExpressionType());
+        assertEquals("tag", data.getSubString());
+    }
+
+    protected void assertClientChannelInfo(ClientChannelInfo clientChannelInfo, String group) {
+        assertEquals(LanguageCode.JAVA, clientChannelInfo.getLanguage());
+        assertEquals(CLIENT_ID, clientChannelInfo.getClientId());
+        assertTrue(clientChannelInfo.getChannel() instanceof GrpcClientChannel);
+        GrpcClientChannel channel = (GrpcClientChannel) clientChannelInfo.getChannel();
+        assertEquals(REMOTE_ADDR, channel.getRemoteAddress());
+        assertEquals(LOCAL_ADDR, channel.getLocalAddress());
+        assertEquals(group, channel.getGroup());
+    }
+
+    @Test
+    public void testProducerNotifyClientTermination() throws Throwable {
+        Context context = createContext();
+
+        when(this.grpcClientSettingsManager.removeClientSettings(eq(CLIENT_ID))).thenReturn(Settings.newBuilder()
+            .setClientType(ClientType.PRODUCER)
+            .setPublishing(Publishing.newBuilder()
+                .addTopics(Resource.newBuilder().setName(TOPIC).build())
+                .build())
+            .build());
+        ArgumentCaptor<ClientChannelInfo> channelInfoArgumentCaptor = ArgumentCaptor.forClass(ClientChannelInfo.class);
+        doNothing().when(this.messagingProcessor).unRegisterProducer(any(), anyString(), channelInfoArgumentCaptor.capture());
+
+        this.sendProducerTelemetry(context);
+        this.sendProducerHeartbeat(context);
+
+        NotifyClientTerminationResponse response = this.clientActivity.notifyClientTermination(
+            context,
+            NotifyClientTerminationRequest.newBuilder()
+                .build()
+        ).get();
+
+        assertEquals(Code.OK, response.getStatus().getCode());
+        ClientChannelInfo clientChannelInfo = channelInfoArgumentCaptor.getValue();
+        assertClientChannelInfo(clientChannelInfo, TOPIC);
+    }
+
+    @Test
+    public void testConsumerNotifyClientTermination() throws Throwable {
+        Context context = createContext();
+
+        when(this.grpcClientSettingsManager.removeClientSettings(eq(CLIENT_ID))).thenReturn(Settings.newBuilder()
+            .setClientType(ClientType.PUSH_CONSUMER)
+            .build());
+        ArgumentCaptor<ClientChannelInfo> channelInfoArgumentCaptor = ArgumentCaptor.forClass(ClientChannelInfo.class);
+        doNothing().when(this.messagingProcessor).unRegisterConsumer(any(), anyString(), channelInfoArgumentCaptor.capture());
+
+        this.sendConsumerTelemetry(context);
+        this.sendConsumerHeartbeat(context);
+
+        NotifyClientTerminationResponse response = this.clientActivity.notifyClientTermination(
+            context,
+            NotifyClientTerminationRequest.newBuilder()
+                .setGroup(Resource.newBuilder().setName(CONSUMER_GROUP).build())
+                .build()
+        ).get();
+
+        assertEquals(Code.OK, response.getStatus().getCode());
+        ClientChannelInfo clientChannelInfo = channelInfoArgumentCaptor.getValue();
+        assertClientChannelInfo(clientChannelInfo, CONSUMER_GROUP);
+    }
+
+    protected CompletableFuture<TelemetryCommand> sendClientTelemetry(Context ctx, Settings settings) {
+        when(grpcClientSettingsManager.getClientSettings(any())).thenReturn(settings);
+
+        CompletableFuture<TelemetryCommand> future = new CompletableFuture<>();
+        StreamObserver<TelemetryCommand> responseObserver = new StreamObserver<TelemetryCommand>() {
+            @Override
+            public void onNext(TelemetryCommand value) {
+                future.complete(value);
+            }
+
+            @Override
+            public void onError(Throwable t) {
+
+            }
+
+            @Override public void onCompleted() {
+
+            }
+        };
+        StreamObserver<TelemetryCommand> requestObserver = this.clientActivity.telemetry(
+            ctx,
+            responseObserver
+        );
+        requestObserver.onNext(TelemetryCommand.newBuilder()
+            .setSettings(settings)
+            .build());
+        return future;
+    }
+}

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

@@ -0,0 +1,64 @@
+/*
+ * 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.grpc.v2.producer;
+
+import apache.rocketmq.v2.Code;
+import apache.rocketmq.v2.ForwardMessageToDeadLetterQueueRequest;
+import apache.rocketmq.v2.ForwardMessageToDeadLetterQueueResponse;
+import apache.rocketmq.v2.Resource;
+import java.util.concurrent.CompletableFuture;
+import org.apache.rocketmq.common.message.MessageClientIDSetter;
+import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.proxy.grpc.v2.BaseActivityTest;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.when;
+
+public class ForwardMessageToDLQActivityTest extends BaseActivityTest {
+
+    private ForwardMessageToDLQActivity forwardMessageToDLQActivity;
+
+    @Before
+    public void before() throws Throwable {
+        super.before();
+        this.forwardMessageToDLQActivity = new ForwardMessageToDLQActivity(this.messagingProcessor, this.grpcClientSettingsManager);
+    }
+
+    @Test
+    public void testForwardMessageToDeadLetterQueue() throws Throwable {
+        when(this.messagingProcessor.forwardMessageToDeadLetterQueue(any(), any(), anyString(), anyString(), anyString()))
+            .thenReturn(CompletableFuture.completedFuture(RemotingCommand.createResponseCommand(ResponseCode.SUCCESS, "")));
+
+        ForwardMessageToDeadLetterQueueResponse response = this.forwardMessageToDLQActivity.forwardMessageToDeadLetterQueue(
+            createContext(),
+            ForwardMessageToDeadLetterQueueRequest.newBuilder()
+                .setTopic(Resource.newBuilder().setName("topic").build())
+                .setGroup(Resource.newBuilder().setName("group").build())
+                .setMessageId(MessageClientIDSetter.createUniqID())
+                .setReceiptHandle(buildReceiptHandle("topic", System.currentTimeMillis(), 3000))
+                .build()
+        ).get();
+
+        assertEquals(Code.OK, response.getStatus().getCode());
+    }
+}

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

@@ -0,0 +1,101 @@
+/*
+ * 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.grpc.v2.transaction;
+
+import apache.rocketmq.v2.Code;
+import apache.rocketmq.v2.EndTransactionRequest;
+import apache.rocketmq.v2.EndTransactionResponse;
+import apache.rocketmq.v2.Resource;
+import apache.rocketmq.v2.TransactionResolution;
+import apache.rocketmq.v2.TransactionSource;
+import java.util.Arrays;
+import java.util.Collection;
+import org.apache.rocketmq.common.message.MessageClientIDSetter;
+import org.apache.rocketmq.proxy.grpc.v2.BaseActivityTest;
+import org.apache.rocketmq.proxy.processor.TransactionStatus;
+import org.apache.rocketmq.proxy.service.transaction.TransactionId;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.mockito.ArgumentCaptor;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doNothing;
+
+@RunWith(Parameterized.class)
+public class EndTransactionActivityTest extends BaseActivityTest {
+
+    private EndTransactionActivity endTransactionActivity;
+    private TransactionResolution resolution;
+    private TransactionSource source;
+    private TransactionStatus transactionStatus;
+    private Boolean fromTransactionCheck;
+
+    public EndTransactionActivityTest(TransactionResolution resolution, TransactionSource source,
+        TransactionStatus transactionStatus, Boolean fromTransactionCheck) {
+        this.resolution = resolution;
+        this.source = source;
+        this.transactionStatus = transactionStatus;
+        this.fromTransactionCheck = fromTransactionCheck;
+    }
+
+    @Before
+    public void before() throws Throwable {
+        super.before();
+        this.endTransactionActivity = new EndTransactionActivity(this.messagingProcessor, this.grpcClientSettingsManager);
+    }
+
+    @Test
+    public void testEndTransaction() throws Throwable {
+        ArgumentCaptor<TransactionStatus> transactionStatusCaptor = ArgumentCaptor.forClass(TransactionStatus.class);
+        ArgumentCaptor<Boolean> fromTransactionCheckCaptor = ArgumentCaptor.forClass(Boolean.class);
+        doNothing().when(this.messagingProcessor).endTransaction(any(), any(), anyString(), anyString(),
+            transactionStatusCaptor.capture(),
+            fromTransactionCheckCaptor.capture());
+
+        EndTransactionResponse response = this.endTransactionActivity.endTransaction(
+            createContext(),
+            EndTransactionRequest.newBuilder()
+                .setResolution(resolution)
+                .setTopic(Resource.newBuilder().setName("topic").build())
+                .setMessageId(MessageClientIDSetter.createUniqID())
+                .setTransactionId(
+                    TransactionId.genByBrokerTransactionId("brokerName", "0", 0, 0)
+                        .getProxyTransactionId())
+                .setSource(source)
+                .build()
+        ).get();
+
+        assertEquals(Code.OK, response.getStatus().getCode());
+        assertEquals(transactionStatus, transactionStatusCaptor.getValue());
+        assertEquals(fromTransactionCheck, fromTransactionCheckCaptor.getValue());
+    }
+
+    @Parameterized.Parameters
+    public static Collection<Object[]> parameters() {
+        Object[][] p = new Object[][]{
+            {TransactionResolution.COMMIT, TransactionSource.SOURCE_CLIENT, TransactionStatus.COMMIT, false},
+            {TransactionResolution.ROLLBACK, TransactionSource.SOURCE_SERVER_CHECK, TransactionStatus.ROLLBACK, true},
+            {TransactionResolution.TRANSACTION_RESOLUTION_UNSPECIFIED, TransactionSource.SOURCE_SERVER_CHECK, TransactionStatus.UNKNOWN, true},
+        };
+        return Arrays.asList(p);
+    }
+}

proxy/src/test/java/org/apache/rocketmq/proxy/service/BaseServiceTest.java

@@ -0,0 +1,83 @@
+/*
+ * 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.service;
+
+import java.util.HashMap;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.common.protocol.route.QueueData;
+import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.proxy.config.InitConfigAndLoggerTest;
+import org.apache.rocketmq.proxy.service.mqclient.MQClientAPIExt;
+import org.apache.rocketmq.proxy.service.mqclient.MQClientAPIFactory;
+import org.apache.rocketmq.proxy.service.route.MessageQueueView;
+import org.apache.rocketmq.proxy.service.route.TopicRouteService;
+import org.assertj.core.util.Lists;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@Ignore
+@RunWith(MockitoJUnitRunner.Silent.class)
+public class BaseServiceTest extends InitConfigAndLoggerTest {
+
+    protected TopicRouteService topicRouteService;
+    protected MQClientAPIFactory mqClientAPIFactory;
+    protected MQClientAPIExt mqClientAPIExt;
+
+    protected static final String ERR_TOPIC = "errTopic";
+    protected static final String TOPIC = "topic";
+    protected static final String GROUP = "group";
+    protected static final String BROKER_NAME = "broker";
+    protected static final String CLUSTER_NAME = "cluster";
+    protected static final String BROKER_ADDR = "127.0.0.1:10911";
+
+    protected final TopicRouteData topicRouteData = new TopicRouteData();
+    protected final QueueData queueData = new QueueData();
+    protected final BrokerData brokerData = new BrokerData();
+
+    @Before
+    public void before() throws Throwable {
+        super.before();
+
+        topicRouteService = mock(TopicRouteService.class);
+        mqClientAPIFactory = mock(MQClientAPIFactory.class);
+        mqClientAPIExt = mock(MQClientAPIExt.class);
+        when(mqClientAPIFactory.getClient()).thenReturn(mqClientAPIExt);
+
+        queueData.setBrokerName(BROKER_NAME);
+        topicRouteData.setQueueDatas(Lists.newArrayList(queueData));
+        brokerData.setCluster(CLUSTER_NAME);
+        brokerData.setBrokerName(BROKER_NAME);
+        HashMap<Long, String> brokerAddrs = new HashMap<>();
+        brokerAddrs.put(MixAll.MASTER_ID, BROKER_ADDR);
+        brokerData.setBrokerAddrs(brokerAddrs);
+        topicRouteData.setBrokerDatas(Lists.newArrayList(brokerData));
+
+        when(this.topicRouteService.getAllMessageQueueView(eq(ERR_TOPIC))).thenThrow(new MQClientException(ResponseCode.TOPIC_NOT_EXIST, ""));
+        when(this.topicRouteService.getAllMessageQueueView(eq(TOPIC))).thenReturn(new MessageQueueView(TOPIC, topicRouteData));
+        when(this.topicRouteService.getAllMessageQueueView(eq(CLUSTER_NAME))).thenReturn(new MessageQueueView(CLUSTER_NAME, topicRouteData));
+    }
+}

proxy/src/test/java/org/apache/rocketmq/proxy/service/metadata/ClusterMetadataServiceTest.java

@@ -0,0 +1,73 @@
+/*
+ * 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.service.metadata;
+
+import java.util.HashMap;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.attribute.TopicMessageType;
+import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.common.statictopic.TopicConfigAndQueueMapping;
+import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.proxy.config.ConfigurationManager;
+import org.apache.rocketmq.proxy.service.BaseServiceTest;
+import org.apache.rocketmq.proxy.service.route.MessageQueueView;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.when;
+
+public class ClusterMetadataServiceTest extends BaseServiceTest {
+
+    private ClusterMetadataService clusterMetadataService;
+
+    @Before
+    public void before() throws Throwable {
+        super.before();
+        ConfigurationManager.getProxyConfig().setRocketMQClusterName(CLUSTER_NAME);
+
+        TopicConfigAndQueueMapping topicConfigAndQueueMapping = new TopicConfigAndQueueMapping();
+        topicConfigAndQueueMapping.setAttributes(new HashMap<>());
+        topicConfigAndQueueMapping.setTopicMessageType(TopicMessageType.NORMAL);
+        when(this.mqClientAPIExt.getTopicConfig(anyString(), eq(TOPIC), anyLong())).thenReturn(topicConfigAndQueueMapping);
+
+        when(this.mqClientAPIExt.getSubscriptionGroupConfig(anyString(), eq(GROUP), anyLong())).thenReturn(new SubscriptionGroupConfig());
+
+        this.clusterMetadataService = new ClusterMetadataService(this.topicRouteService, this.mqClientAPIFactory);
+    }
+
+    @Test
+    public void testGetTopicMessageType() {
+        assertEquals(TopicMessageType.UNSPECIFIED, this.clusterMetadataService.getTopicMessageType(ERR_TOPIC));
+        assertEquals(1, this.clusterMetadataService.topicConfigCache.asMap().size());
+        assertEquals(TopicMessageType.UNSPECIFIED, this.clusterMetadataService.getTopicMessageType(ERR_TOPIC));
+
+        assertEquals(TopicMessageType.NORMAL, this.clusterMetadataService.getTopicMessageType(TOPIC));
+        assertEquals(2, this.clusterMetadataService.topicConfigCache.asMap().size());
+    }
+
+    @Test
+    public void testGetSubscriptionGroupConfig() {
+        assertNotNull(this.clusterMetadataService.getSubscriptionGroupConfig(GROUP));
+        assertEquals(1, this.clusterMetadataService.subscriptionGroupConfigCache.asMap().size());
+    }
+}

proxy/src/test/java/org/apache/rocketmq/proxy/service/route/ClusterTopicRouteServiceTest.java

@@ -0,0 +1,69 @@
+/*
+ * 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.service.route;
+
+import com.google.common.net.HostAndPort;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.proxy.common.Address;
+import org.apache.rocketmq.proxy.service.BaseServiceTest;
+import org.assertj.core.util.Lists;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.catchThrowableOfType;
+import static org.junit.Assert.*;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.when;
+
+public class ClusterTopicRouteServiceTest extends BaseServiceTest {
+
+    private ClusterTopicRouteService topicRouteService;
+
+    @Before
+    public void before() throws Throwable {
+        super.before();
+        this.topicRouteService = new ClusterTopicRouteService(this.mqClientAPIFactory);
+
+        when(this.mqClientAPIExt.getTopicRouteInfoFromNameServer(eq(TOPIC), anyLong())).thenReturn(topicRouteData);
+        when(this.mqClientAPIExt.getTopicRouteInfoFromNameServer(eq(ERR_TOPIC), anyLong())).thenThrow(new MQClientException(ResponseCode.TOPIC_NOT_EXIST, ""));
+    }
+
+    @Test
+    public void testGetCurrentMessageQueueView() throws Throwable {
+        MQClientException exception = catchThrowableOfType(() -> this.topicRouteService.getCurrentMessageQueueView(ERR_TOPIC), MQClientException.class);
+        assertTrue(TopicRouteHelper.isTopicNotExistError(exception));
+        assertEquals(1, this.topicRouteService.topicCache.asMap().size());
+
+        assertNotNull(this.topicRouteService.getCurrentMessageQueueView(TOPIC));
+        assertEquals(2, this.topicRouteService.topicCache.asMap().size());
+    }
+
+    @Test
+    public void testGetTopicRouteForProxy() throws Throwable {
+        List<Address> addressList = Lists.newArrayList(new Address(Address.AddressScheme.IPv4, HostAndPort.fromParts("127.0.0.1", 8888)));
+        ProxyTopicRouteData proxyTopicRouteData = this.topicRouteService.getTopicRouteForProxy(addressList, TOPIC);
+
+        assertEquals(1, proxyTopicRouteData.getBrokerDatas().size());
+        assertEquals(addressList, proxyTopicRouteData.getBrokerDatas().get(0).getBrokerAddrs().get(MixAll.MASTER_ID));
+    }
+}

proxy/src/test/java/org/apache/rocketmq/proxy/service/route/LocalTopicRouteServiceTest.java

@@ -0,0 +1,102 @@
+/*
+ * 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.service.route;
+
+import com.google.common.net.HostAndPort;
+import java.util.ArrayList;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.topic.TopicConfigManager;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.constant.PermName;
+import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.proxy.common.Address;
+import org.apache.rocketmq.proxy.config.ConfigurationManager;
+import org.apache.rocketmq.proxy.service.BaseServiceTest;
+import org.assertj.core.util.Lists;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.when;
+
+public class LocalTopicRouteServiceTest extends BaseServiceTest {
+
+    private static final String LOCAL_BROKER_NAME = "localBroker";
+    private static final String LOCAL_CLUSTER_NAME = "localCluster";
+    private static final String LOCAL_HOST = "127.0.0.2";
+    private static final int LOCAL_PORT = 10911;
+    private static final String LOCAL_ADDR = LOCAL_HOST + ":" + LOCAL_PORT;
+    @Mock
+    private BrokerController brokerController;
+    @Mock
+    private TopicConfigManager topicConfigManager;
+    private ConcurrentMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<>();
+    private BrokerConfig brokerConfig = new BrokerConfig();
+    private LocalTopicRouteService topicRouteService;
+
+    @Before
+    public void before() throws Throwable {
+        super.before();
+        this.brokerConfig.setBrokerName(LOCAL_BROKER_NAME);
+        this.brokerConfig.setBrokerClusterName(LOCAL_CLUSTER_NAME);
+
+        when(this.brokerController.getBrokerAddr()).thenReturn(LOCAL_ADDR);
+        when(this.brokerController.getBrokerConfig()).thenReturn(this.brokerConfig);
+        when(this.brokerController.getTopicConfigManager()).thenReturn(this.topicConfigManager);
+        when(this.topicConfigManager.getTopicConfigTable()).thenReturn(this.topicConfigTable);
+
+        this.topicRouteService = new LocalTopicRouteService(this.brokerController, this.mqClientAPIFactory);
+
+        when(this.mqClientAPIExt.getTopicRouteInfoFromNameServer(eq(TOPIC), anyLong())).thenReturn(topicRouteData);
+        when(this.mqClientAPIExt.getTopicRouteInfoFromNameServer(eq(ERR_TOPIC), anyLong())).thenThrow(new MQClientException(ResponseCode.TOPIC_NOT_EXIST, ""));
+    }
+
+    @Test
+    public void testGetCurrentMessageQueueView() throws Throwable {
+        this.topicConfigTable.put(TOPIC, new TopicConfig(TOPIC, 3, 2, PermName.PERM_WRITE | PermName.PERM_READ));
+        MessageQueueView messageQueueView = this.topicRouteService.getCurrentMessageQueueView(TOPIC);
+        assertEquals(3, messageQueueView.getReadSelector().getQueues().size());
+        assertEquals(2, messageQueueView.getWriteSelector().getQueues().size());
+        assertEquals(1, messageQueueView.getReadSelector().getBrokerActingQueues().size());
+        assertEquals(1, messageQueueView.getWriteSelector().getBrokerActingQueues().size());
+
+        assertEquals(LOCAL_ADDR, messageQueueView.getReadSelector().selectOne(true).getBrokerAddr());
+        assertEquals(LOCAL_BROKER_NAME, messageQueueView.getReadSelector().selectOne(true).getBrokerName());
+        assertEquals(messageQueueView.getReadSelector().selectOne(true), messageQueueView.getWriteSelector().selectOne(true));
+    }
+
+    @Test
+    public void testGetTopicRouteForProxy() throws Throwable {
+        ProxyTopicRouteData proxyTopicRouteData = this.topicRouteService.getTopicRouteForProxy(new ArrayList<>(), TOPIC);
+
+        assertEquals(1, proxyTopicRouteData.getBrokerDatas().size());
+        assertEquals(
+            Lists.newArrayList(new Address(Address.AddressScheme.IPv4, HostAndPort.fromParts(
+                HostAndPort.fromString(BROKER_ADDR).getHost(),
+                ConfigurationManager.getProxyConfig().getGrpcServerPort()))),
+            proxyTopicRouteData.getBrokerDatas().get(0).getBrokerAddrs().get(MixAll.MASTER_ID));
+    }
+}

proxy/src/test/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelectorTest.java

@@ -0,0 +1,84 @@
+/*
+ * 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.service.route;
+
+import org.apache.rocketmq.common.constant.PermName;
+import org.apache.rocketmq.proxy.service.BaseServiceTest;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class MessageQueueSelectorTest extends BaseServiceTest {
+
+    @Test
+    public void testReadMessageQueue() {
+        queueData.setPerm(PermName.PERM_READ);
+        queueData.setReadQueueNums(0);
+        MessageQueueSelector messageQueueSelector = new MessageQueueSelector(new TopicRouteWrapper(topicRouteData, TOPIC), true);
+        assertTrue(messageQueueSelector.getQueues().isEmpty());
+
+        queueData.setPerm(PermName.PERM_READ);
+        queueData.setReadQueueNums(3);
+        messageQueueSelector = new MessageQueueSelector(new TopicRouteWrapper(topicRouteData, TOPIC), true);
+        assertEquals(3, messageQueueSelector.getQueues().size());
+        assertEquals(1, messageQueueSelector.getBrokerActingQueues().size());
+        for (int i = 0; i < messageQueueSelector.getQueues().size(); i++) {
+            SelectableMessageQueue messageQueue = messageQueueSelector.getQueues().get(i);
+            assertEquals(i, messageQueue.getQueueId());
+        }
+
+        SelectableMessageQueue brokerQueue = messageQueueSelector.getQueueByBrokerName(BROKER_NAME);
+        assertEquals(brokerQueue, messageQueueSelector.getBrokerActingQueues().get(0));
+        assertEquals(brokerQueue, messageQueueSelector.selectOne(true));
+        assertEquals(brokerQueue, messageQueueSelector.selectOneByIndex(3, true));
+
+        SelectableMessageQueue queue = messageQueueSelector.selectOne(false);
+        messageQueueSelector.selectOne(false);
+        messageQueueSelector.selectOne(false);
+        assertEquals(queue, messageQueueSelector.selectOne(false));
+    }
+
+    @Test
+    public void testWriteMessageQueue() {
+        queueData.setPerm(PermName.PERM_WRITE);
+        queueData.setReadQueueNums(0);
+        MessageQueueSelector messageQueueSelector = new MessageQueueSelector(new TopicRouteWrapper(topicRouteData, TOPIC), false);
+        assertTrue(messageQueueSelector.getQueues().isEmpty());
+
+        queueData.setPerm(PermName.PERM_WRITE);
+        queueData.setWriteQueueNums(3);
+        messageQueueSelector = new MessageQueueSelector(new TopicRouteWrapper(topicRouteData, TOPIC), false);
+        assertEquals(3, messageQueueSelector.getQueues().size());
+        assertEquals(1, messageQueueSelector.getBrokerActingQueues().size());
+        for (int i = 0; i < messageQueueSelector.getQueues().size(); i++) {
+            SelectableMessageQueue messageQueue = messageQueueSelector.getQueues().get(i);
+            assertEquals(i, messageQueue.getQueueId());
+        }
+
+        SelectableMessageQueue brokerQueue = messageQueueSelector.getQueueByBrokerName(BROKER_NAME);
+        assertEquals(brokerQueue, messageQueueSelector.getBrokerActingQueues().get(0));
+        assertEquals(brokerQueue, messageQueueSelector.selectOne(true));
+        assertEquals(brokerQueue, messageQueueSelector.selectOneByIndex(3, true));
+
+        SelectableMessageQueue queue = messageQueueSelector.selectOne(false);
+        messageQueueSelector.selectOne(false);
+        messageQueueSelector.selectOne(false);
+        assertEquals(queue, messageQueueSelector.selectOne(false));
+    }
+}

proxy/src/test/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionServiceTest.java

@@ -0,0 +1,141 @@
+/*
+ * 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.service.transaction;
+
+import java.time.Duration;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.rocketmq.broker.client.ProducerManager;
+import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData;
+import org.apache.rocketmq.common.protocol.heartbeat.ProducerData;
+import org.apache.rocketmq.proxy.config.ConfigurationManager;
+import org.apache.rocketmq.proxy.service.BaseServiceTest;
+import org.apache.rocketmq.proxy.service.route.MessageQueueView;
+import org.assertj.core.util.Lists;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.when;
+
+
+public class ClusterTransactionServiceTest extends BaseServiceTest {
+
+    @Mock
+    private ProducerManager producerManager;
+
+    private ClusterTransactionService clusterTransactionService;
+
+    @Before
+    public void before() throws Throwable {
+        super.before();
+        this.clusterTransactionService = new ClusterTransactionService(this.topicRouteService, this.producerManager, null,
+            this.mqClientAPIFactory);
+
+        MessageQueueView messageQueueView = new MessageQueueView(TOPIC, topicRouteData);
+        when(this.topicRouteService.getAllMessageQueueView(anyString()))
+            .thenReturn(messageQueueView);
+
+        when(mqClientAPIFactory.getClient()).thenReturn(mqClientAPIExt);
+    }
+
+    @Test
+    public void testAddTransactionSubscription() {
+        this.clusterTransactionService.addTransactionSubscription(GROUP, TOPIC);
+
+        assertEquals(1, this.clusterTransactionService.getGroupClusterData().size());
+        assertEquals(CLUSTER_NAME, this.clusterTransactionService.getGroupClusterData().get(GROUP).stream().findAny().get().getCluster());
+    }
+
+    @Test
+    public void testAddTransactionSubscriptionTopicList() {
+        this.clusterTransactionService.addTransactionSubscription(GROUP, Lists.newArrayList(TOPIC + 1, TOPIC + 2));
+
+        assertEquals(1, this.clusterTransactionService.getGroupClusterData().size());
+        assertEquals(CLUSTER_NAME, this.clusterTransactionService.getGroupClusterData().get(GROUP).stream().findAny().get().getCluster());
+    }
+
+    @Test
+    public void testReplaceTransactionSubscription() {
+        this.clusterTransactionService.addTransactionSubscription(GROUP, TOPIC);
+
+        assertEquals(1, this.clusterTransactionService.getGroupClusterData().size());
+        assertEquals(CLUSTER_NAME, this.clusterTransactionService.getGroupClusterData().get(GROUP).stream().findAny().get().getCluster());
+
+        this.brokerData.setCluster(CLUSTER_NAME + 1);
+        this.clusterTransactionService.replaceTransactionSubscription(GROUP, Lists.newArrayList(TOPIC + 1));
+        assertEquals(1, this.clusterTransactionService.getGroupClusterData().size());
+        assertEquals(CLUSTER_NAME + 1, this.clusterTransactionService.getGroupClusterData().get(GROUP).stream().findAny().get().getCluster());
+    }
+
+    @Test
+    public void testUnSubscribeAllTransactionTopic() {
+        this.clusterTransactionService.addTransactionSubscription(GROUP, TOPIC);
+        this.clusterTransactionService.unSubscribeAllTransactionTopic(GROUP);
+
+        assertEquals(0, this.clusterTransactionService.getGroupClusterData().size());
+    }
+
+    @Test
+    public void testScanProducerHeartBeat() throws Exception {
+        ConfigurationManager.getProxyConfig().setTransactionHeartbeatBatchNum(2);
+        this.clusterTransactionService.start();
+        Set<String> groupSet = new HashSet<>();
+
+        for (int i = 0; i < 3; i++) {
+            groupSet.add(GROUP + i);
+            this.clusterTransactionService.addTransactionSubscription(GROUP + i, TOPIC);
+        }
+
+        ArgumentCaptor<String> brokerAddrArgumentCaptor = ArgumentCaptor.forClass(String.class);
+        ArgumentCaptor<HeartbeatData> heartbeatDataArgumentCaptor = ArgumentCaptor.forClass(HeartbeatData.class);
+        doNothing().when(mqClientAPIExt).sendHeartbeatOneway(
+            brokerAddrArgumentCaptor.capture(),
+            heartbeatDataArgumentCaptor.capture(),
+            anyLong()
+        );
+
+        this.clusterTransactionService.scanProducerHeartBeat();
+
+        await().atMost(Duration.ofSeconds(1)).until(() -> brokerAddrArgumentCaptor.getAllValues().size() == 2);
+
+        assertEquals(Lists.newArrayList(BROKER_ADDR, BROKER_ADDR), brokerAddrArgumentCaptor.getAllValues());
+        List<HeartbeatData> heartbeatDataList = heartbeatDataArgumentCaptor.getAllValues();
+        assertEquals(2, heartbeatDataList.get(0).getProducerDataSet().size());
+        for (ProducerData producerData : heartbeatDataList.get(0).getProducerDataSet()) {
+            groupSet.remove(producerData.getGroupName());
+        }
+
+        assertEquals(1, heartbeatDataList.get(1).getProducerDataSet().size());
+        for (ProducerData producerData : heartbeatDataList.get(1).getProducerDataSet()) {
+            groupSet.remove(producerData.getGroupName());
+        }
+
+        assertTrue(groupSet.isEmpty());
+    }
+}