Fix the incompatibility issue of 4.x client in broker

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

@@ -82,7 +82,12 @@ public class SendMessageProcessor extends AbstractSendMessageProcessor implement
                 this.executeSendMessageHookBefore(ctx, request, mqtraceContext);
 
                 RemotingCommand response;
-                SocketAddress bornHost = RemotingHelper.string2SocketAddress(requestHeader.getBornHost());
+                SocketAddress bornHost = null;
+                if (requestHeader.getBornHost() != null) {
+                    bornHost = RemotingHelper.string2SocketAddress(requestHeader.getBornHost());
+                } else {
+                    bornHost = ctx.channel().remoteAddress();
+                }
 
                 if (requestHeader.isBatch()) {
                     response = this.sendBatchMessage(bornHost, request, mqtraceContext, requestHeader);

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

@@ -326,8 +326,9 @@ public class SnodeController {
         this.snodeServer.registerProcessor(RequestCode.GET_MAX_OFFSET, consumerManageProcessor, this.consumerManageExecutor);
         this.snodeServer.registerProcessor(RequestCode.SEARCH_OFFSET_BY_TIMESTAMP, consumerManageProcessor, this.consumerManageExecutor);
         this.snodeServer.registerProcessor(RequestCode.CREATE_RETRY_TOPIC, consumerManageProcessor, this.consumerManageExecutor);
+        this.snodeServer.registerProcessor(RequestCode.LOCK_BATCH_MQ, consumerManageProcessor, this.consumerManageExecutor);
+        this.snodeServer.registerProcessor(RequestCode.UNLOCK_BATCH_MQ, consumerManageProcessor, this.consumerManageExecutor);
         this.mqttRemotingServer.registerProcessor(RequestCode.MQTT_MESSAGE, defaultMqttMessageProcessor, handleMqttMessageExecutor);
-
     }
 
     public void start() {