feature: Reuse connection to merge branch transactions

changes/en-us/2.x.md

@@ -23,6 +23,7 @@ Add changes here for all PR submitted to the 2.x branch.
 - [[#7261](https://github.com/apache/incubator-seata/pull/7261)] enforce account initialization and disable default credentials
 - [[#7451](https://github.com/apache/incubator-seata/pull/7451)] seata-server supports the HTTP/2 protocol
 - [[#7496](https://github.com/apache/incubator-seata/pull/7496)] add oceanbase oracle support
+- [[#7509](https://github.com/apache/incubator-seata/pull/7509)] Reuse connection to merge branch transactions
 
 
 ### bugfix:

changes/zh-cn/2.x.md

@@ -23,6 +23,7 @@
 - [[#7261](https://github.com/apache/incubator-seata/pull/7261)] 强制进行账户初始化并禁用默认凭据
 - [[#7451](https://github.com/apache/incubator-seata/pull/7451)] seata-server支持HTTP/2协议
 - [[#7496](https://github.com/apache/incubator-seata/pull/7496)] 添加对oceanbase数据库oracle模式的支持
+- [[#7509](https://github.com/apache/incubator-seata/pull/7509)] 复用连接合并分支事务
 
 
 ### bugfix:

core/src/main/java/org/apache/seata/core/context/RootContext.java

@@ -207,6 +207,10 @@ public static boolean inSagaBranch() {
         return BranchType.SAGA == getBranchType();
     }
 
+    public static boolean inXABranch() {
+        return BranchType.XA == getBranchType();
+    }
+
     /**
      * get the branch type
      *

integration-tx-api/pom.xml

@@ -61,7 +61,14 @@
             <groupId>net.bytebuddy</groupId>
             <artifactId>byte-buddy</artifactId>
         </dependency>
-
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-context</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.aspectj</groupId>
+            <artifactId>aspectjweaver</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

integration-tx-api/src/main/java/org/apache/seata/integration/tx/api/combine/CombineAspect.java

@@ -0,0 +1,96 @@
+/*
+ * 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.integration.tx.api.combine;
+
+import org.apache.seata.core.context.RootContext;
+import org.apache.seata.rm.datasource.combine.CombineConnectionHolder;
+import org.apache.seata.rm.datasource.combine.CombineContext;
+import org.apache.seata.rm.datasource.xa.ConnectionProxyXA;
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+@Aspect
+@Component
+public class CombineAspect {
+    private static final Logger LOGGER = LoggerFactory.getLogger(CombineAspect.class);
+
+    @Around("@annotation(org.apache.seata.spring.annotation.CombineTransactional)")
+    public Object handleCombine(ProceedingJoinPoint joinPoint) throws Throwable {
+        if (!RootContext.inGlobalTransaction() || !RootContext.inXABranch()) {
+            // not in transaction, or this interceptor is disabled
+            return joinPoint.proceed();
+        }
+
+        if (!CombineContext.set()) {
+            // The same global transaction, the aspect does not need to enter
+            return joinPoint.proceed();
+        }
+
+        try {
+            // First cut entry
+            Object result = joinPoint.proceed();
+
+            // doCleanupAfterCompletion marks the end of the transaction, resets and closes the connection
+            CombineContext.clear();
+            // doCommit
+            for (ConnectionProxyXA conn : CombineConnectionHolder.getDsConn()) {
+                conn.commit();
+            }
+            return result;
+        } catch (Exception e) {
+            LOGGER.error(
+                    String.format("Failed to handle,xid: %s occur exp msg: %s", RootContext.getXID(), e.getMessage()),
+                    e);
+            CombineContext.clear();
+            // doRollback
+            for (ConnectionProxyXA conn : CombineConnectionHolder.getDsConn()) {
+                conn.rollback();
+            }
+            throw e;
+        } finally {
+            CombineContext.clear();
+            for (ConnectionProxyXA conn : CombineConnectionHolder.getDsConn()) {
+                try {
+                    // Reset autocommit (if not autocommitting)
+                    if (!conn.getAutoCommit()) {
+                        conn.setAutoCommit(true);
+                    }
+                } catch (Throwable t) {
+                    // Record the exception of resetting the auto-commit, but do not interrupt and continue to try to
+                    // close
+                    LOGGER.error("Failed to reset autoCommit to true for connection: {}", conn, t);
+                }
+                try {
+                    if (conn.isClosed()) {
+                        LOGGER.error("Connection is closed: {}", conn);
+                    }
+                    conn.close();
+                } catch (Throwable t) {
+                    // Record the exception of closing the connection, but do not interrupt the loop and continue to
+                    // process the next connection
+                    LOGGER.error("Failed to close connection: {}", conn, t);
+                }
+            }
+            // Clean up local cache connections
+            CombineConnectionHolder.clear();
+        }
+    }
+}

integration-tx-api/src/main/java/org/apache/seata/integration/tx/api/combine/CombineAutoConfiguration.java

@@ -0,0 +1,26 @@
+/*
+ * 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.integration.tx.api.combine;
+
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.EnableAspectJAutoProxy;
+
+@ComponentScan(basePackages = "org.apache.seata.integration.tx.api.combine")
+@Configuration
+@EnableAspectJAutoProxy(proxyTargetClass = true)
+public class CombineAutoConfiguration {}

integration-tx-api/src/main/java/org/apache/seata/spring/annotation/CombineTransactional.java

@@ -0,0 +1,24 @@
+/*
+ * 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.spring.annotation;
+
+import java.lang.annotation.*;
+
+@Target({ElementType.TYPE, ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+public @interface CombineTransactional {}

integration-tx-api/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,19 @@
+#
+# 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.
+#
+# Auto Configure
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+org.apache.seata.integration.tx.api.combine.CombineAutoConfiguration

rm-datasource/src/main/java/org/apache/seata/rm/datasource/combine/CombineConnectionHolder.java

@@ -0,0 +1,60 @@
+/*
+ * 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.rm.datasource.combine;
+
+import org.apache.seata.core.context.RootContext;
+import org.apache.seata.rm.datasource.xa.ConnectionProxyXA;
+
+import javax.sql.DataSource;
+import java.sql.SQLException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class CombineConnectionHolder {
+    private static final ThreadLocal<Map<String, Map<Object, ConnectionProxyXA>>> CONNECTION_HOLDER =
+            ThreadLocal.withInitial(ConcurrentHashMap::new);
+
+    public static ConnectionProxyXA get(DataSource dataSource) {
+        Map<Object, ConnectionProxyXA> connMap = CONNECTION_HOLDER.get().get(RootContext.getXID());
+        if (connMap != null) {
+            return connMap.get(dataSource);
+        }
+        return null;
+    }
+
+    public static Collection<ConnectionProxyXA> getDsConn() {
+        Map<Object, ConnectionProxyXA> connectionMap = CONNECTION_HOLDER.get().get(RootContext.getXID());
+        return connectionMap != null ? connectionMap.values() : Collections.emptyList();
+    }
+
+    public static void putConnection(DataSource dataSource, ConnectionProxyXA connection) throws SQLException {
+        Map<String, Map<Object, ConnectionProxyXA>> concurrentHashMap = CONNECTION_HOLDER.get();
+        String xid = RootContext.getXID();
+        Map<Object, ConnectionProxyXA> connectionProxyMap =
+                concurrentHashMap.computeIfAbsent(xid, k -> new ConcurrentHashMap<>());
+
+        if (connectionProxyMap.putIfAbsent(dataSource, connection) == null) {
+            connection.setAutoCommit(false);
+        }
+    }
+
+    public static void clear() {
+        CONNECTION_HOLDER.get().remove(RootContext.getXID());
+    }
+}

rm-datasource/src/main/java/org/apache/seata/rm/datasource/combine/CombineContext.java

@@ -0,0 +1,56 @@
+/*
+ * 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.rm.datasource.combine;
+
+import org.apache.seata.core.context.RootContext;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class CombineContext {
+    private static final ThreadLocal<Map<String, Boolean>> COMBINE_ASPECT =
+            ThreadLocal.withInitial(ConcurrentHashMap::new);
+    ;
+
+    /**
+     * @return
+     *    false: The specified key already exists (repeatedly enter the aspect)
+     *    true: The specified key does not exist (first time entering the aspect)
+     */
+    public static boolean set() {
+        String xid = RootContext.getXID();
+        if (xid != null) {
+            return !Boolean.TRUE.equals(COMBINE_ASPECT.get().putIfAbsent(xid, Boolean.TRUE));
+        }
+        return false;
+    }
+
+    public static boolean get() {
+        String xid = RootContext.getXID();
+        if (xid == null) {
+            return false;
+        }
+        return Boolean.TRUE.equals(COMBINE_ASPECT.get().get(xid));
+    }
+
+    public static void clear() {
+        String xid = RootContext.getXID();
+        if (xid != null) {
+            COMBINE_ASPECT.get().remove(xid);
+        }
+    }
+}

rm-datasource/src/main/java/org/apache/seata/rm/datasource/xa/ConnectionProxyXA.java

@@ -25,6 +25,7 @@
 import org.apache.seata.core.model.BranchType;
 import org.apache.seata.rm.BaseDataSourceResource;
 import org.apache.seata.rm.DefaultResourceManager;
+import org.apache.seata.rm.datasource.combine.CombineContext;
 import org.apache.seata.rm.datasource.util.SeataXAResource;
 import org.apache.seata.sqlparser.util.JdbcConstants;
 import org.slf4j.Logger;
@@ -227,6 +228,9 @@
     @Override
     public void commit() throws SQLException {
         try (ResourceLock ignored = resourceLock.obtain()) {
+            if (CombineContext.get()) {
+                return;
+            }
             if (currentAutoCommitStatus || isReadOnly()) {
                 // Ignore the committing on an autocommit session and read-only transaction.
                 return;
@@ -239,6 +243,9 @@
 
     @Override
     public void rollback() throws SQLException {
+        if (CombineContext.get()) {
+            return;
+        }
         if (currentAutoCommitStatus || isReadOnly()) {
             // Ignore the committing on an autocommit session and read-only transaction.
             return;
@@ -312,6 +319,9 @@
     @Override
     public void close() throws SQLException {
         try (ResourceLock ignored = resourceLock.obtain()) {
+            if (CombineContext.get()) {
+                return;
+            }
             try {
                 if (xaActive && this.xaBranchXid != null) {
                     // XA End: Success