[RIP-11]add SEND_BATCH_MESSAGE request code to related processor.

remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ClientImpl.java

@@ -82,13 +82,13 @@ public class Http2ClientImpl extends NettyRemotingClientAbstract implements Remo
     public RemotingClient init(ClientConfig clientConfig, ChannelEventListener channelEventListener) {
         this.nettyClientConfig = clientConfig;
         this.channelEventListener = channelEventListener;
-        this.ioGroup = new NioEventLoopGroup(clientConfig.getClientWorkerThreads(), ThreadUtils.newGenericThreadFactory("NettyClientEpollIoThreads",
+        this.ioGroup = new NioEventLoopGroup(clientConfig.getClientWorkerThreads(), ThreadUtils.newGenericThreadFactory("Http2NettyClientEpollIoThreads",
             clientConfig.getClientWorkerThreads()));
         this.publicExecutor = ThreadUtils.newFixedThreadPool(
             clientConfig.getClientCallbackExecutorThreads(),
-            10000, "Remoting-PublicExecutor", true);
+            10000, "Http2Remoting-PublicExecutor", true);
         this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(clientConfig.getClientWorkerThreads(),
-            ThreadUtils.newGenericThreadFactory("NettyClientWorkerThreads", clientConfig.getClientWorkerThreads()));
+            ThreadUtils.newGenericThreadFactory("Http2NettyClientWorkerThreads", clientConfig.getClientWorkerThreads()));
         buildHttp2SslClientContext();
         return this;
     }

remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ServerImpl.java

