optimize: rm appdata size limit (#4473)

changes/en-us/2.x.md

@@ -13,6 +13,7 @@ Add changes here for all PR submitted to the 2.x branch.
 ### optimize:
 - [[#6061](https://github.com/seata/seata/pull/6061)] merge the rpcMergeMessageSend threads of rm and tm and increase the thread hibernation duration
 - [[#6031](https://github.com/seata/seata/pull/6031)] add a check for the existence of the undolog table
+- [[#4473](https://github.com/seata/seata/pull/4473)] rm appdata size limit
 
 ### security:
 - [[#6069](https://github.com/seata/seata/pull/6069)] Upgrade Guava dependencies to fix security vulnerabilities
@@ -30,5 +31,6 @@ Thanks to these contributors for their code commits. Please report an unintended
 - [imcmai](https://github.com/imcmai)
 - [DroidEye2ONGU](https://github.com/DroidEye2ONGU)
 - [funky-eyes](https://github.com/funky-eyes)
+- [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

@@ -13,6 +13,7 @@
 ### optimize:
 - [[#6061](https://github.com/seata/seata/pull/6061)] 合并rm和tm的rpcMergeMessageSend线程，增加线程休眠时长
 - [[#6031](https://github.com/seata/seata/pull/6031)] 添加undo_log表的存在性校验
+- [[#4473](https://github.com/seata/seata/pull/4473)] rm appdata大小限制
 
 
 ### security:
@@ -31,6 +32,6 @@
 - [imcmai](https://github.com/imcmai)
 - [DroidEye2ONGU](https://github.com/DroidEye2ONGU)
 - [funky-eyes](https://github.com/funky-eyes)
-
+- [Bughue](https://github.com/Bughue)
 
 同时，我们收到了社区反馈的很多有价值的issue和建议，非常感谢大家。

common/src/main/java/io/seata/common/ConfigurationKeys.java

@@ -982,4 +982,23 @@ public interface ConfigurationKeys {
      */
     String ENABLE_PARALLEL_HANDLE_BRANCH_KEY = SERVER_PREFIX + "enableParallelHandleBranch";
 
+    /**
+     * The constant RM_APPLICATION_DATA_SIZE_ERROR
+     */
+    String RM_APPLICATION_DATA_SIZE_LIMIT = CLIENT_RM_PREFIX + "applicationDataLimit";
+
+    /**
+     * The constant RM_APPLICATION_DATA_SIZE_CHECK
+     */
+    String RM_APPLICATION_DATA_SIZE_CHECK = CLIENT_RM_PREFIX + "applicationDataLimitCheck";
+
+    /**
+     * The constant SERVER_APPLICATION_DATA_SIZE_ERROR
+     */
+    String SERVER_APPLICATION_DATA_SIZE_LIMIT = SERVER_PREFIX + "applicationDataLimit";
+
+    /**
+     * The constant SERVER_APPLICATION_DATA_SIZE_CHECK
+     */
+    String SERVER_APPLICATION_DATA_SIZE_CHECK = SERVER_PREFIX + "applicationDataLimitCheck";
 }

common/src/main/java/io/seata/common/DefaultValues.java

@@ -197,6 +197,11 @@ public interface DefaultValues {
      */
     long DEFAULT_RPC_TC_REQUEST_TIMEOUT = Duration.ofSeconds(15).toMillis();
 
+    /**
+     * the constant DEFAULT_APPLICATION_DATA_SIZE_LIMIT
+     */
+    int DEFAULT_APPLICATION_DATA_SIZE_LIMIT = 64000;
+
     /**
      * the constant DEFAULT_XAER_NOTA_RETRY_TIMEOUT
      */

common/src/main/java/io/seata/common/util/StringUtils.java

@@ -20,6 +20,7 @@
 import java.lang.annotation.Annotation;
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
 import java.text.SimpleDateFormat;
 import java.util.Collection;
 import java.util.Date;
@@ -29,6 +30,8 @@
 
 import io.seata.common.Constants;
 import io.seata.common.exception.ShouldNeverHappenException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * The type String utils.
@@ -38,6 +41,7 @@
  */
 public class StringUtils {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(StringUtils.class);
     private static final Pattern CAMEL_PATTERN = Pattern.compile("[A-Z]");
     private static final Pattern LINE_PATTERN = Pattern.compile("-(\\w)");
 
@@ -349,6 +353,29 @@ public static String hump2Line(String str) {
         return sb.toString();
     }
 
+    /**
+     * check string data size
+     *
+     * @param data the str
+     * @param dataName the data name
+     * @param errorSize throw exception if size > errorSize
+     * @return boolean
+     */
+    public static boolean checkDataSize(String data, String dataName, int errorSize, boolean throwIfErr) {
+        if (isBlank(data)) {
+            return true;
+        }
+        int length = data.getBytes(StandardCharsets.UTF_8).length;
+        if (length > errorSize) {
+            LOGGER.warn("{} data is large(errorSize), size={}", dataName, length);
+            if (!throwIfErr) {
+                return false;
+            }
+            throw new IllegalArgumentException(dataName + " data is too large, size=" + length);
+        }
+        return true;
+    }
+
     public static boolean hasLowerCase(String str) {
         if (null == str) {
             return false;

common/src/test/java/io/seata/common/util/StringUtilsTest.java

@@ -380,4 +380,15 @@ public String toString() {
                     ')';
         }
     }
+
+    @Test
+    void checkDataSize() {
+        assertThat(StringUtils.checkDataSize("","testdata",10,false)).isEqualTo(Boolean.TRUE);
+        assertThat(StringUtils.checkDataSize("1234567","testdata",17,false)).isEqualTo(Boolean.TRUE);
+        assertThat(StringUtils.checkDataSize("1234567","testdata",4,false)).isEqualTo(Boolean.FALSE);
+        Assertions.assertThrows(IllegalArgumentException.class, () ->
+                StringUtils.checkDataSize("1234567","testdata",6,true)
+        );
+        assertThat( StringUtils.checkDataSize("1234567","testdata",6,false)).isEqualTo(Boolean.FALSE);
+    }
 }

rm/src/main/java/io/seata/rm/AbstractResourceManager.java

@@ -16,8 +16,12 @@
 package io.seata.rm;
 
 import java.util.concurrent.TimeoutException;
-
+import io.seata.common.ConfigurationKeys;
+import io.seata.common.DefaultValues;
 import io.seata.common.exception.NotSupportYetException;
+import io.seata.common.util.StringUtils;
+import io.seata.config.Configuration;
+import io.seata.config.ConfigurationFactory;
 import io.seata.core.exception.RmTransactionException;
 import io.seata.core.exception.TransactionException;
 import io.seata.core.exception.TransactionExceptionCode;
@@ -43,6 +47,12 @@ public abstract class AbstractResourceManager implements ResourceManager {
 
     protected static final Logger LOGGER = LoggerFactory.getLogger(AbstractResourceManager.class);
 
+    private static final Configuration CONFIG = ConfigurationFactory.getInstance();
+
+    private static int appDataErrSize = CONFIG.getInt(ConfigurationKeys.RM_APPLICATION_DATA_SIZE_LIMIT,
+            DefaultValues.DEFAULT_APPLICATION_DATA_SIZE_LIMIT);
+
+    private static boolean throwDataSizeExp = CONFIG.getBoolean(ConfigurationKeys.RM_APPLICATION_DATA_SIZE_CHECK, false);
     /**
      * registry branch record
      *
@@ -57,6 +67,8 @@ public abstract class AbstractResourceManager implements ResourceManager {
     @Override
     public Long branchRegister(BranchType branchType, String resourceId, String clientId, String xid, String applicationData, String lockKeys) throws TransactionException {
         try {
+            StringUtils.checkDataSize(applicationData, "applicationData", appDataErrSize, throwDataSizeExp);
+
             BranchRegisterRequest request = new BranchRegisterRequest();
             request.setXid(xid);
             request.setLockKey(lockKeys);
@@ -94,6 +106,7 @@ public Long branchRegister(BranchType branchType, String resourceId, String clie
     @Override
     public void branchReport(BranchType branchType, String xid, long branchId, BranchStatus status, String applicationData) throws TransactionException {
         try {
+            StringUtils.checkDataSize(applicationData, "applicationData", appDataErrSize, throwDataSizeExp);
             BranchReportRequest request = new BranchReportRequest();
             request.setXid(xid);
             request.setBranchId(branchId);

script/client/conf/file.conf

@@ -81,6 +81,8 @@ client {
     sqlParserType = "druid"
     branchExecutionTimeoutXA = 60000
     connectionTwoPhaseHoldTimeoutXA = 10000
+    applicationDataLimit = 64000
+    applicationDataLimitCheck = false
   }
   tm {
     commitRetryCount = 5

script/client/spring/application.properties

@@ -40,6 +40,8 @@ seata.client.rm.lock.retry-times=30
 seata.client.rm.lock.retry-policy-branch-rollback-on-conflict=true
 seata.client.rm.branchExecutionTimeoutXA=60000
 seata.client.rm.connectionTwoPhaseHoldTimeoutXA=10000
+seata.client.rm.applicationDataLimit=64000
+seata.client.rm.applicationDataLimitCheck=false
 seata.client.tm.commit-retry-count=5
 seata.client.tm.rollback-retry-count=5
 seata.client.tm.default-global-transaction-timeout=60000

script/client/spring/application.yml

@@ -22,6 +22,8 @@ seata:
       saga-compensate-persist-mode-update: false
       tcc-action-interceptor-order: -2147482648 #Ordered.HIGHEST_PRECEDENCE + 1000
       sql-parser-type: druid
+      applicationDataLimit: 64000
+      applicationDataLimitCheck: false
       lock:
         retry-interval: 10
         retry-times: 30

script/config-center/config.txt

@@ -128,6 +128,8 @@ server.session.branchAsyncQueueSize=5000
 server.session.enableBranchAsyncRemove=false
 server.enableParallelRequestHandle=true
 server.enableParallelHandleBranch=false
+server.applicationDataLimit=64000
+server.applicationDataLimitCheck=false
 
 server.raft.cluster=127.0.0.1:7091,127.0.0.1:7092,127.0.0.1:7093
 server.raft.snapshotInterval=600

seata-spring-autoconfigure/seata-spring-autoconfigure-client/src/main/java/io/seata/spring/boot/autoconfigure/properties/client/RmProperties.java

@@ -31,6 +31,7 @@
 import static io.seata.common.DefaultValues.TCC_ACTION_INTERCEPTOR_ORDER;
 import static io.seata.common.DefaultValues.DEFAULT_XA_BRANCH_EXECUTION_TIMEOUT;
 import static io.seata.common.DefaultValues.DEFAULT_XA_CONNECTION_TWO_PHASE_HOLD_TIMEOUT;
+import static io.seata.common.DefaultValues.DEFAULT_APPLICATION_DATA_SIZE_LIMIT;
 import static io.seata.spring.boot.autoconfigure.StarterConstants.CLIENT_RM_PREFIX;
 
 /**
@@ -53,6 +54,9 @@ public class RmProperties {
     private int connectionTwoPhaseHoldTimeoutXA = DEFAULT_XA_CONNECTION_TWO_PHASE_HOLD_TIMEOUT;
     private String sqlParserType = SqlParserType.SQL_PARSER_TYPE_DRUID;
 
+    private Boolean applicationDataLimitCheck = false;
+    private Integer applicationDataLimit = DEFAULT_APPLICATION_DATA_SIZE_LIMIT;
+
     public int getAsyncCommitBufferLimit() {
         return asyncCommitBufferLimit;
     }
@@ -164,4 +168,19 @@ public void setConnectionTwoPhaseHoldTimeoutXA(int connectionTwoPhaseHoldTimeout
         this.connectionTwoPhaseHoldTimeoutXA = connectionTwoPhaseHoldTimeoutXA;
     }
 
+    public Boolean getApplicationDataLimitCheck() {
+        return applicationDataLimitCheck;
+    }
+
+    public void setApplicationDataLimitCheck(Boolean applicationDataLimitCheck) {
+        this.applicationDataLimitCheck = applicationDataLimitCheck;
+    }
+
+    public Integer getApplicationDataLimit() {
+        return applicationDataLimit;
+    }
+
+    public void setApplicationDataLimit(Integer applicationDataLimit) {
+        this.applicationDataLimit = applicationDataLimit;
+    }
 }

seata-spring-autoconfigure/seata-spring-autoconfigure-server/src/main/java/io/seata/spring/boot/autoconfigure/properties/server/ServerProperties.java

@@ -36,6 +36,9 @@ public class ServerProperties {
     private Integer servicePort;
     private Integer xaerNotaRetryTimeout = 60000;
 
+    private Boolean applicationDataLimitCheck = false;
+    private Integer applicationDataLimit = 64000;
+
     public long getMaxCommitRetryTimeout() {
         return maxCommitRetryTimeout;
     }
@@ -114,4 +117,19 @@ public void setEnableParallelHandleBranch(Boolean enableParallelHandleBranch) {
         this.enableParallelHandleBranch = enableParallelHandleBranch;
     }
 
+    public Boolean getApplicationDataLimitCheck() {
+        return applicationDataLimitCheck;
+    }
+
+    public void setApplicationDataLimitCheck(Boolean applicationDataLimitCheck) {
+        this.applicationDataLimitCheck = applicationDataLimitCheck;
+    }
+
+    public Integer getApplicationDataLimit() {
+        return applicationDataLimit;
+    }
+
+    public void setApplicationDataLimit(Integer applicationDataLimit) {
+        this.applicationDataLimit = applicationDataLimit;
+    }
 }

server/src/main/java/io/seata/server/coordinator/AbstractCore.java

@@ -18,6 +18,11 @@
 import java.io.IOException;
 import java.util.concurrent.TimeoutException;
 
+import io.seata.common.ConfigurationKeys;
+import io.seata.common.DefaultValues;
+import io.seata.common.util.StringUtils;
+import io.seata.config.Configuration;
+import io.seata.config.ConfigurationFactory;
 import io.seata.core.context.RootContext;
 import io.seata.core.exception.BranchTransactionException;
 import io.seata.core.exception.GlobalTransactionException;
@@ -59,13 +64,21 @@ public abstract class AbstractCore implements Core {
 
     protected LockManager lockManager = LockerManagerFactory.getLockManager();
 
+    private static final Configuration CONFIG = ConfigurationFactory.getInstance();
+    private int appDataErrSize ;
+    private boolean throwDataSizeExp ;
+
     protected RemotingServer remotingServer;
 
     public AbstractCore(RemotingServer remotingServer) {
         if (remotingServer == null) {
             throw new IllegalArgumentException("remotingServer must be not null");
         }
         this.remotingServer = remotingServer;
+        this.appDataErrSize = CONFIG.getInt(ConfigurationKeys.SERVER_APPLICATION_DATA_SIZE_LIMIT,
+            DefaultValues.DEFAULT_APPLICATION_DATA_SIZE_LIMIT);
+        this.throwDataSizeExp = CONFIG.getBoolean(ConfigurationKeys.SERVER_APPLICATION_DATA_SIZE_CHECK, false);
+
     }
 
     public abstract BranchType getHandleBranchType();
@@ -74,6 +87,13 @@ public AbstractCore(RemotingServer remotingServer) {
     public Long branchRegister(BranchType branchType, String resourceId, String clientId, String xid,
                                String applicationData, String lockKeys) throws TransactionException {
         GlobalSession globalSession = assertGlobalSessionNotNull(xid, false);
+        try {
+            StringUtils.checkDataSize(applicationData, "applicationData", appDataErrSize, throwDataSizeExp);
+        } catch (RuntimeException e) {
+            throw new BranchTransactionException(TransactionExceptionCode.FailedToAddBranch,
+                    String.format("Failed to store branch xid = %s ", globalSession.getXid()), e);
+        }
+
         return SessionHolder.lockAndExecute(globalSession, () -> {
             globalSessionStatusCheck(globalSession);
             BranchSession branchSession = SessionHelper.newBranchByGlobal(globalSession, branchType, resourceId,
@@ -90,7 +110,7 @@ public Long branchRegister(BranchType branchType, String resourceId, String clie
             }
             if (LOGGER.isInfoEnabled()) {
                 LOGGER.info("Register branch successfully, xid = {}, branchId = {}, resourceId = {} ,lockKeys = {}",
-                        globalSession.getXid(), branchSession.getBranchId(), resourceId, lockKeys);
+                    globalSession.getXid(), branchSession.getBranchId(), resourceId, lockKeys);
             }
             return branchSession.getBranchId();
         });

server/src/main/resources/application.example.yml

@@ -140,6 +140,8 @@ seata:
     retry-dead-threshold: 130000
     xaer-nota-retry-timeout: 60000
     enableParallelRequestHandle: true
+    applicationDataLimitCheck: true
+    applicationDataLimit: 64000
     recovery:
       committing-retry-period: 1000
       async-committing-retry-period: 1000

server/src/main/resources/application.raft.example.yml

@@ -95,6 +95,8 @@ seata:
     retry-dead-threshold: 130000
     xaer-nota-retry-timeout: 60000
     enableParallelRequestHandle: true
+    applicationDataLimitCheck: true
+    applicationDataLimit: 64000
     recovery:
       committing-retry-period: 1000
       async-committing-retry-period: 1000