Merge branch 'snode' of github.com:apache/rocketmq into snode

client/src/main/java/org/apache/rocketmq/client/consumer/store/LocalFileOffsetStore.java

@@ -160,6 +160,12 @@ public class LocalFileOffsetStore implements OffsetStore {
 
     }
 
+    @Override
+    public void updateConsumeOffsetToSnode(final MessageQueue mq, final long offset, final boolean isOneway)
+        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+
+    }
+
     @Override
     public void updateConsumeOffsetToBroker(final MessageQueue mq, final long offset, final boolean isOneway)
         throws RemotingException, MQBrokerException, InterruptedException, MQClientException {

client/src/main/java/org/apache/rocketmq/client/consumer/store/OffsetStore.java

@@ -71,4 +71,12 @@ public interface OffsetStore {
      */
     void updateConsumeOffsetToBroker(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,
         MQBrokerException, InterruptedException, MQClientException;
+
+    /**
+     * @param mq
+     * @param offset
+     * @param isOneway
+     */
+    void updateConsumeOffsetToSnode(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,
+        MQBrokerException, InterruptedException, MQClientException;
 }

client/src/main/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStore.java

@@ -87,7 +87,7 @@ public class RemoteBrokerOffsetStore implements OffsetStore {
                 }
                 case READ_FROM_STORE: {
                     try {
-                        long brokerOffset = this.fetchConsumeOffsetFromBroker(mq);
+                        long brokerOffset = this.fetchConsumeOffsetFromSnode(mq);
                         AtomicLong offset = new AtomicLong(brokerOffset);
                         this.updateOffset(mq, offset.get(), false);
                         return brokerOffset;
@@ -195,14 +195,14 @@ public class RemoteBrokerOffsetStore implements OffsetStore {
 
     private void updateConsumeOffsetToSnode(MessageQueue mq, long offset) throws RemotingException,
         MQBrokerException, InterruptedException, MQClientException {
-        updateConsumeOffsetToBroker(mq, offset, true);
+        updateConsumeOffsetToSnode(mq, offset, true);
     }
 
     /**
      * Update the Consumer Offset synchronously, once the Master is off, updated to Slave, here need to be optimized.
      */
     @Override
-    public void updateConsumeOffsetToBroker(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,
+    public void updateConsumeOffsetToSnode(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,
         MQBrokerException, InterruptedException, MQClientException {
 
         String snodeAddr = this.mQClientFactory.findSnodeAddressInPublish();
@@ -226,10 +226,65 @@ public class RemoteBrokerOffsetStore implements OffsetStore {
                     snodeAddr, requestHeader, 1000 * 5);
             }
         } else {
-            throw new MQClientException("Update offset to Broker[" + mq.getBrokerName() + "] failed, Snode is null.", null);
+            throw new MQClientException("Update offset to Snode[" + mq.getBrokerName() + "] failed, Snode is null.", null);
+        }
+    }
+
+    /**
+     * Preserved firstly,Compatible with RocketMQ 4.X Version
+     */
+    @Override
+    public void updateConsumeOffsetToBroker(MessageQueue mq, long offset,
+        boolean isOneway) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInAdmin(mq.getBrokerName());
+        if (null == findBrokerResult) {
+            this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
+            findBrokerResult = this.mQClientFactory.findBrokerAddressInAdmin(mq.getBrokerName());
+        }
+
+        if (findBrokerResult != null) {
+            UpdateConsumerOffsetRequestHeader requestHeader = new UpdateConsumerOffsetRequestHeader();
+            requestHeader.setTopic(mq.getTopic());
+            requestHeader.setConsumerGroup(this.groupName);
+            requestHeader.setQueueId(mq.getQueueId());
+            requestHeader.setCommitOffset(offset);
+            requestHeader.setEnodeName(mq.getBrokerName());
+            if (isOneway) {
+                this.mQClientFactory.getMQClientAPIImpl().updateConsumerOffsetOneway(
+                    findBrokerResult.getBrokerAddr(), requestHeader, 1000 * 5);
+            } else {
+                this.mQClientFactory.getMQClientAPIImpl().updateConsumerOffset(
+                    findBrokerResult.getBrokerAddr(), requestHeader, 1000 * 5);
+            }
+        } else {
+            throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
         }
     }
 
+    private long fetchConsumeOffsetFromSnode(MessageQueue mq) throws RemotingException, MQBrokerException,
+        InterruptedException, MQClientException {
+        String snodeAddr = this.mQClientFactory.findSnodeAddressInPublish();
+        if (null == snodeAddr) {
+            this.mQClientFactory.updateSnodeInfoFromNameServer();
+            snodeAddr = this.mQClientFactory.findSnodeAddressInPublish();
+        }
+
+        if (snodeAddr != null) {
+            QueryConsumerOffsetRequestHeader requestHeader = new QueryConsumerOffsetRequestHeader();
+            requestHeader.setTopic(mq.getTopic());
+            requestHeader.setConsumerGroup(this.groupName);
+            requestHeader.setQueueId(mq.getQueueId());
+            requestHeader.setEnodeName(mq.getBrokerName());
+            return this.mQClientFactory.getMQClientAPIImpl().queryConsumerOffset(
+                snodeAddr, requestHeader, 1000 * 5);
+        } else {
+            throw new MQClientException("Get Offset from Snode[" + mq.getBrokerName() + "] failed, Snode is not exist", null);
+        }
+    }
+
+    /**
+     * Preserved firstly,Compatible with RocketMQ 4.X Version
+     */
     private long fetchConsumeOffsetFromBroker(MessageQueue mq) throws RemotingException, MQBrokerException,
         InterruptedException, MQClientException {
         String snodeAddr = this.mQClientFactory.findSnodeAddressInPublish();

client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java

@@ -1194,7 +1194,7 @@ public class MQClientAPIImpl {
     public SnodeClusterInfo getSnodeClusterInfo(
         //Todo Redifine snode exception
         final long timeoutMillis) throws InterruptedException, RemotingTimeoutException,
-        RemotingSendRequestException, RemotingConnectException , MQBrokerException{
+        RemotingSendRequestException, RemotingConnectException , MQBrokerException {
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_SNODE_CLUSTER_INFO, null);
 
         RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis);

client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java

@@ -480,9 +480,9 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
         sendMessageBack(msg, delayLevel, brokerName, this.defaultMQPullConsumer.getConsumerGroup());
     }
 
-    public void updateConsumeOffsetToBroker(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,
+    public void updateConsumeOffsetToSnode(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,
         MQBrokerException, InterruptedException, MQClientException {
-        this.offsetStore.updateConsumeOffsetToBroker(mq, offset, isOneway);
+        this.offsetStore.updateConsumeOffsetToSnode(mq, offset, isOneway);
     }
 
     public void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName, String consumerGroup)

client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPullConsumerTest.java

@@ -77,6 +77,8 @@ public class DefaultMQPullConsumerTest {
         field.set(mQClientFactory, mQClientAPIImpl);
 
         when(mQClientFactory.findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean())).thenReturn(new FindBrokerResult("127.0.0.1:10911", false));
+        when(mQClientFactory.findSnodeAddressInPublish()).thenReturn("127.0.0.1:10911");
+
     }
 
     @After

client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java

@@ -147,6 +147,8 @@ public class DefaultMQPushConsumerTest {
             });
 
         doReturn(new FindBrokerResult("127.0.0.1:10911", false)).when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
+        doReturn("127.0.0.1:10911").when(mQClientFactory).findSnodeAddressInPublish();
+
         Set<MessageQueue> messageQueueSet = new HashSet<MessageQueue>();
         messageQueueSet.add(createPullRequest().getMessageQueue());
         pushConsumer.getDefaultMQPushConsumerImpl().updateTopicSubscribeInfo(topic, messageQueueSet);

client/src/test/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStoreTest.java

@@ -20,7 +20,6 @@ import java.util.Collections;
 import java.util.HashSet;
 import org.apache.rocketmq.client.ClientConfig;
 import org.apache.rocketmq.client.exception.MQBrokerException;
-import org.apache.rocketmq.client.impl.FindBrokerResult;
 import org.apache.rocketmq.client.impl.MQClientAPIImpl;
 import org.apache.rocketmq.client.impl.factory.MQClientInstance;
 import org.apache.rocketmq.common.message.MessageQueue;
@@ -34,7 +33,6 @@ import org.mockito.Mock;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.mockito.stubbing.Answer;
-
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
@@ -58,8 +56,8 @@ public class RemoteBrokerOffsetStoreTest {
         System.setProperty("rocketmq.client.localOffsetStoreDir", System.getProperty("java.io.tmpdir") + ".rocketmq_offsets");
         String clientId = new ClientConfig().buildMQClientId() + "#TestNamespace" + System.currentTimeMillis();
         when(mQClientFactory.getClientId()).thenReturn(clientId);
-        when(mQClientFactory.findBrokerAddressInAdmin(brokerName)).thenReturn(new FindBrokerResult("127.0.0.1", false));
         when(mQClientFactory.getMQClientAPIImpl()).thenReturn(mqClientAPI);
+        when(mQClientFactory.findSnodeAddressInPublish()).thenReturn("127.0.0.1:10911");
     }
 
     @Test

client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java

@@ -106,6 +106,8 @@ public class DefaultMQProducerTest {
         when(mQClientAPIImpl.sendMessage(anyString(), anyString(), any(Message.class), any(SendMessageRequestHeader.class), anyLong(), any(CommunicationMode.class),
             nullable(SendCallback.class), nullable(TopicPublishInfo.class), nullable(MQClientInstance.class), anyInt(), nullable(SendMessageContext.class), any(DefaultMQProducerImpl.class)))
             .thenReturn(createSendResult(SendStatus.SEND_OK));
+        when(mQClientFactory.findSnodeAddressInPublish()).thenReturn("127.0.0.1:10911");
+
     }
 
     @After

snode/src/main/java/org/apache/rocketmq/snode/config/SnodeConfig.java → common/src/main/java/org/apache/rocketmq/common/SnodeConfig.java

@@ -14,21 +14,21 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.snode.config;
+package org.apache.rocketmq.common;
 
 import java.net.InetAddress;
 import java.net.UnknownHostException;
-import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.annotation.ImportantField;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.remoting.common.RemotingUtil;
-import static org.apache.rocketmq.client.ClientConfig.SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY;
 
 public class SnodeConfig {
 
-    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
+    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.SNODE_LOGGER_NAME);
+
+    public static final String SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY = "com.rocketmq.sendMessageWithVIPChannel";
 
     private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
 

snode/src/main/java/org/apache/rocketmq/snode/SnodeController.java

@@ -24,6 +24,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import org.apache.rocketmq.acl.AccessValidator;
+import org.apache.rocketmq.common.SnodeConfig;
 import org.apache.rocketmq.common.ThreadFactoryImpl;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.protocol.RequestCode;
@@ -55,7 +56,6 @@ import org.apache.rocketmq.snode.client.impl.IOTClientManagerImpl;
 import org.apache.rocketmq.snode.client.impl.ProducerManagerImpl;
 import org.apache.rocketmq.snode.client.impl.SlowConsumerServiceImpl;
 import org.apache.rocketmq.snode.client.impl.SubscriptionManagerImpl;
-import org.apache.rocketmq.snode.config.SnodeConfig;
 import org.apache.rocketmq.snode.offset.ConsumerOffsetManager;
 import org.apache.rocketmq.snode.processor.ConsumerManageProcessor;
 import org.apache.rocketmq.snode.processor.DefaultMqttMessageProcessor;

snode/src/main/java/org/apache/rocketmq/snode/SnodeStartup.java

@@ -26,11 +26,13 @@ import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.Option;
 import org.apache.commons.cli.Options;
 import org.apache.commons.cli.PosixParser;
 import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.SnodeConfig;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
@@ -39,7 +41,6 @@ import org.apache.rocketmq.remoting.ServerConfig;
 import org.apache.rocketmq.remoting.common.TlsMode;
 import org.apache.rocketmq.remoting.netty.TlsSystemConfig;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
-import org.apache.rocketmq.snode.config.SnodeConfig;
 import org.apache.rocketmq.srvutil.ServerUtil;
 import org.slf4j.LoggerFactory;
 
@@ -123,19 +124,31 @@ public class SnodeStartup {
             nettyClientConfig,
             snodeConfig);
 
+        boolean initResult = snodeController.initialize();
+        if (!initResult) {
+            snodeController.shutdown();
+            System.exit(-3);
+        }
+
         Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
             private volatile boolean hasShutdown = false;
+            private AtomicInteger shutdownTimes = new AtomicInteger(0);
 
             @Override
             public void run() {
                 synchronized (this) {
+                    log.info("Shutdown hook was invoked, {}", this.shutdownTimes.incrementAndGet());
+
                     if (!this.hasShutdown) {
                         this.hasShutdown = true;
+                        long beginTime = System.currentTimeMillis();
                         snodeController.shutdown();
+                        long consumingTimeTotal = System.currentTimeMillis() - beginTime;
+                        log.info("Shutdown hook over, consuming total time(ms): {}", consumingTimeTotal);
                     }
                 }
             }
-        }));
+        },"ShutdownHook"));
         return snodeController;
     }
 
