feature: add single server rate limit

build/pom.xml

@@ -89,6 +89,7 @@
         <kafka-clients.version>3.6.1</kafka-clients.version>
         <snakeyaml.version>2.0</snakeyaml.version>
         <jackson-mapper.version>1.9.13</jackson-mapper.version>
+        <bucket4j.version>8.1.0</bucket4j.version>
 
         <!-- For test -->
         <junit-jupiter.version>5.8.2</junit-jupiter.version>

common/src/main/java/org/apache/seata/common/ConfigurationKeys.java

@@ -1110,4 +1110,29 @@ public interface ConfigurationKeys {
      * The constant META_PREFIX
      */
     String META_PREFIX = SEATA_FILE_ROOT_CONFIG + FILE_CONFIG_SPLIT_CHAR + FILE_ROOT_REGISTRY + FILE_CONFIG_SPLIT_CHAR + "metadata.";
+
+    /**
+     * The constant RATE_LIMIT_PREFIX.
+     */
+    String RATE_LIMIT_PREFIX = SERVER_PREFIX + "ratelimit";
+
+    /**
+     * The constant RATE_LIMIT_BUCKET_TOKEN_NUM_PER_SECOND.
+     */
+    String RATE_LIMIT_BUCKET_TOKEN_NUM_PER_SECOND = RATE_LIMIT_PREFIX + ".bucketTokenNumPerSecond";
+
+    /**
+     * The constant RATE_LIMIT_ENABLE.
+     */
+    String RATE_LIMIT_ENABLE = RATE_LIMIT_PREFIX + ".enable";
+
+    /**
+     * The constant RATE_LIMIT_BUCKET_TOKEN_MAX_NUM.
+     */
+    String RATE_LIMIT_BUCKET_TOKEN_MAX_NUM = RATE_LIMIT_PREFIX + ".bucketTokenMaxNum";
+
+    /**
+     * The constant RATE_LIMIT_BUCKET_TOKEN_INITIAL_NUM.
+     */
+    String RATE_LIMIT_BUCKET_TOKEN_INITIAL_NUM = RATE_LIMIT_PREFIX + ".bucketTokenInitialNum";
 }

common/src/main/java/org/apache/seata/common/DefaultValues.java

@@ -497,6 +497,12 @@ public interface DefaultValues {
      */
     int DEFAULT_ROCKET_MQ_MSG_TIMEOUT = 60 * 1000;
 
+    /**
+     * The constant DEFAULT_RATE_LIMIT_ENABLE.
+     */
+    boolean DEFAULT_RATE_LIMIT_ENABLE = false;
+
+
     /**
      * The constant DEFAULT_RAFT_SSL_ENABLED.
      */

compatible/src/main/java/io/seata/tm/api/DefaultFailureHandlerImpl.java

@@ -28,6 +28,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+
 /**
  * The type Default failure handler.
  */

core/src/main/java/org/apache/seata/core/event/RateLimitEvent.java

@@ -0,0 +1,104 @@
+/*
+ * 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.seata.core.event;
+
+public class RateLimitEvent implements Event {
+
+    /**
+     * The Trace id.
+     */
+    private String traceId;
+
+    /**
+     * The Limit type (like GlobalBeginFailed).
+     */
+    private String limitType;
+
+    /**
+     * The Application id.
+     */
+    private String applicationId;
+
+    /**
+     * The Client id.
+     */
+    private String clientId;
+
+    /**
+     * The Server ip address and port.
+     */
+    private String serverIpAddressAndPort;
+
+    public String getTraceId() {
+        return traceId;
+    }
+
+    public void setTraceId(String traceId) {
+        this.traceId = traceId;
+    }
+
+    public String getLimitType() {
+        return limitType;
+    }
+
+    public void setLimitType(String limitType) {
+        this.limitType = limitType;
+    }
+
+    public String getApplicationId() {
+        return applicationId;
+    }
+
+    public void setApplicationId(String applicationId) {
+        this.applicationId = applicationId;
+    }
+
+    public String getClientId() {
+        return clientId;
+    }
+
+    public void setClientId(String clientId) {
+        this.clientId = clientId;
+    }
+
+    public String getServerIpAddressAndPort() {
+        return serverIpAddressAndPort;
+    }
+
+    public void setServerIpAddressAndPort(String serverIpAddressAndPort) {
+        this.serverIpAddressAndPort = serverIpAddressAndPort;
+    }
+
+    public RateLimitEvent(String traceId, String limitType, String applicationId, String clientId, String serverIpAddressAndPort) {
+        this.traceId = traceId;
+        this.limitType = limitType;
+        this.applicationId = applicationId;
+        this.clientId = clientId;
+        this.serverIpAddressAndPort = serverIpAddressAndPort;
+    }
+
+    @Override
+    public String toString() {
+        return "RateLimitEvent{" +
+                "traceId='" + traceId + '\'' +
+                ", limitType='" + limitType + '\'' +
+                ", applicationId='" + applicationId + '\'' +
+                ", clientId='" + clientId + '\'' +
+                ", serverIpAddressAndPort='" + serverIpAddressAndPort + '\'' +
+                '}';
+    }
+}

