[ISSUE #3949] Do some renaming work.

2026-10-25 10:03:57 +08:00 · 2022-07-13 11:29:10 +08:00
parent 63211bdcf8
commit 5c6fccacc8
31 changed files with 106 additions and 108 deletions
@@ -20,16 +20,17 @@ import com.google.common.cache.CacheLoader;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListenableFutureTask;
 import java.util.concurrent.ThreadPoolExecutor;
+import javax.annotation.Nonnull;

-public abstract class RetainCacheLoader<K, V> extends CacheLoader<K, V> {
+public abstract class AbstractCacheLoader<K, V> extends CacheLoader<K, V> {
    private final ThreadPoolExecutor cacheRefreshExecutor;

-    public RetainCacheLoader(ThreadPoolExecutor cacheRefreshExecutor) {
+    public AbstractCacheLoader(ThreadPoolExecutor cacheRefreshExecutor) {
        this.cacheRefreshExecutor = cacheRefreshExecutor;
    }

    @Override
-    public ListenableFuture<V> reload(K key, V oldValue) throws Exception {
+    public ListenableFuture<V> reload(@Nonnull K key, @Nonnull V oldValue) throws Exception {
        ListenableFutureTask<V> task = ListenableFutureTask.create(() -> {
            try {
                return getDirectly(key);
@@ -43,7 +44,7 @@ public abstract class RetainCacheLoader<K, V> extends CacheLoader<K, V> {
    }

    @Override
-    public V load(K key) throws Exception {
+    public V load(@Nonnull K key) throws Exception {
        return getDirectly(key);
    }

@@ -14,11 +14,11 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client;
+package org.apache.rocketmq.proxy.connector;

 import java.util.concurrent.ThreadLocalRandom;
 import org.apache.rocketmq.client.impl.MQClientAPIExtImpl;
-import org.apache.rocketmq.proxy.client.factory.ForwardClientFactory;
+import org.apache.rocketmq.proxy.connector.factory.ForwardClientFactory;
 import org.apache.rocketmq.proxy.common.StartAndShutdown;

 public abstract class AbstractForwardClient implements StartAndShutdown {
@@ -14,13 +14,14 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client;
+package org.apache.rocketmq.proxy.connector;

-import org.apache.rocketmq.proxy.client.factory.ForwardClientFactory;
-import org.apache.rocketmq.proxy.client.transaction.TransactionStateChecker;
+import org.apache.rocketmq.proxy.connector.factory.ForwardClientFactory;
+import org.apache.rocketmq.proxy.connector.route.TopicRouteCache;
+import org.apache.rocketmq.proxy.connector.transaction.TransactionStateChecker;
 import org.apache.rocketmq.proxy.common.AbstractStartAndShutdown;

-public class ForwardClientManager extends AbstractStartAndShutdown {
+public class ConnectorManager extends AbstractStartAndShutdown {
    private final ForwardClientFactory forwardClientFactory;
    private final DefaultForwardClient defaultForwardClient;
    private final ForwardProducer forwardProducer;
@@ -29,7 +30,7 @@ public class ForwardClientManager extends AbstractStartAndShutdown {

    private final TopicRouteCache topicRouteCache;

-    public ForwardClientManager(TransactionStateChecker transactionStateChecker) {
+    public ConnectorManager(TransactionStateChecker transactionStateChecker) {
        this.forwardClientFactory = new ForwardClientFactory(transactionStateChecker);
        this.defaultForwardClient = new DefaultForwardClient(this.forwardClientFactory);
        this.forwardProducer = new ForwardProducer(this.forwardClientFactory);
@@ -14,7 +14,7 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client;
+package org.apache.rocketmq.proxy.connector;

 import java.util.List;
 import java.util.concurrent.CompletableFuture;
@@ -22,15 +22,15 @@ import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.impl.MQClientAPIExtImpl;
 import org.apache.rocketmq.common.protocol.header.GetConsumerListByGroupRequestHeader;
 import org.apache.rocketmq.common.protocol.route.TopicRouteData;
-import org.apache.rocketmq.proxy.client.factory.ForwardClientFactory;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
+import org.apache.rocketmq.proxy.connector.factory.ForwardClientFactory;
 import org.apache.rocketmq.remoting.exception.RemotingException;

 public class DefaultForwardClient extends AbstractForwardClient {
    private static final String CID_PREFIX = "CID_RMQ_PROXY_DEFAULT_";

-    public DefaultForwardClient(ForwardClientFactory clientFactory) {
-        super(clientFactory);
+    public DefaultForwardClient(ForwardClientFactory forwardClientFactory) {
+        super(forwardClientFactory);
    }

    @Override
@@ -52,7 +52,10 @@ public class DefaultForwardClient extends AbstractForwardClient {
    }

    public CompletableFuture<List<String>> getConsumerListByGroup(
-        String brokerAddr, GetConsumerListByGroupRequestHeader requestHeader, long timeoutMillis) {
+        String brokerAddr,
+        GetConsumerListByGroupRequestHeader requestHeader,
+        long timeoutMillis
+    ) {
        return getClient().getConsumerListByGroup(brokerAddr, requestHeader, timeoutMillis);
    }

@@ -14,7 +14,7 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client;
+package org.apache.rocketmq.proxy.connector;

 import java.util.concurrent.CompletableFuture;
 import org.apache.rocketmq.client.impl.MQClientAPIExtImpl;
@@ -23,7 +23,7 @@ import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.protocol.header.ConsumerSendMsgBackRequestHeader;
 import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
 import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData;
-import org.apache.rocketmq.proxy.client.factory.ForwardClientFactory;
+import org.apache.rocketmq.proxy.connector.factory.ForwardClientFactory;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;

@@ -14,7 +14,7 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client;
+package org.apache.rocketmq.proxy.connector;

 import java.util.concurrent.CompletableFuture;
 import org.apache.rocketmq.client.consumer.PopResult;
@@ -22,7 +22,7 @@ import org.apache.rocketmq.client.consumer.PullResult;
 import org.apache.rocketmq.client.impl.MQClientAPIExtImpl;
 import org.apache.rocketmq.common.protocol.header.PopMessageRequestHeader;
 import org.apache.rocketmq.common.protocol.header.PullMessageRequestHeader;
-import org.apache.rocketmq.proxy.client.factory.ForwardClientFactory;
+import org.apache.rocketmq.proxy.connector.factory.ForwardClientFactory;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;

 public class ForwardReadConsumer extends AbstractForwardClient {
@@ -14,7 +14,7 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client;
+package org.apache.rocketmq.proxy.connector;

 import java.util.concurrent.CompletableFuture;
 import org.apache.rocketmq.client.consumer.AckResult;
@@ -22,7 +22,7 @@ import org.apache.rocketmq.client.impl.MQClientAPIExtImpl;
 import org.apache.rocketmq.common.protocol.header.AckMessageRequestHeader;
 import org.apache.rocketmq.common.protocol.header.ChangeInvisibleTimeRequestHeader;
 import org.apache.rocketmq.common.protocol.header.UpdateConsumerOffsetRequestHeader;
-import org.apache.rocketmq.proxy.client.factory.ForwardClientFactory;
+import org.apache.rocketmq.proxy.connector.factory.ForwardClientFactory;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
 import org.apache.rocketmq.remoting.exception.RemotingException;

@@ -14,7 +14,7 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client.factory;
+package org.apache.rocketmq.proxy.connector.factory;

 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -23,21 +23,21 @@ import org.apache.rocketmq.remoting.netty.NettyClientConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

-public abstract class AbstractMQClientFactory<T>  {
-    private static final Logger LOGGER = LoggerFactory.getLogger(AbstractMQClientFactory.class);
+public abstract class AbstractClientFactory<T>  {
+    private static final Logger LOGGER = LoggerFactory.getLogger(AbstractClientFactory.class);

    protected Map<String, T> cacheTable = new ConcurrentHashMap<>();
    protected RPCHook rpcHook;

-    public AbstractMQClientFactory(RPCHook rpcHook) {
+    public AbstractClientFactory(RPCHook rpcHook) {
        this.rpcHook = rpcHook;
    }

-    abstract T newOne(String instanceName, RPCHook rpcHook, int bootstrapWorkerThreads) throws Throwable;
+    protected abstract T newOne(String instanceName, RPCHook rpcHook, int bootstrapWorkerThreads) throws Throwable;

-    abstract boolean tryStart(T t);
+    protected abstract boolean tryStart(T t);

-    abstract void shutdown(T t);
+    protected abstract void shutdown(T t);

    protected static NettyClientConfig createNettyClientConfig(int bootstrapWorkerThreads) {
        NettyClientConfig nettyClientConfig = new NettyClientConfig();
@@ -14,13 +14,13 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client.factory;
+package org.apache.rocketmq.proxy.connector.factory;

 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.client.ClientConfig;
 import org.apache.rocketmq.client.impl.MQClientAPIExtImpl;
 import org.apache.rocketmq.common.MixAll;
-import org.apache.rocketmq.proxy.client.transaction.TransactionStateChecker;
+import org.apache.rocketmq.proxy.connector.transaction.TransactionStateChecker;
 import org.apache.rocketmq.proxy.common.StartAndShutdown;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
 import org.apache.rocketmq.remoting.RPCHook;
@@ -30,13 +30,13 @@ public class ForwardClientFactory implements StartAndShutdown {
    private RPCHook rpcHook = null;

    private final MQClientFactory mqClientFactory;
-    private final TransactionalProducerFactory transactionalProducerFactory;
+    private final TransactionProducerFactory transactionalProducerFactory;

    public ForwardClientFactory(TransactionStateChecker transactionStateChecker) {
        this.init();

        this.mqClientFactory = new MQClientFactory(this.rpcHook);
-        this.transactionalProducerFactory = new TransactionalProducerFactory(this.rpcHook, transactionStateChecker);
+        this.transactionalProducerFactory = new TransactionProducerFactory(this.rpcHook, transactionStateChecker);
    }

    private void init() {
@@ -14,21 +14,21 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client.factory;
+package org.apache.rocketmq.proxy.connector.factory;

 import org.apache.rocketmq.client.ClientConfig;
 import org.apache.rocketmq.client.impl.MQClientAPIExtImpl;
-import org.apache.rocketmq.proxy.client.processor.DoNothingClientRemotingProcessor;
+import org.apache.rocketmq.proxy.connector.processor.DoNothingClientRemotingProcessor;
 import org.apache.rocketmq.remoting.RPCHook;

-public class MQClientFactory extends AbstractMQClientFactory<MQClientAPIExtImpl> {
+public class MQClientFactory extends AbstractClientFactory<MQClientAPIExtImpl> {

    public MQClientFactory(RPCHook rpcHook) {
        super(rpcHook);
    }

    @Override
-    MQClientAPIExtImpl newOne(String instanceName, RPCHook rpcHook, int bootstrapWorkerThreads) {
+    protected MQClientAPIExtImpl newOne(String instanceName, RPCHook rpcHook, int bootstrapWorkerThreads) {
        ClientConfig clientConfig = new ClientConfig();
        clientConfig.setInstanceName(instanceName);

@@ -41,13 +41,13 @@ public class MQClientFactory extends AbstractMQClientFactory<MQClientAPIExtImpl>
    }

    @Override
-    boolean tryStart(MQClientAPIExtImpl client) {
+    protected boolean tryStart(MQClientAPIExtImpl client) {
        client.start();
        return true;
    }

    @Override
-    void shutdown(MQClientAPIExtImpl client) {
+    protected void shutdown(MQClientAPIExtImpl client) {
        client.shutdown();
    }
 }
@@ -14,24 +14,24 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client.factory;
+package org.apache.rocketmq.proxy.connector.factory;

 import org.apache.rocketmq.client.ClientConfig;
 import org.apache.rocketmq.client.impl.MQClientAPIExtImpl;
-import org.apache.rocketmq.proxy.client.processor.ProxyClientRemotingProcessor;
-import org.apache.rocketmq.proxy.client.transaction.TransactionStateChecker;
+import org.apache.rocketmq.proxy.connector.processor.ProxyClientRemotingProcessor;
+import org.apache.rocketmq.proxy.connector.transaction.TransactionStateChecker;
 import org.apache.rocketmq.remoting.RPCHook;

-public class TransactionalProducerFactory extends AbstractMQClientFactory<MQClientAPIExtImpl> {
+public class TransactionProducerFactory extends AbstractClientFactory<MQClientAPIExtImpl> {
    private final TransactionStateChecker transactionStateChecker;

-    public TransactionalProducerFactory(RPCHook rpcHook, TransactionStateChecker transactionStateChecker) {
+    public TransactionProducerFactory(RPCHook rpcHook, TransactionStateChecker transactionStateChecker) {
        super(rpcHook);
        this.transactionStateChecker = transactionStateChecker;
    }

    @Override
-    MQClientAPIExtImpl newOne(String instanceName, RPCHook rpcHook, int bootstrapWorkerThreads) {
+    public MQClientAPIExtImpl newOne(String instanceName, RPCHook rpcHook, int bootstrapWorkerThreads) {
        ClientConfig clientConfig = new ClientConfig();
        clientConfig.setInstanceName(instanceName);

@@ -44,13 +44,13 @@ public class TransactionalProducerFactory extends AbstractMQClientFactory<MQClie
    }

    @Override
-    boolean tryStart(MQClientAPIExtImpl client) {
+    protected boolean tryStart(MQClientAPIExtImpl client) {
        client.start();
        return true;
    }

    @Override
-    void shutdown(MQClientAPIExtImpl client) {
+    protected void shutdown(MQClientAPIExtImpl client) {
        client.shutdown();
    }
 }
@@ -14,7 +14,7 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client.processor;
+package org.apache.rocketmq.proxy.connector.processor;

 import io.netty.channel.ChannelHandlerContext;
 import org.apache.rocketmq.client.impl.ClientRemotingProcessor;
@@ -14,7 +14,7 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client.processor;
+package org.apache.rocketmq.proxy.connector.processor;

 import io.netty.channel.ChannelHandlerContext;
 import java.nio.ByteBuffer;
@@ -24,9 +24,9 @@ import org.apache.rocketmq.common.message.MessageDecoder;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.protocol.RequestCode;
 import org.apache.rocketmq.common.protocol.header.CheckTransactionStateRequestHeader;
-import org.apache.rocketmq.proxy.client.transaction.TransactionId;
-import org.apache.rocketmq.proxy.client.transaction.TransactionStateCheckRequest;
-import org.apache.rocketmq.proxy.client.transaction.TransactionStateChecker;
+import org.apache.rocketmq.proxy.connector.transaction.TransactionId;
+import org.apache.rocketmq.proxy.connector.transaction.TransactionStateCheckRequest;
+import org.apache.rocketmq.proxy.connector.transaction.TransactionStateChecker;
 import org.apache.rocketmq.remoting.exception.RemotingCommandException;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;

@@ -14,7 +14,7 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client.route;
+package org.apache.rocketmq.proxy.connector.route;

 import java.util.ArrayList;
 import java.util.Collections;
@@ -14,7 +14,7 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client.route;
+package org.apache.rocketmq.proxy.connector.route;

 import org.apache.rocketmq.common.protocol.route.TopicRouteData;

@@ -14,7 +14,7 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client.route;
+package org.apache.rocketmq.proxy.connector.route;

 import java.util.Objects;
 import org.apache.rocketmq.common.message.MessageQueue;
@@ -14,7 +14,7 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client;
+package org.apache.rocketmq.proxy.connector.route;

 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.LoadingCache;
@@ -26,10 +26,8 @@ import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.protocol.ResponseCode;
 import org.apache.rocketmq.common.protocol.route.TopicRouteData;
 import org.apache.rocketmq.common.thread.ThreadPoolMonitor;
-import org.apache.rocketmq.proxy.client.route.SelectableMessageQueue;
-import org.apache.rocketmq.proxy.client.route.MessageQueueWrapper;
-import org.apache.rocketmq.proxy.common.RetainCacheLoader;
-import org.apache.rocketmq.proxy.common.RocketMQHelper;
+import org.apache.rocketmq.proxy.connector.DefaultForwardClient;
+import org.apache.rocketmq.proxy.common.AbstractCacheLoader;
 import org.apache.rocketmq.proxy.common.utils.ProxyUtils;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
 import org.apache.rocketmq.proxy.config.ProxyConfig;
@@ -98,7 +96,7 @@ public class TopicRouteCache {
            && routeData.getBrokerDatas() != null && !routeData.getBrokerDatas().isEmpty();
    }

-    protected abstract class AbstractTopicRouteCacheLoader extends RetainCacheLoader<String, MessageQueueWrapper> {
+    protected abstract class AbstractTopicRouteCacheLoader extends AbstractCacheLoader<String, MessageQueueWrapper> {

        public AbstractTopicRouteCacheLoader() {
            super(cacheRefreshExecutor);
@@ -124,7 +122,7 @@ public class TopicRouteCache {
                }
                return MessageQueueWrapper.WRAPPED_EMPTY_QUEUE;
            } catch (Exception e) {
-                if (RocketMQHelper.isTopicNotExistError(e)) {
+                if (TopicRouteHelper.isTopicNotExistError(e)) {
                    return MessageQueueWrapper.WRAPPED_EMPTY_QUEUE;
                }
                throw e;
@@ -14,14 +14,14 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.common;
+package org.apache.rocketmq.proxy.connector.route;

 import org.apache.rocketmq.client.common.ClientErrorCode;
 import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.protocol.ResponseCode;

-public class RocketMQHelper {
+public class TopicRouteHelper {

    public static boolean isTopicNotExistError(Throwable e) {
        if (e instanceof MQBrokerException) {
@@ -29,24 +29,20 @@ public class RocketMQHelper {
                return true;
            }
        }
-        if (e instanceof MQClientException) {
-            if (((MQClientException) e).getResponseCode() == ResponseCode.TOPIC_NOT_EXIST) {
-                return true;
-            }

-            if (((MQClientException) e).getResponseCode() == ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION) {
+        if (e instanceof MQClientException) {
+            int code = ((MQClientException) e).getResponseCode();
+            if (code == ResponseCode.TOPIC_NOT_EXIST || code == ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION) {
                return true;
            }

            Throwable cause = e.getCause();
            if (cause instanceof MQClientException) {
-                if (((MQClientException) cause).getResponseCode() == ResponseCode.TOPIC_NOT_EXIST) {
-                    return true;
-                }
-
-                return ((MQClientException) cause).getResponseCode() == ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION;
+                int causeCode = ((MQClientException) cause).getResponseCode();
+                return causeCode == ResponseCode.TOPIC_NOT_EXIST || causeCode == ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION;
            }
        }
+
        return false;
    }
 }
@@ -14,7 +14,7 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client.route;
+package org.apache.rocketmq.proxy.connector.route;

 import java.util.HashMap;
 import java.util.List;
@@ -14,7 +14,7 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client.transaction;
+package org.apache.rocketmq.proxy.connector.transaction;

 import java.net.InetAddress;
 import java.net.InetSocketAddress;
@@ -14,7 +14,7 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client.transaction;
+package org.apache.rocketmq.proxy.connector.transaction;

 import org.apache.rocketmq.common.message.MessageExt;

@@ -14,7 +14,7 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
-package org.apache.rocketmq.proxy.client.transaction;
+package org.apache.rocketmq.proxy.connector.transaction;

 public interface TransactionStateChecker {
    void checkTransactionState(TransactionStateCheckRequest checkData);
@@ -54,7 +54,7 @@ import apache.rocketmq.v1.SendMessageResponse;
 import io.grpc.Context;
 import java.util.concurrent.CompletableFuture;
 import org.apache.rocketmq.common.constant.LoggerName;
-import org.apache.rocketmq.proxy.client.ForwardClientManager;
+import org.apache.rocketmq.proxy.connector.ConnectorManager;
 import org.apache.rocketmq.proxy.common.AbstractStartAndShutdown;
 import org.apache.rocketmq.proxy.grpc.service.cluster.ProducerService;
 import org.apache.rocketmq.proxy.grpc.service.cluster.RouteService;
@@ -64,12 +64,12 @@ import org.slf4j.LoggerFactory;
 public class ClusterGrpcService extends AbstractStartAndShutdown implements GrpcForwardService {
    private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.GRPC_LOGGER_NAME);

-    private final ForwardClientManager forwardClientManager;
+    private final ConnectorManager forwardClientManager;
    private final ProducerService producerService;
    private final RouteService routeService;

    public ClusterGrpcService() {
-        this.forwardClientManager = new ForwardClientManager(checkData -> {
+        this.forwardClientManager = new ConnectorManager(checkData -> {
        });
        this.producerService = new ProducerService(forwardClientManager);
        this.routeService = new RouteService(forwardClientManager);
@@ -16,13 +16,13 @@
 */
 package org.apache.rocketmq.proxy.grpc.service.cluster;

-import org.apache.rocketmq.proxy.client.ForwardClientManager;
+import org.apache.rocketmq.proxy.connector.ConnectorManager;

 public class BaseService {

-    protected final ForwardClientManager clientManager;
+    protected final ConnectorManager clientManager;

-    public BaseService(ForwardClientManager clientManager) {
+    public BaseService(ConnectorManager clientManager) {
        this.clientManager = clientManager;
    }
 }
@@ -20,11 +20,11 @@ import apache.rocketmq.v1.ReceiveMessageRequest;
 import apache.rocketmq.v1.ReceiveMessageResponse;
 import io.grpc.Context;
 import java.util.concurrent.CompletableFuture;
-import org.apache.rocketmq.proxy.client.ForwardClientManager;
+import org.apache.rocketmq.proxy.connector.ConnectorManager;

 public class ConsumerService extends BaseService {

-    public ConsumerService(ForwardClientManager clientManager) {
+    public ConsumerService(ConnectorManager clientManager) {
        super(clientManager);
    }

@@ -25,8 +25,8 @@ import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.message.MessageConst;
 import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
-import org.apache.rocketmq.proxy.client.ForwardClientManager;
-import org.apache.rocketmq.proxy.client.route.SelectableMessageQueue;
+import org.apache.rocketmq.proxy.connector.ConnectorManager;
+import org.apache.rocketmq.proxy.connector.route.SelectableMessageQueue;
 import org.apache.rocketmq.proxy.common.utils.ProxyUtils;
 import org.apache.rocketmq.proxy.grpc.common.Converter;
 import org.apache.rocketmq.proxy.grpc.common.ProxyException;
@@ -42,7 +42,7 @@ public class ProducerService extends BaseService {
    private volatile ProducerServiceHook producerServiceHook = null;
    private volatile MessageQueueSelector messageQueueSelector = new DefaultMessageQueueSelector();

-    public ProducerService(ForwardClientManager clientManager) {
+    public ProducerService(ConnectorManager clientManager) {
        super(clientManager);
    }

@@ -34,10 +34,10 @@ import java.util.concurrent.CompletableFuture;
 import org.apache.rocketmq.common.constant.PermName;
 import org.apache.rocketmq.common.protocol.route.QueueData;
 import org.apache.rocketmq.common.protocol.route.TopicRouteData;
-import org.apache.rocketmq.proxy.client.ForwardClientManager;
-import org.apache.rocketmq.proxy.client.route.SelectableMessageQueue;
-import org.apache.rocketmq.proxy.client.route.MessageQueueWrapper;
-import org.apache.rocketmq.proxy.common.RocketMQHelper;
+import org.apache.rocketmq.proxy.connector.ConnectorManager;
+import org.apache.rocketmq.proxy.connector.route.SelectableMessageQueue;
+import org.apache.rocketmq.proxy.connector.route.MessageQueueWrapper;
+import org.apache.rocketmq.proxy.connector.route.TopicRouteHelper;
 import org.apache.rocketmq.proxy.grpc.common.Converter;
 import org.apache.rocketmq.proxy.grpc.common.ResponseBuilder;

@@ -47,7 +47,7 @@ public class RouteService extends BaseService {
    private volatile QueryRouteHook queryRouteHook = null;
    private volatile QueryAssignmentHook queryAssignmentHook = null;

-    public RouteService(ForwardClientManager clientManager) {
+    public RouteService(ConnectorManager clientManager) {
        super(clientManager);
    }

@@ -126,7 +126,7 @@ public class RouteService extends BaseService {
                .build();
            future.complete(response);
        } catch (Throwable t) {
-            if (RocketMQHelper.isTopicNotExistError(t)) {
+            if (TopicRouteHelper.isTopicNotExistError(t)) {
                future.complete(QueryRouteResponse.newBuilder()
                    .setCommon(ResponseBuilder.buildCommon(Code.NOT_FOUND, t.getMessage()))
                    .build());
@@ -15,9 +15,9 @@
 * limitations under the License.
 */

-package org.apache.rocketmq.proxy.client;
+package org.apache.rocketmq.proxy.connector;

-import org.apache.rocketmq.proxy.client.transaction.TransactionStateChecker;
+import org.apache.rocketmq.proxy.connector.transaction.TransactionStateChecker;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
 import org.apache.rocketmq.proxy.config.InitConfigAndLoggerTest;
 import org.junit.Test;
@@ -30,7 +30,7 @@ public class ForwardClientManagerTest extends InitConfigAndLoggerTest {
    @Test
    public void testClientManager() throws Exception {
        TransactionStateChecker mockedTransactionStateChecker = Mockito.mock(TransactionStateChecker.class);
-        ForwardClientManager clientManager = new ForwardClientManager(mockedTransactionStateChecker);
+        ConnectorManager clientManager = new ConnectorManager(mockedTransactionStateChecker);
        clientManager.start();

        assertThat(clientManager.getDefaultForwardClient()).isNotNull();
@@ -16,12 +16,12 @@
 */
 package org.apache.rocketmq.proxy.grpc.service.cluster;

-import org.apache.rocketmq.proxy.client.ForwardClientManager;
-import org.apache.rocketmq.proxy.client.DefaultForwardClient;
-import org.apache.rocketmq.proxy.client.ForwardProducer;
-import org.apache.rocketmq.proxy.client.ForwardReadConsumer;
-import org.apache.rocketmq.proxy.client.TopicRouteCache;
-import org.apache.rocketmq.proxy.client.ForwardWriteConsumer;
+import org.apache.rocketmq.proxy.connector.ConnectorManager;
+import org.apache.rocketmq.proxy.connector.DefaultForwardClient;
+import org.apache.rocketmq.proxy.connector.ForwardProducer;
+import org.apache.rocketmq.proxy.connector.ForwardReadConsumer;
+import org.apache.rocketmq.proxy.connector.route.TopicRouteCache;
+import org.apache.rocketmq.proxy.connector.ForwardWriteConsumer;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.runner.RunWith;
@@ -35,7 +35,7 @@ import static org.mockito.Mockito.when;
 public abstract class BaseServiceTest {

    @Mock
-    protected ForwardClientManager clientManager;
+    protected ConnectorManager clientManager;
    @Mock
    protected DefaultForwardClient defaultClient;
    @Mock
@@ -35,7 +35,7 @@ import org.apache.rocketmq.client.producer.SendStatus;
 import org.apache.rocketmq.common.message.MessageConst;
 import org.apache.rocketmq.common.message.MessageQueue;
 import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
-import org.apache.rocketmq.proxy.client.route.SelectableMessageQueue;
+import org.apache.rocketmq.proxy.connector.route.SelectableMessageQueue;
 import org.apache.rocketmq.proxy.grpc.common.ProxyException;
 import org.apache.rocketmq.proxy.grpc.common.ProxyResponseCode;
 import org.junit.Test;
@@ -19,13 +19,12 @@ 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.client.route.MessageQueueWrapper;
+import org.apache.rocketmq.proxy.connector.route.MessageQueueWrapper;
 import org.apache.rocketmq.proxy.grpc.common.ResponseBuilder;
 import org.junit.Test;

 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
-import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.when;

 public class RouteServiceTest extends BaseServiceTest {