@@ -143,6 +156,15 @@ public class SnodeStartup {
         Option opt = new Option("c", "configFile", true, "SNode config properties file");
         opt.setRequired(false);
         options.addOption(opt);
+
+        opt = new Option("p", "printConfigItem", false, "Print all config item");
+        opt.setRequired(false);
+        options.addOption(opt);
+
+        opt = new Option("m", "printImportantConfig", false, "Print important config item");
+        opt.setRequired(false);
+        options.addOption(opt);
+        
         return options;
     }
 }

snode/src/main/java/org/apache/rocketmq/snode/client/ClientHousekeepingService.java

@@ -28,7 +28,7 @@ import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
 import org.apache.rocketmq.snode.constant.SnodeConstant;
 
 public class ClientHousekeepingService implements ChannelEventListener {
-    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
+    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.SNODE_LOGGER_NAME);
     private final ClientManager producerManager;
     private final ClientManager consumerManager;
     private final ClientManager iotClientManager;

snode/src/main/java/org/apache/rocketmq/snode/processor/ConsumerManageProcessor.java

@@ -43,7 +43,7 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.snode.SnodeController;
 
 public class ConsumerManageProcessor implements RequestProcessor {
-    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
+    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.SNODE_LOGGER_NAME);
 
     private final SnodeController snodeController;
 

