bugfix: dubbo reference annotation failed to register transaction on tcc mode

core/src/main/java/io/seata/core/context/RootContext.java

@@ -16,7 +16,8 @@
 package io.seata.core.context;
 
 import io.seata.common.exception.ShouldNeverHappenException;
-
+import io.seata.core.model.BranchType;
+import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -36,6 +37,8 @@ public class RootContext {
      */
     public static final String KEY_XID = "TX_XID";
 
+    public static final String KEY_XID_FILTER_TYPE = "TX_XID_FILTER_TYPE";
+
     public static final String KEY_GLOBAL_LOCK_FLAG = "TX_LOCK";
 
     private static ContextCore CONTEXT_HOLDER = ContextCoreLoader.load();
@@ -46,7 +49,25 @@ public class RootContext {
      * @return the xid
      */
     public static String getXID() {
-        return CONTEXT_HOLDER.get(KEY_XID);
+        String xid = CONTEXT_HOLDER.get(KEY_XID);
+        if (StringUtils.isNotBlank(xid)) {
+            return xid;
+        }
+
+        String xidType = CONTEXT_HOLDER.get(KEY_XID_FILTER_TYPE);
+        if (StringUtils.isNotBlank(xidType)) {
+            return xidType.split("_")[0];
+        }
+        return null;
+    }
+
+    /**
+     * Gets xid.
+     *
+     * @return the xid
+     */
+    public static String getXIDFilterType() {
+        return CONTEXT_HOLDER.get(KEY_XID_FILTER_TYPE);
     }
 
     /**
@@ -61,6 +82,30 @@ public static void bind(String xid) {
         CONTEXT_HOLDER.put(KEY_XID, xid);
     }
 
+    /**
+     * Bind filter type
+     *
+     * @param xidType
+     */
+    public static void bindFilterType(String xidType) {
+        String[] xidTypes = xidType.split("_");
+        bindFilterType(xidTypes[0], BranchType.valueOf(xidTypes[1]));
+    }
+
+    /**
+     * Bind filter type
+     *
+     * @param xid
+     * @param branchType
+     */
+    public static void bindFilterType(String xid, BranchType branchType) {
+        String xidType = String.format("%s_%s", xid, branchType.name());
+        if (LOGGER.isDebugEnabled()) {
+            LOGGER.debug("bind filter type xid={} branchType={}", xid, branchType);
+        }
+        CONTEXT_HOLDER.put(KEY_XID_FILTER_TYPE, xidType);
+    }
+
     /**
      * declare local transactions will use global lock check for update/delete/insert/selectForUpdate SQL
      */
@@ -82,11 +127,24 @@ public static void bindGlobalLockFlag() {
     public static String unbind() {
         String xid = CONTEXT_HOLDER.remove(KEY_XID);
         if (LOGGER.isDebugEnabled()) {
-            LOGGER.debug("unbind " + xid);
+            LOGGER.debug("unbind {} ", xid);
         }
         return xid;
     }
 
+    /**
+     * Unbind temporary string
+     *
+     * @return the string
+     */
+    public static String unbindFilterType() {
+        String xidType = CONTEXT_HOLDER.remove(KEY_XID_FILTER_TYPE);
+        if (LOGGER.isDebugEnabled()) {
+            LOGGER.debug("unbind filter type {}", xidType);
+        }
+        return xidType;
+    }
+
     public static void unbindGlobalLockFlag() {
         String lockFlag = CONTEXT_HOLDER.remove(KEY_GLOBAL_LOCK_FLAG);
         if (LOGGER.isDebugEnabled() && lockFlag != null) {

integration/dubbo-alibaba/pom.xml

@@ -35,6 +35,11 @@
             <artifactId>seata-tm</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>seata-tcc</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>com.alibaba</groupId>
             <artifactId>dubbo</artifactId>

integration/dubbo-alibaba/src/main/java/io/seata/integration/dubbo/alibaba/TccReferenceAnnotationFilter.java

@@ -0,0 +1,88 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  Licensed 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 io.seata.integration.dubbo.alibaba;
+
+import com.alibaba.dubbo.common.Constants;
+import com.alibaba.dubbo.common.extension.Activate;
+import com.alibaba.dubbo.rpc.Filter;
+import com.alibaba.dubbo.rpc.Invocation;
+import com.alibaba.dubbo.rpc.Invoker;
+import com.alibaba.dubbo.rpc.Result;
+import com.alibaba.dubbo.rpc.RpcException;
+import com.alibaba.dubbo.rpc.RpcInvocation;
+import io.seata.core.context.RootContext;
+import io.seata.core.model.BranchType;
+import io.seata.rm.tcc.api.TwoPhaseBusinessAction;
+import io.seata.rm.tcc.interceptor.ActionInterceptorHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Method;
+import java.util.Map;
+
+@Activate(group = {Constants.CONSUMER}, order = 95)
+public class TccReferenceAnnotationFilter implements Filter {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(TccReferenceAnnotationFilter.class);
+
+    private ActionInterceptorHandler actionInterceptorHandler = new ActionInterceptorHandler();
+
+    private final static String DUBBO_GENERIC_SERVICE_INVOKE = "$invoke";
+
+    @Override
+    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
+        try {
+            RpcInvocation rpcInvocation = (RpcInvocation) invocation;
+            String methodName = rpcInvocation.getMethodName();
+            Class<?>[] parameterTypes = rpcInvocation.getParameterTypes();
+            Object[] arguments = rpcInvocation.getArguments();
+            Class interfaceClass = invoker.getInterface();
+            if (DUBBO_GENERIC_SERVICE_INVOKE.equals(methodName)) {
+                return invoker.invoke(invocation);
+            }
+
+            Method method = interfaceClass.getMethod(methodName, parameterTypes);
+            //not in transaction
+            if (!RootContext.inGlobalTransaction()) {
+                return invoker.invoke(invocation);
+            }
+
+            TwoPhaseBusinessAction businessAction = method.getAnnotation(TwoPhaseBusinessAction.class);
+            //try method
+            if (businessAction != null) {
+                //save the xid
+                String xid = RootContext.getXID();
+                //clear the context
+                RootContext.unbind();
+                RootContext.bindFilterType(xid, BranchType.TCC);
+                try {
+                    Map<String, Object> ret = actionInterceptorHandler.proceed(method, arguments, xid, businessAction,
+                            () -> invoker.invoke(invocation));
+                    return (Result) ret.get(io.seata.common.Constants.TCC_METHOD_RESULT);
+                } catch (Throwable throwable) {
+                    throw throwable;
+                } finally {
+                    //recovery the context
+                    RootContext.unbindFilterType();
+                    RootContext.bind(xid);
+                }
+            }
+        } catch (Throwable e) {
+            LOGGER.error("Tcc dubbo invokes service to register branch transaction exception:{}", e.getMessage(), e);
+        }
+        return invoker.invoke(invocation);
+    }
+}

integration/dubbo-alibaba/src/main/java/io/seata/integration/dubbo/alibaba/TransactionPropagationFilter.java

@@ -24,7 +24,6 @@
 import com.alibaba.dubbo.rpc.RpcContext;
 import com.alibaba.dubbo.rpc.RpcException;
 import io.seata.core.context.RootContext;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,19 +40,25 @@ public class TransactionPropagationFilter implements Filter {
     @Override
     public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
         String xid = RootContext.getXID();
+        String xidFilterType = RootContext.getXIDFilterType();
+
         String rpcXid = RpcContext.getContext().getAttachment(RootContext.KEY_XID);
+        String rpcXidFilterType = RpcContext.getContext().getAttachment(RootContext.KEY_XID_FILTER_TYPE);
         if (LOGGER.isDebugEnabled()) {
             LOGGER.debug("xid in RootContext[" + xid + "] xid in RpcContext[" + rpcXid + "]");
         }
         boolean bind = false;
         if (xid != null) {
             RpcContext.getContext().setAttachment(RootContext.KEY_XID, xid);
+            RpcContext.getContext().setAttachment(RootContext.KEY_XID_FILTER_TYPE, xidFilterType);
         } else {
             if (rpcXid != null) {
                 RootContext.bind(rpcXid);
+                RootContext.bindFilterType(rpcXidFilterType);
                 bind = true;
                 if (LOGGER.isDebugEnabled()) {
-                    LOGGER.debug("bind[" + rpcXid + "] to RootContext");
+                    LOGGER.debug("bind[{}] to RootContext", rpcXid);
+                    LOGGER.debug("bind filterType[{}] to RootContext", rpcXidFilterType);
                 }
             }
         }
@@ -63,14 +68,19 @@ public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcExcept
         } finally {
             if (bind) {
                 String unbindXid = RootContext.unbind();
+                String unbindFilterType = RootContext.unbindFilterType();
                 if (LOGGER.isDebugEnabled()) {
-                    LOGGER.debug("unbind[" + unbindXid + "] from RootContext");
+                    LOGGER.debug("unbind[{}] from RootContext", unbindXid);
+                    LOGGER.debug("unbind filterType[{}] from RootContext", unbindFilterType);
                 }
                 if (!rpcXid.equalsIgnoreCase(unbindXid)) {
-                    LOGGER.warn("xid in change during RPC from " + rpcXid + " to " + unbindXid);
+                    LOGGER.warn("xid in change during RPC from {} to {}", rpcXid, unbindXid);
+                    LOGGER.warn("xidFilterType in change during RPC from {} to {}", rpcXidFilterType, unbindFilterType);
                     if (unbindXid != null) {
                         RootContext.bind(unbindXid);
-                        LOGGER.warn("bind [" + unbindXid + "] back to RootContext");
+                        RootContext.bindFilterType(unbindFilterType);
+                        LOGGER.warn("bind [{}] back to RootContext", unbindXid);
+                        LOGGER.warn("bind filterType [{}] back to RootContext", unbindFilterType);
                     }
                 }
             }

integration/dubbo-alibaba/src/main/resources/META-INF/services/com.alibaba.dubbo.rpc.Filter

@@ -1 +1,2 @@
-io.seata.integration.dubbo.alibaba.TransactionPropagationFilter
+io.seata.integration.dubbo.alibaba.TransactionPropagationFilter
+io.seata.integration.dubbo.alibaba.TccReferenceAnnotationFilter

integration/dubbo/pom.xml

@@ -34,6 +34,11 @@
             <artifactId>seata-tm</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>seata-tcc</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.dubbo</groupId>
             <artifactId>dubbo</artifactId>

integration/dubbo/src/main/java/io/seata/integration/dubbo/TccReferenceAnnotationFilter.java

@@ -0,0 +1,88 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  Licensed 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 io.seata.integration.dubbo;
+
+import io.seata.core.context.RootContext;
+import io.seata.core.model.BranchType;
+import io.seata.rm.tcc.api.TwoPhaseBusinessAction;
+import io.seata.rm.tcc.interceptor.ActionInterceptorHandler;
+import org.apache.dubbo.common.Constants;
+import org.apache.dubbo.common.extension.Activate;
+import org.apache.dubbo.rpc.Filter;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.RpcInvocation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Method;
+import java.util.Map;
+
+@Activate(group = {Constants.CONSUMER}, order = 95)
+public class TccReferenceAnnotationFilter implements Filter {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(TccReferenceAnnotationFilter.class);
+
+    private ActionInterceptorHandler actionInterceptorHandler = new ActionInterceptorHandler();
+
+    private final static String DUBBO_GENERIC_SERVICE_INVOKE = "$invoke";
+
+    @Override
+    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
+        try {
+            RpcInvocation rpcInvocation = (RpcInvocation) invocation;
+            String methodName = rpcInvocation.getMethodName();
+            Class<?>[] parameterTypes = rpcInvocation.getParameterTypes();
+            Object[] arguments = rpcInvocation.getArguments();
+            Class interfaceClass = invoker.getInterface();
+            if (DUBBO_GENERIC_SERVICE_INVOKE.equals(methodName)) {
+                return invoker.invoke(invocation);
+            }
+
+            Method method = interfaceClass.getMethod(methodName, parameterTypes);
+            //not in transaction
+            if (!RootContext.inGlobalTransaction()) {
+                return invoker.invoke(invocation);
+            }
+
+            TwoPhaseBusinessAction businessAction = method.getAnnotation(TwoPhaseBusinessAction.class);
+            //try method
+            if (businessAction != null) {
+                //save the xid
+                String xid = RootContext.getXID();
+                //clear the context
+                RootContext.unbind();
+                RootContext.bindFilterType(xid, BranchType.TCC);
+                try {
+                    Map<String, Object> ret = actionInterceptorHandler.proceed(method, arguments, xid, businessAction,
+                            () -> invoker.invoke(invocation));
+                    return (Result) ret.get(io.seata.common.Constants.TCC_METHOD_RESULT);
+                } catch (Throwable throwable) {
+                    throw throwable;
+                } finally {
+                    //recovery the context
+                    RootContext.unbindFilterType();
+                    RootContext.bind(xid);
+                }
+            }
+        } catch (Throwable e) {
+            LOGGER.error("Tcc dubbo invokes service to register branch transaction exception:{}", e.getMessage(), e);
+        }
+        return invoker.invoke(invocation);
+    }
+}