@@ -97,23 +97,23 @@ public class Http2ServerImpl extends NettyRemotingServerAbstract implements Remo
         this.channelEventListener = channelEventListener;
         this.publicExecutor = ThreadUtils.newFixedThreadPool(
             serverConfig.getServerCallbackExecutorThreads(),
-            10000, "Remoting-PublicExecutor", true);
+            10000, "Http2Remoting-PublicExecutor", true);
         if (JvmUtils.isLinux() && this.serverConfig.isUseEpollNativeSelector()) {
-            this.ioGroup = new EpollEventLoopGroup(serverConfig.getServerSelectorThreads(), ThreadUtils.newGenericThreadFactory("NettyEpollIoThreads",
+            this.ioGroup = new EpollEventLoopGroup(serverConfig.getServerSelectorThreads(), ThreadUtils.newGenericThreadFactory("Http2NettyEpollIoThreads",
                 serverConfig.getServerSelectorThreads()));
-            this.bossGroup = new EpollEventLoopGroup(serverConfig.getServerAcceptorThreads(), ThreadUtils.newGenericThreadFactory("NettyBossThreads",
+            this.bossGroup = new EpollEventLoopGroup(serverConfig.getServerAcceptorThreads(), ThreadUtils.newGenericThreadFactory("Http2NettyBossThreads",
                 serverConfig.getServerAcceptorThreads()));
             this.socketChannelClass = EpollServerSocketChannel.class;
         } else {
-            this.bossGroup = new NioEventLoopGroup(serverConfig.getServerAcceptorThreads(), ThreadUtils.newGenericThreadFactory("NettyBossThreads",
+            this.bossGroup = new NioEventLoopGroup(serverConfig.getServerAcceptorThreads(), ThreadUtils.newGenericThreadFactory("Http2NettyBossThreads",
                 serverConfig.getServerAcceptorThreads()));
-            this.ioGroup = new NioEventLoopGroup(serverConfig.getServerSelectorThreads(), ThreadUtils.newGenericThreadFactory("NettyNioIoThreads",
+            this.ioGroup = new NioEventLoopGroup(serverConfig.getServerSelectorThreads(), ThreadUtils.newGenericThreadFactory("Http2NettyNioIoThreads",
                 serverConfig.getServerSelectorThreads()));
             this.socketChannelClass = NioServerSocketChannel.class;
         }
 
         this.workerGroup = new DefaultEventExecutorGroup(serverConfig.getServerWorkerThreads(),
-            ThreadUtils.newGenericThreadFactory("NettyWorkerThreads", serverConfig.getServerWorkerThreads()));
+            ThreadUtils.newGenericThreadFactory("Http2NettyWorkerThreads", serverConfig.getServerWorkerThreads()));
         this.port = nettyServerConfig.getListenPort();
         buildHttp2SslServerContext();
         return this;

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

@@ -51,13 +51,15 @@ public class SendMessageProcessor implements RequestProcessor {
         }
         String enodeName;
         SendMessageRequestHeaderV2 sendMessageRequestHeaderV2 = null;
+        ConsumerSendMsgBackRequestHeader consumerSendMsgBackRequestHeader = null;
         boolean isSendBack = false;
-        if (request.getCode() == RequestCode.SEND_MESSAGE_V2) {
+        if (request.getCode() == RequestCode.SEND_MESSAGE_V2 ||
+            request.getCode() == RequestCode.SEND_BATCH_MESSAGE) {
             sendMessageRequestHeaderV2 = (SendMessageRequestHeaderV2) request.decodeCommandCustomHeader(SendMessageRequestHeaderV2.class);
             enodeName = sendMessageRequestHeaderV2.getN();
         } else {
             isSendBack = true;
-            ConsumerSendMsgBackRequestHeader consumerSendMsgBackRequestHeader = (ConsumerSendMsgBackRequestHeader) request.decodeCommandCustomHeader(ConsumerSendMsgBackRequestHeader.class);
+            consumerSendMsgBackRequestHeader = (ConsumerSendMsgBackRequestHeader) request.decodeCommandCustomHeader(ConsumerSendMsgBackRequestHeader.class);
             enodeName = consumerSendMsgBackRequestHeader.getEnodeName();
         }
 

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

@@ -71,12 +71,18 @@ public class SendMessageProcessorTest {
     }
 
     @Test
-    public void testProcessRequest() throws RemotingCommandException {
+    public void testSendMessageV2ProcessRequest() throws RemotingCommandException {
+        CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
+        RemotingCommand request = createSendMesssageV2Command();
+        when(this.snodeController.getEnodeService().sendMessage(anyString(), any(RemotingCommand.class))).thenReturn(future);
+        sendMessageProcessor.processRequest(remotingChannel, request);
+    }
+
+    @Test
+    public void testSendBatchMessageProcessRequest() throws RemotingCommandException {
         snodeController.setEnodeService(enodeService);
         CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
-        RemotingCommand request = createSendMesssageCommand();
-        SendMessageRequestHeaderV2 sendMessageRequestHeaderV2 = (SendMessageRequestHeaderV2) request.decodeCommandCustomHeader(SendMessageRequestHeaderV2.class);
-        System.out.println("sendMessageRequestHeaderV2: " + sendMessageRequestHeaderV2);
+        RemotingCommand request = createSendBatchMesssageCommand();
         when(this.snodeController.getEnodeService().sendMessage(anyString(), any(RemotingCommand.class))).thenReturn(future);
         sendMessageProcessor.processRequest(remotingChannel, request);
     }
@@ -95,7 +101,7 @@ public class SendMessageProcessorTest {
         return requestHeader;
     }
 
-    private RemotingCommand createSendMesssageCommand() {
+    private RemotingCommand createSendMesssageV2Command() {
         SendMessageRequestHeaderV2 sendMessageRequestHeaderV2 = createSendMsgRequestHeader();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE_V2, sendMessageRequestHeaderV2);
         request.setBody(new byte[] {'a'});
@@ -103,6 +109,14 @@ public class SendMessageProcessorTest {
         return request;
     }
 
+    private RemotingCommand createSendBatchMesssageCommand() {
+        SendMessageRequestHeaderV2 sendMessageRequestHeaderV2 = createSendMsgRequestHeader();
+        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.SEND_BATCH_MESSAGE, sendMessageRequestHeaderV2);
+        request.setBody(new byte[] {'b'});
+        CodecHelper.makeCustomHeaderToNet(request);
+        return request;
+    }
+
     RemotingCommand createSendMessageResponse(int responseCode) {
         return RemotingCommand.createResponseCommand(ResponseCode.SUCCESS, null);
     }