snode/src/main/java/org/apache/rocketmq/snode/processor/DefaultMqttMessageProcessor.java

@@ -28,6 +28,9 @@ import java.io.UnsupportedEncodingException;
 import java.nio.charset.Charset;
 import java.util.HashMap;
 import java.util.Map;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.remoting.RemotingChannel;
 import org.apache.rocketmq.remoting.RequestProcessor;
 import org.apache.rocketmq.remoting.common.RemotingUtil;
@@ -38,6 +41,7 @@ import org.apache.rocketmq.remoting.transport.mqtt.RocketMQMqttConnectPayload;
 import org.apache.rocketmq.snode.processor.mqtthandler.MessageHandler;
 
 public class DefaultMqttMessageProcessor implements RequestProcessor {
+    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.SNODE_LOGGER_NAME);
 
     private Map<MqttMessageType, MessageHandler> type2handler = new HashMap<>();
     private static final int MIN_AVAILABLE_VERSION = 3;

snode/src/main/java/org/apache/rocketmq/snode/service/NnodeService.java

@@ -18,12 +18,12 @@ package org.apache.rocketmq.snode.service;
 
 import java.util.Set;
 import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.SnodeConfig;
 import org.apache.rocketmq.common.protocol.body.ClusterInfo;
 import org.apache.rocketmq.common.protocol.route.TopicRouteData;
 import org.apache.rocketmq.remoting.exception.RemotingConnectException;
 import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
 import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
