undo log skip

Bughue · Nov 7, 2024 · a2cad95 · a2cad95
1 parent 6cfdf72
commit a2cad95
Show file tree

Hide file tree

Showing 5 changed files with 30 additions and 4 deletions.
diff --git a/core/src/main/java/org/apache/seata/core/protocol/Version.java b/core/src/main/java/org/apache/seata/core/protocol/Version.java
@@ -45,6 +45,7 @@ public class Version {
     /**
      * The constant VERSION_MAP.
      */
+    public static final Map<String, String> VERSION_MAP = new ConcurrentHashMap<>();
 
     private Version() {
 
@@ -59,6 +60,26 @@ public static String getCurrent() {
         return CURRENT;
     }
 
+    /**
+     * Put channel version.
+     *
+     * @param c the c
+     * @param v the v
+     */
+    public static void putChannelVersion(Channel c, String v) {
+        VERSION_MAP.put(NetUtil.toStringAddress(c.remoteAddress()), v);
+    }
+
+    /**
+     * Gets channel version.
+     *
+     * @param c the c
+     * @return the channel version
+     */
+    public static String getChannelVersion(Channel c) {
+        return VERSION_MAP.get(NetUtil.toStringAddress(c.remoteAddress()));
+    }
+
     /**
      * Determine whether the client version is greater than or equal to version 1.5.0
      *

diff --git a/core/src/main/java/org/apache/seata/core/rpc/MsgVersionHelper.java b/core/src/main/java/org/apache/seata/core/rpc/MsgVersionHelper.java
@@ -36,15 +36,15 @@ public class MsgVersionHelper {
     private static final List<Short> SKIP_MSG_CODE_V0 = Arrays.asList(MessageType.TYPE_RM_DELETE_UNDOLOG);
 
     public static boolean versionNotSupport(Channel channel, RpcMessage rpcMessage){
-        if(rpcMessage==null || rpcMessage.getBody() == null){
+        if(rpcMessage==null || rpcMessage.getBody() == null || channel == null){
             return false;
         }
         Object msg = rpcMessage.getBody();
-        RpcContext rpcContext = ChannelManager.getContextFromIdentified(channel);
-        if (rpcContext == null || StringUtils.isBlank(rpcContext.getVersion()) || msg == null) {
+        String version = Version.getChannelVersion(channel);
+        if (StringUtils.isBlank(version) || msg == null) {
             return false;
         }
-        boolean aboveV0 = Version.isAboveOrEqualVersion071(rpcContext.getVersion());
+        boolean aboveV0 = Version.isAboveOrEqualVersion071(version);
         if(aboveV0 || !(msg instanceof MessageTypeAware)){
             return false;
         }

diff --git a/core/src/main/java/org/apache/seata/core/rpc/processor/server/RegRmProcessor.java b/core/src/main/java/org/apache/seata/core/rpc/processor/server/RegRmProcessor.java
@@ -65,6 +65,7 @@ private void onRegRmMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) {
         try {
             if (null == checkAuthHandler || checkAuthHandler.regResourceManagerCheckAuth(message)) {
                 ChannelManager.registerRMChannel(message, ctx.channel());
+                Version.putChannelVersion(ctx.channel(), message.getVersion());
                 isSuccess = true;
                 if (LOGGER.isDebugEnabled()) {
                     LOGGER.debug("RM checkAuth for client:{},vgroup:{},applicationId:{} is OK", ipAndPort, message.getTransactionServiceGroup(), message.getApplicationId());

diff --git a/core/src/main/java/org/apache/seata/core/rpc/processor/server/RegTmProcessor.java b/core/src/main/java/org/apache/seata/core/rpc/processor/server/RegTmProcessor.java
@@ -60,11 +60,13 @@ public void process(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exc
     private void onRegTmMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) {
         RegisterTMRequest message = (RegisterTMRequest) rpcMessage.getBody();
         String ipAndPort = NetUtil.toStringAddress(ctx.channel().remoteAddress());
+        Version.putChannelVersion(ctx.channel(), message.getVersion());
         boolean isSuccess = false;
         String errorInfo = StringUtils.EMPTY;
         try {
             if (null == checkAuthHandler || checkAuthHandler.regTransactionManagerCheckAuth(message)) {
                 ChannelManager.registerTMChannel(message, ctx.channel());
+                Version.putChannelVersion(ctx.channel(), message.getVersion());
                 isSuccess = true;
                 if (LOGGER.isDebugEnabled()) {
                     LOGGER.debug("TM checkAuth for client:{},vgroup:{},applicationId:{} is OK",

diff --git a/...ock-server/src/main/java/org/apache/seata/mockserver/processor/MockRegisterProcessor.java b/...ock-server/src/main/java/org/apache/seata/mockserver/processor/MockRegisterProcessor.java
@@ -54,6 +54,7 @@ public void process(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exc
             LOGGER.info("message = " + message);
 
             ChannelManager.registerTMChannel(message, ctx.channel());
+            Version.putChannelVersion(ctx.channel(), message.getVersion());
 
             RegisterTMResponse resp = new RegisterTMResponse();
             remotingServer.sendAsyncResponse(rpcMessage, ctx.channel(), resp);
@@ -63,6 +64,7 @@ public void process(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exc
             LOGGER.info("message = " + message);
 
             ChannelManager.registerRMChannel(message, ctx.channel());
+            Version.putChannelVersion(ctx.channel(), message.getVersion());
 
             RegisterRMResponse resp = new RegisterRMResponse();
             remotingServer.sendAsyncResponse(rpcMessage, ctx.channel(), resp);