core/src/main/java/org/apache/seata/core/exception/TransactionExceptionCode.java

@@ -31,7 +31,6 @@ public enum TransactionExceptionCode {
      * BeginFailed
      */
     BeginFailed,
-
     /**
      * Lock key conflict transaction exception code.
      */

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

@@ -22,6 +22,11 @@
  */
 public interface MessageType {
 
+    /**
+     * The constant TYPE_NOT_EXIST.
+     */
+    short TYPE_NOT_EXIST = 0;
+
     /**
      * The constant TYPE_GLOBAL_BEGIN.
      */

core/src/test/java/org/apache/seata/core/protocol/ResultCodeTest.java

@@ -45,7 +45,8 @@ void getInt() {
 
     @Test
     void values() {
-        Assertions.assertArrayEquals(new ResultCode[]{ResultCode.Failed, ResultCode.Success}, ResultCode.values());
+        Assertions.assertArrayEquals(new ResultCode[]{ResultCode.Failed, ResultCode.Success},
+                ResultCode.values());
     }
 
     @Test

dependencies/pom.xml

@@ -68,6 +68,7 @@
         <sofa.registry.version>6.3.0</sofa.registry.version>
         <motan.version>1.0.0</motan.version>
         <jcommander.version>1.82</jcommander.version>
+        <bucket4j.version>8.1.0</bucket4j.version>
         <commons-compress.version>1.21</commons-compress.version>
         <ant.version>1.10.12</ant.version>
         <lz4.version>1.7.1</lz4.version>
@@ -615,6 +616,11 @@
                 <version>${jcommander.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>com.bucket4j</groupId>
+                <artifactId>bucket4j_jdk8-core</artifactId>
+                <version>${bucket4j.version}</version>
+            </dependency>
             <dependency>
                 <groupId>io.grpc</groupId>
                 <artifactId>grpc-testing</artifactId>

integration-tx-api/src/main/java/org/apache/seata/integration/tx/api/interceptor/handler/GlobalTransactionalInterceptorHandler.java

@@ -56,6 +56,7 @@
 import org.apache.seata.tm.api.transaction.NoRollbackRule;
 import org.apache.seata.tm.api.transaction.RollbackRule;
 import org.apache.seata.tm.api.transaction.TransactionInfo;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

metrics/seata-metrics-api/src/main/java/org/apache/seata/metrics/IdConstants.java

@@ -25,6 +25,8 @@ public interface IdConstants {
 
     String SEATA_EXCEPTION = "seata.exception";
 
+    String SEATA_RATE_LIMIT = "seata.rate.limit";
+
     String APP_ID_KEY = "applicationId";
 
     String GROUP_KEY = "group";
@@ -79,4 +81,9 @@ public interface IdConstants {
 
     String STATUS_VALUE_AFTER_ROLLBACKED_KEY = "AfterRollbacked";
 
+    String LIMIT_TYPE_KEY = "limitType";
+
+    String CLIENT_ID_KEY = "clientId";
+
+    String HOST_AND_PORT = "hostAndPort";
 }

saga/seata-saga-spring/src/main/java/org/apache/seata/saga/engine/tm/DefaultSagaTransactionalTemplate.java

@@ -16,15 +16,14 @@
  */
 package org.apache.seata.saga.engine.tm;
 
-import java.util.List;
-
 import org.apache.seata.common.exception.FrameworkErrorCode;
+import org.apache.seata.common.util.StringUtils;
 import org.apache.seata.core.exception.TransactionException;
 import org.apache.seata.core.model.BranchStatus;
 import org.apache.seata.core.model.BranchType;
 import org.apache.seata.core.model.GlobalStatus;
-import org.apache.seata.core.rpc.netty.RmNettyRemotingClient;
 import org.apache.seata.core.rpc.ShutdownHook;
+import org.apache.seata.core.rpc.netty.RmNettyRemotingClient;
 import org.apache.seata.core.rpc.netty.TmNettyRemotingClient;
 import org.apache.seata.rm.DefaultResourceManager;
 import org.apache.seata.rm.RMClient;
@@ -39,7 +38,6 @@
 import org.apache.seata.tm.api.transaction.TransactionHook;
 import org.apache.seata.tm.api.transaction.TransactionHookManager;
 import org.apache.seata.tm.api.transaction.TransactionInfo;
-import org.apache.seata.common.util.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
@@ -49,6 +47,8 @@
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ConfigurableApplicationContext;
 
+import java.util.List;
+
 /**
  * Template of executing business logic with a global transaction for SAGA mode
  */
@@ -93,7 +93,6 @@ public GlobalTransaction beginTransaction(TransactionInfo txInfo) throws Transac
             triggerAfterBegin(tx);
         } catch (TransactionException txe) {
             throw new TransactionalExecutor.ExecutionException(tx, txe, TransactionalExecutor.Code.BeginFailure);
-
         }
         return tx;
     }

script/config-center/config.txt

@@ -162,7 +162,10 @@ server.raft.serialization=jackson
 server.raft.compressor=none
 server.raft.sync=true
 
-
+server.ratelimit.enable=false
+server.ratelimit.bucketTokenNumPerSecond = 999999
+server.ratelimit.bucketTokenMaxNum = 999999
+server.ratelimit.bucketTokenInitialNum = 999999
 
 #Metrics configuration, only for the server
 metrics.enabled=true

seata-spring-autoconfigure/seata-spring-autoconfigure-core/src/main/java/org/apache/seata/spring/boot/autoconfigure/StarterConstants.java

@@ -69,6 +69,7 @@ public interface StarterConstants {
 
 
     String SERVER_PREFIX = SEATA_PREFIX + ".server";
+    String SERVER_RATELIMIT_PREFIX = SERVER_PREFIX + ".ratelimit";
     String SERVER_UNDO_PREFIX = SERVER_PREFIX + ".undo";
     String SERVER_RAFT_PREFIX = SERVER_PREFIX + ".raft";
     String SERVER_RECOVERY_PREFIX = SERVER_PREFIX + ".recovery";

seata-spring-autoconfigure/seata-spring-autoconfigure-server/src/main/java/org/apache/seata/spring/boot/autoconfigure/SeataServerEnvironmentPostProcessor.java

@@ -17,6 +17,8 @@
 package org.apache.seata.spring.boot.autoconfigure;
 
 import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.seata.spring.boot.autoconfigure.properties.server.ServerRateLimitProperties;
 import org.apache.seata.spring.boot.autoconfigure.properties.server.store.StoreProperties;
 import org.apache.seata.spring.boot.autoconfigure.properties.server.MetricsProperties;
 import org.apache.seata.spring.boot.autoconfigure.properties.server.ServerProperties;
@@ -38,6 +40,7 @@
 import static org.apache.seata.spring.boot.autoconfigure.StarterConstants.PROPERTY_BEAN_MAP;
 import static org.apache.seata.spring.boot.autoconfigure.StarterConstants.SERVER_PREFIX;
 import static org.apache.seata.spring.boot.autoconfigure.StarterConstants.SERVER_RAFT_PREFIX;
+import static org.apache.seata.spring.boot.autoconfigure.StarterConstants.SERVER_RATELIMIT_PREFIX;
 import static org.apache.seata.spring.boot.autoconfigure.StarterConstants.SERVER_RECOVERY_PREFIX;
 import static org.apache.seata.spring.boot.autoconfigure.StarterConstants.SERVER_UNDO_PREFIX;
 import static org.apache.seata.spring.boot.autoconfigure.StarterConstants.SESSION_PREFIX;
@@ -81,6 +84,7 @@ public static void init() {
             PROPERTY_BEAN_MAP.put(SERVER_RAFT_PREFIX, ServerRaftProperties.class);
             PROPERTY_BEAN_MAP.put(SESSION_PREFIX, SessionProperties.class);
             PROPERTY_BEAN_MAP.put(STORE_PREFIX, StoreProperties.class);
+            PROPERTY_BEAN_MAP.put(SERVER_RATELIMIT_PREFIX, ServerRateLimitProperties.class);
         }
     }