-import org.apache.rocketmq.snode.config.SnodeConfig;
 
 public interface NnodeService {
     /**

snode/src/main/java/org/apache/rocketmq/snode/service/impl/NnodeServiceImpl.java

@@ -22,6 +22,7 @@ import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.SnodeConfig;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.namesrv.TopAddressing;
 import org.apache.rocketmq.common.protocol.RequestCode;
@@ -37,7 +38,6 @@ import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
 import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.snode.SnodeController;
-import org.apache.rocketmq.snode.config.SnodeConfig;
 import org.apache.rocketmq.snode.constant.SnodeConstant;
 import org.apache.rocketmq.snode.service.NnodeService;
 

snode/src/main/java/org/apache/rocketmq/snode/service/impl/ScheduledServiceImpl.java

@@ -20,6 +20,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import org.apache.rocketmq.common.SnodeConfig;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.protocol.RequestCode;
 import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData;
@@ -28,7 +29,6 @@ import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.snode.SnodeController;
-import org.apache.rocketmq.snode.config.SnodeConfig;
 import org.apache.rocketmq.snode.service.ScheduledService;
 
 public class ScheduledServiceImpl implements ScheduledService {

snode/src/test/java/org/apache/rocketmq/snode/SnodeControllerTest.java

@@ -16,9 +16,9 @@
  */
 package org.apache.rocketmq.snode;
 
+import org.apache.rocketmq.common.SnodeConfig;
 import org.apache.rocketmq.remoting.ClientConfig;
 import org.apache.rocketmq.remoting.ServerConfig;
-import org.apache.rocketmq.snode.config.SnodeConfig;
 import org.junit.Test;
 
 import static org.assertj.core.api.Assertions.assertThat;

snode/src/test/java/org/apache/rocketmq/snode/processor/DefaultMqttMessageProcessorTest.java

@@ -22,6 +22,7 @@ import io.netty.handler.codec.mqtt.MqttMessageType;
 import io.netty.handler.codec.mqtt.MqttQoS;
 import java.io.UnsupportedEncodingException;
 import java.nio.charset.Charset;
+import org.apache.rocketmq.common.SnodeConfig;
 import org.apache.rocketmq.common.protocol.RequestCode;
 import org.apache.rocketmq.remoting.ClientConfig;
 import org.apache.rocketmq.remoting.RemotingChannel;
@@ -33,7 +34,6 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.remoting.transport.mqtt.MqttHeader;
 import org.apache.rocketmq.remoting.transport.mqtt.RocketMQMqttConnectPayload;
 import org.apache.rocketmq.snode.SnodeController;
-import org.apache.rocketmq.snode.config.SnodeConfig;
 import org.apache.rocketmq.snode.processor.mqtthandler.MqttConnectMessageHandler;
 import org.junit.Before;
 import org.junit.Test;

snode/src/test/java/org/apache/rocketmq/snode/processor/MqttConnectMessageHandlerTest.java

@@ -22,11 +22,11 @@ import io.netty.handler.codec.mqtt.MqttConnectVariableHeader;
 import io.netty.handler.codec.mqtt.MqttFixedHeader;
 import io.netty.handler.codec.mqtt.MqttMessageType;
 import io.netty.handler.codec.mqtt.MqttQoS;
+import org.apache.rocketmq.common.SnodeConfig;
 import org.apache.rocketmq.remoting.ClientConfig;
 import org.apache.rocketmq.remoting.RemotingChannel;
 import org.apache.rocketmq.remoting.ServerConfig;
 import org.apache.rocketmq.snode.SnodeController;
-import org.apache.rocketmq.snode.config.SnodeConfig;
 import org.apache.rocketmq.snode.processor.mqtthandler.MqttConnectMessageHandler;
 import org.junit.Test;
 import org.junit.runner.RunWith;

snode/src/test/java/org/apache/rocketmq/snode/processor/SendMessageProcessorTest.java

@@ -18,6 +18,7 @@ package org.apache.rocketmq.snode.processor;
 
 import java.util.concurrent.CompletableFuture;
 import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.SnodeConfig;
 import org.apache.rocketmq.common.protocol.RequestCode;
 import org.apache.rocketmq.common.protocol.ResponseCode;
 import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeaderV2;
@@ -28,7 +29,6 @@ import org.apache.rocketmq.remoting.exception.RemotingCommandException;
 import org.apache.rocketmq.remoting.netty.CodecHelper;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.snode.SnodeController;
-import org.apache.rocketmq.snode.config.SnodeConfig;
 import org.apache.rocketmq.snode.service.EnodeService;
 import org.apache.rocketmq.snode.service.NnodeService;
 import org.junit.Before;

snode/src/test/java/org/apache/rocketmq/snode/service/EnodeServiceImplTest.java

@@ -18,6 +18,7 @@ package org.apache.rocketmq.snode.service;
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+import org.apache.rocketmq.common.SnodeConfig;
 import org.apache.rocketmq.common.TopicConfig;
 import org.apache.rocketmq.common.protocol.RequestCode;
 import org.apache.rocketmq.common.protocol.ResponseCode;
@@ -30,7 +31,6 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.remoting.transport.rocketmq.NettyRemotingClient;
 import org.apache.rocketmq.snode.SnodeController;
 import org.apache.rocketmq.snode.SnodeTestBase;
-import org.apache.rocketmq.snode.config.SnodeConfig;
 import org.apache.rocketmq.snode.service.impl.EnodeServiceImpl;
 import org.apache.rocketmq.store.GetMessageResult;
 import org.apache.rocketmq.store.GetMessageStatus;

snode/src/test/java/org/apache/rocketmq/snode/service/NnodeServiceImplTest.java

@@ -18,6 +18,7 @@ package org.apache.rocketmq.snode.service;
 
 import java.util.ArrayList;
 import java.util.List;
+import org.apache.rocketmq.common.SnodeConfig;
 import org.apache.rocketmq.remoting.ClientConfig;
 import org.apache.rocketmq.remoting.ServerConfig;
 import org.apache.rocketmq.remoting.exception.RemotingConnectException;
@@ -27,7 +28,6 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.remoting.transport.rocketmq.NettyRemotingClient;
 import org.apache.rocketmq.snode.SnodeController;
 import org.apache.rocketmq.snode.SnodeTestBase;
-import org.apache.rocketmq.snode.config.SnodeConfig;
 import org.apache.rocketmq.snode.service.impl.NnodeServiceImpl;
 import org.junit.Before;
 import org.junit.Test;

snode/src/test/java/org/apache/rocketmq/snode/service/SlowConsumerServiceImplTest.java

@@ -16,12 +16,12 @@
  */
 package org.apache.rocketmq.snode.service;
 
+import org.apache.rocketmq.common.SnodeConfig;
 import org.apache.rocketmq.remoting.ClientConfig;
 import org.apache.rocketmq.remoting.ServerConfig;
 import org.apache.rocketmq.snode.SnodeController;
 import org.apache.rocketmq.snode.client.SlowConsumerService;
 import org.apache.rocketmq.snode.client.impl.SlowConsumerServiceImpl;
-import org.apache.rocketmq.snode.config.SnodeConfig;
 import org.apache.rocketmq.snode.offset.ConsumerOffsetManager;
 import org.junit.Before;
 import org.junit.Test;