optimize: Only AT mode try to get channel with other app (#5153)

apache · Jan 17, 2023 · a84bbcc · a84bbcc
1 parent d19cfee
commit a84bbcc
Show file tree

Hide file tree

Showing 8 changed files with 21 additions and 8 deletions.
diff --git a/changes/en-us/develop.md b/changes/en-us/develop.md
@@ -18,6 +18,7 @@ Add changes here for all PR submitted to the develop branch.
 - [[#5208](https://github.com/seata/seata/pull/5208)] optimize throwable getCause once more
 - [[#5212](https://github.com/seata/seata/pull/5212)] optimize log message level
 - [[#5237](https://github.com/seata/seata/pull/5237)] optimize exception log message print(EnhancedServiceLoader.loadFile#cahtch)
+- [[#5153](https://github.com/seata/seata/pull/5153)] Only AT mode try to get channel with other app
 
 
 ### test:
@@ -32,5 +33,6 @@ Thanks to these contributors for their code commits. Please report an unintended
 - [PeppaO](https://github.com/PeppaO)
 - [yuruixin](https://github.com/yuruixin)
 - [xingfudeshi](https://github.com/xingfudeshi)
+- [Bughue](https://github.com/Bughue)
 
 Also, we receive many valuable issues, questions and advices from our community. Thanks for you all.
diff --git a/changes/zh-cn/develop.md b/changes/zh-cn/develop.md
@@ -17,6 +17,7 @@
 - [[#5208](https://github.com/seata/seata/pull/5208)] 优化多次重复获取Throwable#getCause问题
 - [[#5212](https://github.com/seata/seata/pull/5212)] 优化不合理的日志信息级别
 - [[#5237](https://github.com/seata/seata/pull/5237)] 优化异常日志打印(EnhancedServiceLoader.loadFile#cahtch)
+- [[#5153](https://github.com/seata/seata/pull/5153)] 只允许AT去尝试跨RM获取channel
 
 
 ### test:
@@ -31,5 +32,6 @@
 - [PeppaO](https://github.com/PeppaO)
 - [yuruixin](https://github.com/yuruixin)
 - [xingfudeshi](https://github.com/xingfudeshi)
+- [Bughue](https://github.com/Bughue)
 
 同时，我们收到了社区反馈的很多有价值的issue和建议，非常感谢大家。
diff --git a/core/src/main/java/io/seata/core/rpc/RemotingServer.java b/core/src/main/java/io/seata/core/rpc/RemotingServer.java
@@ -37,10 +37,11 @@ public interface RemotingServer {
      * @param resourceId rm client resourceId
      * @param clientId   rm client id
      * @param msg        transaction message {@code io.seata.core.protocol}
+     * @param tryOtherApp   try other app
      * @return client result message
      * @throws TimeoutException TimeoutException
      */
-    Object sendSyncRequest(String resourceId, String clientId, Object msg) throws TimeoutException;
+    Object sendSyncRequest(String resourceId, String clientId, Object msg, boolean tryOtherApp) throws TimeoutException;
 
     /**
      * server send sync request.

diff --git a/core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemotingServer.java b/core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemotingServer.java
@@ -63,8 +63,9 @@ public AbstractNettyRemotingServer(ThreadPoolExecutor messageExecutor, NettyServ
     }
 
     @Override
-    public Object sendSyncRequest(String resourceId, String clientId, Object msg) throws TimeoutException {
-        Channel channel = ChannelManager.getChannel(resourceId, clientId);
+    public Object sendSyncRequest(String resourceId, String clientId, Object msg, boolean tryOtherApp)
+        throws TimeoutException {
+        Channel channel = ChannelManager.getChannel(resourceId, clientId, tryOtherApp);
         if (channel == null) {
             throw new RuntimeException("rm client is not connected. dbkey:" + resourceId + ",clientId:" + clientId);
         }

diff --git a/core/src/main/java/io/seata/core/rpc/netty/ChannelManager.java b/core/src/main/java/io/seata/core/rpc/netty/ChannelManager.java
@@ -276,7 +276,7 @@ private static Channel getChannelFromSameClientMap(Map<Integer, RpcContext> clie
      * @param clientId   Client ID - ApplicationId:IP:Port
      * @return Corresponding channel, NULL if not found.
      */
-    public static Channel getChannel(String resourceId, String clientId) {
+    public static Channel getChannel(String resourceId, String clientId, boolean tryOtherApp) {
         Channel resultChannel = null;
 
         String[] clientIdInfo = readClientId(clientId);
@@ -381,7 +381,7 @@ public static Channel getChannel(String resourceId, String clientId) {
             }
         }
 
-        if (resultChannel == null) {
+        if (resultChannel == null && tryOtherApp) {
             resultChannel = tryOtherApp(applicationIdMap, targetApplicationId);
 
             if (resultChannel == null) {

diff --git a/server/src/main/java/io/seata/server/coordinator/AbstractCore.java b/server/src/main/java/io/seata/server/coordinator/AbstractCore.java
@@ -172,8 +172,9 @@ public BranchStatus branchCommit(GlobalSession globalSession, BranchSession bran
 
     protected BranchStatus branchCommitSend(BranchCommitRequest request, GlobalSession globalSession,
                                             BranchSession branchSession) throws IOException, TimeoutException {
+
         BranchCommitResponse response = (BranchCommitResponse) remotingServer.sendSyncRequest(
-                branchSession.getResourceId(), branchSession.getClientId(), request);
+            branchSession.getResourceId(), branchSession.getClientId(), request, branchSession.isAT());
         return response.getBranchStatus();
     }
 
@@ -196,8 +197,9 @@ public BranchStatus branchRollback(GlobalSession globalSession, BranchSession br
 
     protected BranchStatus branchRollbackSend(BranchRollbackRequest request, GlobalSession globalSession,
                                               BranchSession branchSession) throws IOException, TimeoutException {
+
         BranchRollbackResponse response = (BranchRollbackResponse) remotingServer.sendSyncRequest(
-                branchSession.getResourceId(), branchSession.getClientId(), request);
+            branchSession.getResourceId(), branchSession.getClientId(), request, branchSession.isAT());
         return response.getBranchStatus();
     }
 

diff --git a/server/src/main/java/io/seata/server/session/BranchSession.java b/server/src/main/java/io/seata/server/session/BranchSession.java
@@ -297,6 +297,10 @@ public boolean unlock() throws TransactionException {
         return true;
     }
 
+    public boolean isAT() {
+        return this.getBranchType() == BranchType.AT;
+    }
+
     public LockStatus getLockStatus() {
         return lockStatus;
     }

diff --git a/server/src/test/java/io/seata/server/coordinator/DefaultCoordinatorTest.java b/server/src/test/java/io/seata/server/coordinator/DefaultCoordinatorTest.java
@@ -249,7 +249,8 @@ static Stream<Arguments> xidAndBranchIdProviderForRollback() throws Exception {
     public static class MockServerMessageSender implements RemotingServer {
 
         @Override
-        public Object sendSyncRequest(String resourceId, String clientId, Object message) throws TimeoutException {
+        public Object sendSyncRequest(String resourceId, String clientId, Object message, boolean tryOtherApp)
+            throws TimeoutException {
             if (message instanceof BranchCommitRequest) {
                 final BranchCommitResponse branchCommitResponse = new BranchCommitResponse();
                 branchCommitResponse.setBranchStatus(BranchStatus.PhaseTwo_Committed);