optimize: load SeataSerializer by version (#6208)

Author: Bughue

changes/en-us/2.x.md

@@ -11,6 +11,7 @@ Add changes here for all PR submitted to the 2.x branch.
 
 ### optimize:
 - [[#6499](https://github.com/apache/incubator-seata/pull/6499)] split the task thread pool for committing and rollbacking statuses
+- [[#6208](https://github.com/apache/incubator-seata/pull/6208)] optimize : load SeataSerializer by version
 
 ### refactor:
 - [[#6534](https://github.com/apache/incubator-seata/pull/6534)] optimize: send async response
@@ -31,5 +32,6 @@ Thanks to these contributors for their code commits. Please report an unintended
 - [tuwenlin](https://github.com/tuwenlin)
 - [YeonCheolGit](https://github.com/YeonCheolGit)
 - [liuqiufeng](https://github.com/liuqiufeng)
+- [Bughue](https://github.com/Bughue)
 
 Also, we receive many valuable issues, questions and advices from our community. Thanks for you all.

changes/zh-cn/2.x.md

@@ -10,6 +10,8 @@
 
 ### optimize:
 - [[#6499](https://github.com/apache/incubator-seata/pull/6499)] 拆分 committing 和 rollbacking 状态的任务线程池
+- [[#6208](https://github.com/apache/incubator-seata/pull/6208)] 支持多版本的Seata序列化
+
 
 ### refactor:
 - [[#6534](https://github.com/apache/incubator-seata/pull/6534)] 优化: 发送异步响应
@@ -27,5 +29,6 @@
 - [tuwenlin](https://github.com/tuwenlin)
 - [YeonCheolGit](https://github.com/YeonCheolGit)
 - [liuqiufeng](https://github.com/liuqiufeng)
+- [Bughue](https://github.com/Bughue)
 
 同时，我们收到了社区反馈的很多有价值的issue和建议，非常感谢大家。

core/src/main/java/org/apache/seata/core/protocol/ProtocolConstants.java

@@ -32,10 +32,20 @@ public interface ProtocolConstants {
      */
     byte[] MAGIC_CODE_BYTES = {(byte) 0xda, (byte) 0xda};
 
+    /**
+     * Old protocol version
+     */
+    byte VERSION_0 = 0;
+
+    /**
+     * Protocol version
+     */
+    byte VERSION_1 = 1;
+
     /**
      * Protocol version
      */
-    byte VERSION = 1;
+    byte VERSION = VERSION_1;
 
     /**
      * Max frame length

core/src/main/java/org/apache/seata/core/rpc/netty/v1/ProtocolV1Decoder.java

@@ -152,7 +152,7 @@ public Object decodeFrame(ByteBuf frame) {
                 bs = compressor.decompress(bs);
                 SerializerType protocolType = SerializerType.getByCode(rpcMessage.getCodec());
                 if (this.supportDeSerializerTypes.contains(protocolType)) {
-                    Serializer serializer = SerializerServiceLoader.load(protocolType);
+                    Serializer serializer = SerializerServiceLoader.load(protocolType, ProtocolConstants.VERSION_1);
                     rpcMessage.setBody(serializer.deserialize(bs));
                 } else {
                     throw new IllegalArgumentException("SerializerType not match: " + protocolType.name());

core/src/main/java/org/apache/seata/core/rpc/netty/v1/ProtocolV1Encoder.java

@@ -93,7 +93,7 @@ public void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) {
                 if (messageType != ProtocolConstants.MSGTYPE_HEARTBEAT_REQUEST
                         && messageType != ProtocolConstants.MSGTYPE_HEARTBEAT_RESPONSE) {
                     // heartbeat has no body
-                    Serializer serializer = SerializerServiceLoader.load(SerializerType.getByCode(rpcMessage.getCodec()));
+                    Serializer serializer = SerializerServiceLoader.load(SerializerType.getByCode(rpcMessage.getCodec()), ProtocolConstants.VERSION_1);
                     bodyBytes = serializer.serialize(rpcMessage.getBody());
                     Compressor compressor = CompressorFactory.getCompressor(rpcMessage.getCompressor());
                     bodyBytes = compressor.compress(bodyBytes);

core/src/main/java/org/apache/seata/core/serializer/SerializerServiceLoader.java

@@ -35,23 +35,26 @@
 import static org.apache.seata.core.serializer.SerializerType.PROTOBUF;
 import static org.apache.seata.core.serializer.SerializerType.SEATA;
 
+import java.util.HashMap;
+import java.util.Map;
+
 /**
  * The Service Loader for the interface {@link Serializer}
- *
  */
 public final class SerializerServiceLoader {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(SerializerServiceLoader.class);
     private static final Configuration CONFIG = ConfigurationFactory.getInstance();
 
-    private static final SerializerType[] DEFAULT_SERIALIZER_TYPE = new SerializerType[] {SEATA, PROTOBUF, KRYO, HESSIAN};
+    private static final SerializerType[] DEFAULT_SERIALIZER_TYPE = new SerializerType[]{SEATA, PROTOBUF, KRYO, HESSIAN};
+
+    private final static Map<String, Serializer> SERIALIZER_MAP = new HashMap<>();
 
     private static final String SPLIT_CHAR = ",";
 
     private SerializerServiceLoader() {
     }
 
-
     private static final String PROTOBUF_SERIALIZER_CLASS_NAME = "org.apache.seata.serializer.protobuf.ProtobufSerializer";
 
     /**
@@ -61,7 +64,7 @@ private SerializerServiceLoader() {
      * @return the service of {@link Serializer}
      * @throws EnhancedServiceNotFoundException the enhanced service not found exception
      */
-    public static Serializer load(SerializerType type) throws EnhancedServiceNotFoundException {
+    public static Serializer load(SerializerType type, byte version) throws EnhancedServiceNotFoundException {
         if (type == SerializerType.PROTOBUF) {
             try {
                 ReflectionUtil.getClassByName(PROTOBUF_SERIALIZER_CLASS_NAME);
@@ -70,9 +73,28 @@ public static Serializer load(SerializerType type) throws EnhancedServiceNotFoun
                         "Please manually reference 'org.apache.seata:seata-serializer-protobuf' dependency ", e);
             }
         }
-        return EnhancedServiceLoader.load(Serializer.class, type.name());
+
+        String key = serialzerKey(type, version);
+        Serializer serializer = SERIALIZER_MAP.get(key);
+        if (serializer == null) {
+            if (type == SerializerType.SEATA) {
+                serializer = EnhancedServiceLoader.load(Serializer.class, type.name(), new Object[]{version});
+            } else {
+                serializer = EnhancedServiceLoader.load(Serializer.class, type.name());
+            }
+            SERIALIZER_MAP.put(key, serializer);
+        }
+        return serializer;
     }
 
+    private static String serialzerKey(SerializerType type, byte version) {
+        if (type == SerializerType.SEATA) {
+            return type.name() + version;
+        }
+        return type.name();
+    }
+
+
     public static List<SerializerType> getSupportedSerializers() {
         List<SerializerType> supportedSerializers = new ArrayList<>();
         String defaultSupportSerializers = Arrays.stream(DEFAULT_SERIALIZER_TYPE).map(SerializerType::name).collect(Collectors.joining(SPLIT_CHAR));
@@ -93,4 +115,4 @@ public static SerializerType getDefaultSerializerType() {
         return getSupportedSerializers().get(0);
     }
 
-}
+}

serializer/seata-serializer-seata/src/main/java/org/apache/seata/serializer/seata/MessageCodecFactory.java

@@ -80,7 +80,6 @@
 
 /**
  * The type Message codec factory.
- *
  */
 public class MessageCodecFactory {
 
@@ -95,8 +94,8 @@ public class MessageCodecFactory {
      * @param abstractMessage the abstract message
      * @return the message codec
      */
-    public static MessageSeataCodec getMessageCodec(AbstractMessage abstractMessage) {
-        return getMessageCodec(abstractMessage.getTypeCode());
+    public static MessageSeataCodec getMessageCodec(AbstractMessage abstractMessage, byte version) {
+        return getMessageCodec(abstractMessage.getTypeCode(), version);
     }
 
     /**
@@ -105,14 +104,14 @@ public static MessageSeataCodec getMessageCodec(AbstractMessage abstractMessage)
      * @param typeCode the type code
      * @return the msg instance by code
      */
-    public static MessageSeataCodec getMessageCodec(short typeCode) {
+    public static MessageSeataCodec getMessageCodec(short typeCode, byte version) {
         MessageSeataCodec msgCodec = null;
         switch (typeCode) {
             case MessageType.TYPE_SEATA_MERGE:
-                msgCodec = new MergedWarpMessageCodec();
+                msgCodec = new MergedWarpMessageCodec(version);
                 break;
             case MessageType.TYPE_SEATA_MERGE_RESULT:
-                msgCodec = new MergeResultMessageCodec();
+                msgCodec = new MergeResultMessageCodec(version);
                 break;
             case MessageType.TYPE_REG_CLT:
                 msgCodec = new RegisterTMRequestCodec();
@@ -136,7 +135,7 @@ public static MessageSeataCodec getMessageCodec(short typeCode) {
                 msgCodec = new GlobalReportRequestCodec();
                 break;
             case MessageType.TYPE_BATCH_RESULT_MSG:
-                msgCodec = new BatchResultMessageCodec();
+                msgCodec = new BatchResultMessageCodec(version);
                 break;
             default:
                 break;
@@ -147,15 +146,15 @@ public static MessageSeataCodec getMessageCodec(short typeCode) {
         }
 
         try {
-            msgCodec = getMergeRequestMessageSeataCodec(typeCode);
+            msgCodec = getMergeRequestMessageSeataCodec(typeCode, version);
         } catch (Exception exx) {
         }
 
         if (msgCodec != null) {
             return msgCodec;
         }
 
-        msgCodec = getMergeResponseMessageSeataCodec(typeCode);
+        msgCodec = getMergeResponseMessageSeataCodec(typeCode, version);
 
         return msgCodec;
     }
@@ -166,7 +165,7 @@ public static MessageSeataCodec getMessageCodec(short typeCode) {
      * @param typeCode the type code
      * @return the merge request instance by code
      */
-    protected static MessageSeataCodec getMergeRequestMessageSeataCodec(int typeCode) {
+    protected static MessageSeataCodec getMergeRequestMessageSeataCodec(int typeCode, byte version) {
         switch (typeCode) {
             case MessageType.TYPE_GLOBAL_BEGIN:
                 return new GlobalBeginRequestCodec();
@@ -195,7 +194,7 @@ protected static MessageSeataCodec getMergeRequestMessageSeataCodec(int typeCode
      * @param typeCode the type code
      * @return the merge response instance by code
      */
-    protected static MessageSeataCodec getMergeResponseMessageSeataCodec(int typeCode) {
+    protected static MessageSeataCodec getMergeResponseMessageSeataCodec(int typeCode, byte version) {
         switch (typeCode) {
             case MessageType.TYPE_GLOBAL_BEGIN_RESULT:
                 return new GlobalBeginResponseCodec();