feature: Integrate RocketMQ into Seata

dependencies/pom.xml

@@ -113,6 +113,7 @@
         <mockito.version>2.23.4</mockito.version>
         <assertj-core.version>3.12.2</assertj-core.version>
         <jetty-version>9.4.38.v20210224</jetty-version>
+        <rocketmq-version>4.9.0</rocketmq-version>
         <bytebuddy.version>1.12.17</bytebuddy.version>
     </properties>
 
@@ -687,6 +688,12 @@
                 <artifactId>jetty-servlet</artifactId>
                 <version>${jetty-version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.apache.rocketmq</groupId>
+                <artifactId>rocketmq-client</artifactId>
+                <version>${rocketmq-version}</version>
+                <optional>true</optional>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 </project>

integration-tx-api/pom.xml

@@ -14,8 +14,8 @@
   ~  See the License for the specific language governing permissions and
   ~  limitations under the License.
   -->
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xmlns="http://maven.apache.org/POM/4.0.0"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <parent>
         <groupId>io.seata</groupId>

integration-tx-api/src/main/java/io/seata/integration/tx/api/fence/DefaultCommonFenceHandler.java

@@ -28,10 +28,6 @@ public class DefaultCommonFenceHandler implements FenceHandler {
 
     private FenceHandler fenceHandler;
 
-    private static class SingletonHolder {
-        private static final DefaultCommonFenceHandler INSTANCE = new DefaultCommonFenceHandler();
-    }
-
     public static DefaultCommonFenceHandler get() {
         return DefaultCommonFenceHandler.SingletonHolder.INSTANCE;
     }
@@ -69,4 +65,8 @@ public int deleteFenceByDate(Date datetime) {
         check();
         return fenceHandler.deleteFenceByDate(datetime);
     }
+
+    private static class SingletonHolder {
+        private static final DefaultCommonFenceHandler INSTANCE = new DefaultCommonFenceHandler();
+    }
 }

integration-tx-api/src/main/java/io/seata/integration/tx/api/fence/config/CommonFenceConfig.java

@@ -46,24 +46,20 @@ public class CommonFenceConfig implements Disposable {
      * Common fence clean period max value. maximum interval is 68 years
      */
     private static final Duration MAX_PERIOD = Duration.ofSeconds(Integer.MAX_VALUE);
-
+    /**
+     * Common fence clean scheduled thread pool
+     */
+    private final ScheduledThreadPoolExecutor tccFenceClean = new ScheduledThreadPoolExecutor(1,
+            new NamedThreadFactory("tccFenceClean", 1));
     /**
      * Common fence clean period. only duration type format are supported
      */
     private Duration cleanPeriod = Duration.ofDays(DefaultValues.DEFAULT_COMMON_FENCE_CLEAN_PERIOD);
-
     /**
      * Common fence log table name
      */
     private String logTableName = DefaultValues.DEFAULT_COMMON_FENCE_LOG_TABLE_NAME;
 
-
-    /**
-     * Common fence clean scheduled thread pool
-     */
-    private final ScheduledThreadPoolExecutor tccFenceClean = new ScheduledThreadPoolExecutor(1,
-            new NamedThreadFactory("tccFenceClean", 1));
-
     public AtomicBoolean getInitialized() {
         return initialized;
     }

integration-tx-api/src/main/java/io/seata/integration/tx/api/fence/store/db/CommonFenceStoreDataBaseDAO.java

@@ -43,13 +43,12 @@
  */
 public class CommonFenceStoreDataBaseDAO implements CommonFenceStore {
 
+    private static volatile CommonFenceStoreDataBaseDAO instance = null;
     /**
      * Common fence log table name
      */
     private String logTableName = DefaultValues.DEFAULT_COMMON_FENCE_LOG_TABLE_NAME;
 
-    private static volatile CommonFenceStoreDataBaseDAO instance = null;
-
     private CommonFenceStoreDataBaseDAO() {}
 
     public static CommonFenceStore getInstance() {

integration-tx-api/src/main/java/io/seata/integration/tx/api/fence/store/db/sql/CommonFenceStoreSqls.java

@@ -24,10 +24,6 @@
  */
 public class CommonFenceStoreSqls {
 
-    private CommonFenceStoreSqls() {
-        throw new IllegalStateException("Utility class");
-    }
-
     /**
      * The constant LOCAL_TCC_LOG_PLACEHOLD.
      */
@@ -46,21 +42,27 @@ private CommonFenceStoreSqls() {
             + " (xid, branch_id, action_name, status, gmt_create, gmt_modified) "
             + " values (?, ?, ?, ?, ?, ?) ";
 
+    /**
+     * The constant QUERY_END_STATUS_BY_DATE.
+     */
+    protected static final String QUERY_END_STATUS_BY_DATE = "select xid, branch_id, status, gmt_create, gmt_modified "
+        + "from " + LOCAL_TCC_LOG_PLACEHOLD
+        + " where  gmt_modified < ? "
+        + " and status in (" + CommonFenceConstant.STATUS_COMMITTED + " , " + CommonFenceConstant.STATUS_ROLLBACKED + " , " + CommonFenceConstant.STATUS_SUSPENDED + ")"
+        + " limit ?";
+
     /**
      * The constant QUERY_BY_BRANCH_ID_AND_XID.
      */
     protected static final String QUERY_BY_BRANCH_ID_AND_XID = "select xid, branch_id, status, gmt_create, gmt_modified "
             + "from " + LOCAL_TCC_LOG_PLACEHOLD
             + " where xid = ? and branch_id = ? for update";
-
     /**
-     * The constant QUERY_END_STATUS_BY_DATE.
+     * The constant DELETE_BY_DATE_AND_STATUS.
      */
-    protected static final String QUERY_END_STATUS_BY_DATE = "select xid, branch_id, status, gmt_create, gmt_modified "
-            + "from " + LOCAL_TCC_LOG_PLACEHOLD
-            + " where  gmt_modified < ? "
-            + " and status in (" + CommonFenceConstant.STATUS_COMMITTED + " , " + CommonFenceConstant.STATUS_ROLLBACKED + " , " + CommonFenceConstant.STATUS_SUSPENDED + ")"
-            + " limit ?";
+    protected static final String DELETE_BY_DATE_AND_STATUS = "delete from " + LOCAL_TCC_LOG_PLACEHOLD
+            + " where gmt_modified < ? "
+            + " and status in (" + CommonFenceConstant.STATUS_COMMITTED + " , " + CommonFenceConstant.STATUS_ROLLBACKED + " , " + CommonFenceConstant.STATUS_SUSPENDED + ")";
 
     /**
      * The constant UPDATE_STATUS_BY_BRANCH_ID_AND_XID.
@@ -79,12 +81,9 @@ private CommonFenceStoreSqls() {
     protected static final String DELETE_BY_BRANCH_XIDS = "delete from " + LOCAL_TCC_LOG_PLACEHOLD + " where xid in (" + PRAMETER_PLACEHOLD + ")";
 
 
-    /**
-     * The constant DELETE_BY_DATE_AND_STATUS.
-     */
-    protected static final String DELETE_BY_DATE_AND_STATUS = "delete from " + LOCAL_TCC_LOG_PLACEHOLD
-            + " where gmt_modified < ? "
-            + " and status in (" + CommonFenceConstant.STATUS_COMMITTED + " , " + CommonFenceConstant.STATUS_ROLLBACKED + " , " + CommonFenceConstant.STATUS_SUSPENDED + ")";
+    private CommonFenceStoreSqls() {
+        throw new IllegalStateException("Utility class");
+    }
 
     public static String getInsertLocalTCCLogSQL(String localTccTable) {
         return INSERT_LOCAL_TCC_LOG.replace(LOCAL_TCC_LOG_PLACEHOLD, localTccTable);

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

@@ -72,43 +72,21 @@ public class GlobalTransactionalInterceptorHandler extends AbstractProxyInvocati
 
     private final TransactionalTemplate transactionalTemplate = new TransactionalTemplate();
     private final GlobalLockTemplate globalLockTemplate = new GlobalLockTemplate();
-
-    private Set<String> methodsToProxy;
+    private static int degradeCheckAllowTimes;
 
     private volatile boolean disable;
     private static final AtomicBoolean ATOMIC_DEGRADE_CHECK = new AtomicBoolean(false);
     private static volatile Integer degradeNum = 0;
     private static volatile Integer reachNum = 0;
-    private static int degradeCheckAllowTimes;
-    protected AspectTransactional aspectTransactional;
     private static int degradeCheckPeriod;
-
     private static int defaultGlobalTransactionTimeout = 0;
-
     private final FailureHandler failureHandler;
+    protected AspectTransactional aspectTransactional;
+    private Set<String> methodsToProxy;
 
     private static final EventBus EVENT_BUS = new GuavaEventBus("degradeCheckEventBus", true);
     private static volatile ScheduledThreadPoolExecutor executor;
 
-    private void initDefaultGlobalTransactionTimeout() {
-        if (GlobalTransactionalInterceptorHandler.defaultGlobalTransactionTimeout <= 0) {
-            int defaultGlobalTransactionTimeout;
-            try {
-                defaultGlobalTransactionTimeout = ConfigurationFactory.getInstance().getInt(
-                        ConfigurationKeys.DEFAULT_GLOBAL_TRANSACTION_TIMEOUT, DEFAULT_GLOBAL_TRANSACTION_TIMEOUT);
-            } catch (Exception e) {
-                LOGGER.error("Illegal global transaction timeout value: " + e.getMessage());
-                defaultGlobalTransactionTimeout = DEFAULT_GLOBAL_TRANSACTION_TIMEOUT;
-            }
-            if (defaultGlobalTransactionTimeout <= 0) {
-                LOGGER.warn("Global transaction timeout value '{}' is illegal, and has been reset to the default value '{}'",
-                        defaultGlobalTransactionTimeout, DEFAULT_GLOBAL_TRANSACTION_TIMEOUT);
-                defaultGlobalTransactionTimeout = DEFAULT_GLOBAL_TRANSACTION_TIMEOUT;
-            }
-            GlobalTransactionalInterceptorHandler.defaultGlobalTransactionTimeout = defaultGlobalTransactionTimeout;
-        }
-    }
-
     public GlobalTransactionalInterceptorHandler(FailureHandler failureHandler, Set<String> methodsToProxy) {
         this.failureHandler = failureHandler == null ? FailureHandlerHolder.getFailureHandler() : failureHandler;
         this.methodsToProxy = methodsToProxy;
@@ -134,6 +112,25 @@ public GlobalTransactionalInterceptorHandler(FailureHandler failureHandler, Set<
         this.aspectTransactional = aspectTransactional;
     }
 
+    private void initDefaultGlobalTransactionTimeout() {
+        if (GlobalTransactionalInterceptorHandler.defaultGlobalTransactionTimeout <= 0) {
+            int defaultGlobalTransactionTimeout;
+            try {
+                defaultGlobalTransactionTimeout = ConfigurationFactory.getInstance().getInt(
+                        ConfigurationKeys.DEFAULT_GLOBAL_TRANSACTION_TIMEOUT, DEFAULT_GLOBAL_TRANSACTION_TIMEOUT);
+            } catch (Exception e) {
+                LOGGER.error("Illegal global transaction timeout value: " + e.getMessage());
+                defaultGlobalTransactionTimeout = DEFAULT_GLOBAL_TRANSACTION_TIMEOUT;
+            }
+            if (defaultGlobalTransactionTimeout <= 0) {
+                LOGGER.warn("Global transaction timeout value '{}' is illegal, and has been reset to the default value '{}'",
+                        defaultGlobalTransactionTimeout, DEFAULT_GLOBAL_TRANSACTION_TIMEOUT);
+                defaultGlobalTransactionTimeout = DEFAULT_GLOBAL_TRANSACTION_TIMEOUT;
+            }
+            GlobalTransactionalInterceptorHandler.defaultGlobalTransactionTimeout = defaultGlobalTransactionTimeout;
+        }
+    }
+
     @Override
     public Set<String> getMethodsToProxy() {
         return methodsToProxy;

integration-tx-api/src/main/java/io/seata/integration/tx/api/interceptor/parser/DefaultInterfaceParser.java

@@ -31,18 +31,14 @@ public class DefaultInterfaceParser implements InterfaceParser {
     protected static final List<InterfaceParser> ALL_INTERFACE_PARSERS = new ArrayList<>();
 
 
-    private static class SingletonHolder {
-        private static final DefaultInterfaceParser INSTANCE = new DefaultInterfaceParser();
+    protected DefaultInterfaceParser() {
+        initInterfaceParser();
     }
 
     public static DefaultInterfaceParser get() {
         return DefaultInterfaceParser.SingletonHolder.INSTANCE;
     }
 
-    protected DefaultInterfaceParser() {
-        initInterfaceParser();
-    }
-
     /**
      * init parsers
      */
@@ -64,4 +60,8 @@ public ProxyInvocationHandler parserInterfaceToProxy(Object target) throws Excep
         return null;
     }
 
+    private static class SingletonHolder {
+        private static final DefaultInterfaceParser INSTANCE = new DefaultInterfaceParser();
+    }
+
 }

integration-tx-api/src/main/java/io/seata/integration/tx/api/interceptor/parser/DefaultResourceRegisterParser.java

@@ -29,22 +29,18 @@ public class DefaultResourceRegisterParser {
 
     protected static List<RegisterResourceParser> allRegisterResourceParsers = new ArrayList<>();
 
-    public void registerResource(Object target) {
-        for (RegisterResourceParser registerResourceParser : allRegisterResourceParsers) {
-            registerResourceParser.registerResource(target);
-        }
-    }
-
-    private static class SingletonHolder {
-        private static final DefaultResourceRegisterParser INSTANCE = new DefaultResourceRegisterParser();
+    protected DefaultResourceRegisterParser() {
+        initResourceRegisterParser();
     }
 
     public static DefaultResourceRegisterParser get() {
         return DefaultResourceRegisterParser.SingletonHolder.INSTANCE;
     }
 
-    protected DefaultResourceRegisterParser() {
-        initResourceRegisterParser();
+    public void registerResource(Object target) {
+        for (RegisterResourceParser registerResourceParser : allRegisterResourceParsers) {
+            registerResourceParser.registerResource(target);
+        }
     }
 
     /**
@@ -57,5 +53,9 @@ protected void initResourceRegisterParser() {
         }
     }
 
+    private static class SingletonHolder {
+        private static final DefaultResourceRegisterParser INSTANCE = new DefaultResourceRegisterParser();
+    }
+
 
 }

integration-tx-api/src/main/java/io/seata/integration/tx/api/interceptor/parser/DefaultTargetClassParser.java

@@ -30,18 +30,14 @@ public class DefaultTargetClassParser implements TargetClassParser {
     protected static List<TargetClassParser> allTargetClassParsers = new ArrayList<>();
 
 
-    private static class SingletonHolder {
-        private static final DefaultTargetClassParser INSTANCE = new DefaultTargetClassParser();
+    protected DefaultTargetClassParser() {
+        initTargetClassParser();
     }
 
     public static DefaultTargetClassParser get() {
         return DefaultTargetClassParser.SingletonHolder.INSTANCE;
     }
 
-    protected DefaultTargetClassParser() {
-        initTargetClassParser();
-    }
-
     /**
      * init parsers
      */
@@ -73,4 +69,8 @@ public Class<?>[] findInterfaces(Object target) throws Exception {
         }
         return target.getClass().getInterfaces();
     }
+
+    private static class SingletonHolder {
+        private static final DefaultTargetClassParser INSTANCE = new DefaultTargetClassParser();
+    }
 }

integration-tx-api/src/main/java/io/seata/integration/tx/api/json/DefaultJsonParser.java

@@ -31,10 +31,6 @@ public class DefaultJsonParser implements JsonParser {
 
     protected static List<JsonParser> allJsonParsers = new ArrayList<>();
 
-    private static class SingletonHolder {
-        private static final DefaultJsonParser INSTANCE = new DefaultJsonParser();
-    }
-
     private DefaultJsonParser() {
         initJsonParser();
     }
@@ -73,4 +69,8 @@ public <T> T parseObject(String text, Class<T> clazz) {
         return null;
     }
 
+    private static class SingletonHolder {
+        private static final DefaultJsonParser INSTANCE = new DefaultJsonParser();
+    }
+
 }

script/client/spring/application.properties

@@ -15,6 +15,7 @@
 #
 
 seata.enabled=true
+seata.rocketmq-enabled=false
 seata.scan-packages=firstPackage,secondPackage
 seata.excludes-for-scanning=firstBeanNameForExclude,secondBeanNameForExclude
 seata.excludes-for-auto-proxying=firstClassNameForExclude,secondClassNameForExclude

script/client/spring/application.yml

@@ -1,5 +1,6 @@
 seata:
   enabled: true
+  rocketmq-enabled: false
   application-id: applicationName
   tx-service-group: default_tx_group
   access-key: aliyunAccessKey

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

@@ -32,6 +32,10 @@ public class SeataProperties {
      * whether enable auto configuration
      */
     private boolean enabled = true;
+    /**
+     * whether enable rocketmq integrate
+     */
+    private boolean rocketmqEnabled = false;
     /**
      * application id
      */
@@ -87,6 +91,14 @@ public SeataProperties setEnabled(boolean enabled) {
         return this;
     }
 
+    public boolean isRocketmqEnabled() {
+        return rocketmqEnabled;
+    }
+
+    public void setRocketmqEnabled(boolean rocketmqEnabled) {
+        this.rocketmqEnabled = rocketmqEnabled;
+    }
+
     public String getApplicationId() {
         if (applicationId == null) {
             applicationId = springCloudAlibabaConfiguration.getApplicationId();