[ISSUE #9992] Fix remoting server netty server codec thread reuse problem (#9993)

Co-authored-by: maowei.ymw <maowei.ymw@alibaba-inc.com>

Author: ymwneu

remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java

@@ -272,9 +272,9 @@ public void run(Timeout timeout) {
      */
     protected ChannelPipeline configChannel(SocketChannel ch) {
         return ch.pipeline()
-            .addLast(nettyServerConfig.isServerNettyWorkerGroupEnable() ? defaultEventExecutorGroup : null,
+            .addLast(getDefaultEventExecutorGroup(),
                 HANDSHAKE_HANDLER_NAME, new HandshakeHandler())
-            .addLast(nettyServerConfig.isServerNettyWorkerGroupEnable() ? defaultEventExecutorGroup : null,
+            .addLast(getDefaultEventExecutorGroup(),
                 encoder,
                 new NettyDecoder(),
                 distributionHandler,
@@ -430,7 +430,7 @@ private void printRemotingCodeDistribution() {
     }
 
     public DefaultEventExecutorGroup getDefaultEventExecutorGroup() {
-        return defaultEventExecutorGroup;
+        return nettyServerConfig.isServerNettyWorkerGroupEnable() ? defaultEventExecutorGroup : null;
     }
 
     public NettyEncoder getEncoder() {
@@ -462,11 +462,11 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List<Object> o
                     return;
                 }
                 if (detectionResult.state() == ProtocolDetectionState.DETECTED) {
-                    ctx.pipeline().addAfter(defaultEventExecutorGroup, ctx.name(), HA_PROXY_DECODER, new HAProxyMessageDecoder())
-                        .addAfter(defaultEventExecutorGroup, HA_PROXY_DECODER, HA_PROXY_HANDLER, new HAProxyMessageHandler())
-                        .addAfter(defaultEventExecutorGroup, HA_PROXY_HANDLER, TLS_MODE_HANDLER, tlsModeHandler);
+                    ctx.pipeline().addAfter(getDefaultEventExecutorGroup(), ctx.name(), HA_PROXY_DECODER, new HAProxyMessageDecoder())
+                        .addAfter(getDefaultEventExecutorGroup(), HA_PROXY_DECODER, HA_PROXY_HANDLER, new HAProxyMessageHandler())
+                        .addAfter(getDefaultEventExecutorGroup(), HA_PROXY_HANDLER, TLS_MODE_HANDLER, tlsModeHandler);
                 } else {
-                    ctx.pipeline().addAfter(defaultEventExecutorGroup, ctx.name(), TLS_MODE_HANDLER, tlsModeHandler);
+                    ctx.pipeline().addAfter(getDefaultEventExecutorGroup(), ctx.name(), TLS_MODE_HANDLER, tlsModeHandler);
                 }
 
                 try {
@@ -509,8 +509,8 @@ protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
                     case ENFORCING:
                         if (null != sslContext) {
                             ctx.pipeline()
-                                .addAfter(defaultEventExecutorGroup, TLS_MODE_HANDLER, TLS_HANDLER_NAME, sslContext.newHandler(ctx.channel().alloc()))
-                                .addAfter(defaultEventExecutorGroup, TLS_HANDLER_NAME, FILE_REGION_ENCODER_NAME, new FileRegionEncoder());
+                                .addAfter(getDefaultEventExecutorGroup(), TLS_MODE_HANDLER, TLS_HANDLER_NAME, sslContext.newHandler(ctx.channel().alloc()))
+                                .addAfter(getDefaultEventExecutorGroup(), TLS_HANDLER_NAME, FILE_REGION_ENCODER_NAME, new FileRegionEncoder());
                             log.info("Handlers prepended to channel pipeline to establish SSL connection");
                         } else {
                             ctx.close();