[ISSUE #5684] request code should be short, to support serializeTypeCurrentRPC=ROCKETMQ

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

@@ -113,7 +113,7 @@ private RemotingCommand processRequest(final Channel channel, RemotingCommand re
         BatchAckMessageRequestBody reqBody = null;
         final RemotingCommand response = RemotingCommand.createResponseCommand(ResponseCode.SUCCESS, null);
         response.setOpaque(request.getOpaque());
-        if (request.getCode() == RequestCode.ACK_MESSAGE) {
+        if (request.getCode() == RequestCode.ACK_MESSAGE || request.getCode() == RequestCode.ACK_MESSAGE_S) {
             requestHeader = (AckMessageRequestHeader) request.decodeCommandCustomHeader(AckMessageRequestHeader.class);
 
             TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
@@ -145,7 +145,7 @@ private RemotingCommand processRequest(final Channel channel, RemotingCommand re
             }
 
             appendAck(requestHeader, null, response, channel, null);
-        } else if (request.getCode() == RequestCode.BATCH_ACK_MESSAGE) {
+        } else if (request.getCode() == RequestCode.BATCH_ACK_MESSAGE || request.getCode() == RequestCode.BATCH_ACK_MESSAGE_S) {
             if (request.getBody() != null) {
                 reqBody = BatchAckMessageRequestBody.decode(request.getBody(), BatchAckMessageRequestBody.class);
             }

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

@@ -338,6 +338,7 @@ public void registerProcessor(int requestCode, NettyRequestProcessor processor,
 
         Pair<NettyRequestProcessor, ExecutorService> pair = new Pair<>(processor, executorThis);
         this.processorTable.put(requestCode, pair);
+        this.processorTable.putIfAbsent((int) (short) requestCode, pair);
     }
 
     @Override
@@ -666,6 +667,7 @@ public void registerProcessor(final int requestCode, final NettyRequestProcessor
 
             Pair<NettyRequestProcessor, ExecutorService> pair = new Pair<>(processor, executorThis);
             this.processorTable.put(requestCode, pair);
+            this.processorTable.putIfAbsent((int) (short) requestCode, pair);
         }
 
         @Override

remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RequestCode.java

@@ -87,12 +87,19 @@ public class RequestCode {
     public static final int GET_TIMER_METRICS = 61;
 
     public static final int POP_MESSAGE = 200050;
+    public static final int POP_MESSAGE_S = (short) POP_MESSAGE;
     public static final int ACK_MESSAGE = 200051;
+    public static final int ACK_MESSAGE_S = (short) ACK_MESSAGE;
     public static final int BATCH_ACK_MESSAGE = 200151;
+    public static final int BATCH_ACK_MESSAGE_S = (short) BATCH_ACK_MESSAGE;
     public static final int PEEK_MESSAGE = 200052;
+    public static final int PEEK_MESSAGE_S = (short) PEEK_MESSAGE;
     public static final int CHANGE_MESSAGE_INVISIBLETIME = 200053;
+    public static final int CHANGE_MESSAGE_INVISIBLETIME_S = (short) CHANGE_MESSAGE_INVISIBLETIME;
     public static final int NOTIFICATION = 200054;
+    public static final int NOTIFICATION_S = (short) NOTIFICATION;
     public static final int POLLING_INFO = 200055;
+    public static final int POLLING_INFO_S = (short) POLLING_INFO;
 
     public static final int PUT_KV_CONFIG = 100;
 

remoting/src/test/java/org/apache/rocketmq/remoting/protocol/RequestCodeTest.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.remoting.protocol;
+
+import java.lang.reflect.Field;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class RequestCodeTest {
+
+    @Test
+    public void testRequestCodeShouldBeShort() throws IllegalAccessException {
+        Set<String> ignored = new HashSet<>();
+        ignored.add("POP_MESSAGE");
+        ignored.add("ACK_MESSAGE");
+        ignored.add("BATCH_ACK_MESSAGE");
+        ignored.add("PEEK_MESSAGE");
+        ignored.add("CHANGE_MESSAGE_INVISIBLETIME");
+        ignored.add("NOTIFICATION");
+        ignored.add("POLLING_INFO");
+
+        Map<Integer, String> occupied = new HashMap<>();
+        Class clazz = RequestCode.class;
+        Field[] fields = clazz.getFields();
+        for (Field field : fields) {
+            if (ignored.contains(field.getName())) {
+                continue;
+            }
+            if (field.getInt(clazz) > Short.MAX_VALUE) {
+                Assert.fail(field.getName() + "=" + field.getInt(clazz) + " should be short, to support serializeTypeCurrentRPC=ROCKETMQ");
+            }
+            String name = occupied.get(field.getInt(clazz));
+            if (name != null) {
+                Assert.fail(field.getName() + "=" + field.getInt(clazz) + " is occupied by " + name);
+            }
+            occupied.put(field.getInt(clazz), field.getName());
+        }
+    }
+}