This is an automated email from the ASF dual-hosted git repository.
funky-eyes pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/incubator-seata.git
The following commit(s) were added to refs/heads/2.x by this push:
new 1edc3e7ef4 reactor: refactor the integration-RPC module (#8103)
1edc3e7ef4 is described below
commit 1edc3e7ef44aa90599828b84b62b389ec914953a
Author: jimin <[email protected]>
AuthorDate: Mon May 25 10:13:59 2026 +0800
reactor: refactor the integration-RPC module (#8103)
---
.../etcd3/EtcdRegistryServiceImplMockTest.java | 55 +++++-
.../TransactionPropagationClientInterceptor.java | 54 ++----
.../TransactionPropagationServerInterceptor.java | 68 +++-----
extensions/rpc/seata-dubbo-alibaba/pom.xml | 2 +-
.../AlibabaDubboTransactionConsumerFilter.java | 46 ++---
.../AlibabaDubboTransactionProviderFilter.java | 132 +-------------
extensions/rpc/seata-grpc/pom.xml | 2 +-
.../client/ClientTransactionInterceptor.java | 19 ++-
.../interceptor/server/ServerListenerProxy.java | 54 +-----
.../server/ServerTransactionInterceptor.java | 48 +-----
.../server/ServerListenerProxyTest.java | 14 +-
.../server/ServerTransactionInterceptorTest.java | 16 --
extensions/rpc/seata-hsf/pom.xml | 2 +-
.../hsf/HsfTransactionConsumerFilter.java | 72 ++------
.../hsf/HsfTransactionProviderFilter.java | 159 ++---------------
.../integration/hsf/HsfTransactionFilterTest.java | 190 +++++++++++++++++++++
extensions/rpc/seata-http-jakarta/pom.xml | 6 +
.../http/JakartaSeataWebMvcConfigurerTest.java | 42 +++++
...kartaTransactionPropagationInterceptorTest.java | 116 +++++++++++++
extensions/rpc/seata-motan/pom.xml | 2 +-
.../motan/MotanTransactionConsumerFilter.java | 51 ++++++
.../integration/motan/MotanTransactionFilter.java | 114 -------------
.../motan/MotanTransactionProviderFilter.java | 57 +++++++
.../services/com.weibo.api.motan.filter.Filter | 3 +-
.../seata/integration/rpc/core/BaseRpcFilter.java | 59 -------
.../integration/rpc/core/ConsumerRpcFilter.java | 91 ----------
.../integration/rpc/core/ProviderRpcFilter.java | 110 ------------
.../rpc/core/TransactionPropagationHandler.java | 77 +++++++++
.../core/TransactionPropagationHandlerTest.java | 140 +++++++++++++++
extensions/rpc/seata-sofa-rpc/pom.xml | 2 +-
.../sofa/rpc/TransactionContextConsumerFilter.java | 91 +---------
.../sofa/rpc/TransactionContextProviderFilter.java | 83 +--------
.../seata/rm/datasource/ConnectionProxyTest.java | 77 +++++----
.../seata/rm/datasource/DataSourceProxyTest.java | 9 +-
.../seata/core/rpc/netty/TmNettyClientTest.java | 4 +-
35 files changed, 912 insertions(+), 1155 deletions(-)
diff --git
a/discovery/seata-discovery-etcd3/src/test/java/org/apache/seata/discovery/registry/etcd3/EtcdRegistryServiceImplMockTest.java
b/discovery/seata-discovery-etcd3/src/test/java/org/apache/seata/discovery/registry/etcd3/EtcdRegistryServiceImplMockTest.java
index 93b3cf20c1..e3c52f04b8 100644
---
a/discovery/seata-discovery-etcd3/src/test/java/org/apache/seata/discovery/registry/etcd3/EtcdRegistryServiceImplMockTest.java
+++
b/discovery/seata-discovery-etcd3/src/test/java/org/apache/seata/discovery/registry/etcd3/EtcdRegistryServiceImplMockTest.java
@@ -53,8 +53,11 @@ import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -100,7 +103,7 @@ public class EtcdRegistryServiceImplMockTest {
private static final String CLUSTER_NAME = "default";
@BeforeEach
- public void setUp() throws NoSuchFieldException, IllegalAccessException {
+ public void setUp() throws Exception {
MockitoAnnotations.openMocks(this);
registryService = (EtcdRegistryServiceImpl) spy(new
EtcdRegistryProvider().provide());
@@ -109,18 +112,64 @@ public class EtcdRegistryServiceImplMockTest {
when(mockClient.getWatchClient()).thenReturn(mockWatchClient);
when(mockClient.getKVClient()).thenReturn(mockKVClient);
- // inject spy executorService
+ // cancel any running lifeKeeper from previous tests
+ Field lifeKeeperFutureField =
EtcdRegistryServiceImpl.class.getDeclaredField("lifeKeeperFuture");
+ lifeKeeperFutureField.setAccessible(true);
+ Future<?> oldFuture = (Future<?>)
lifeKeeperFutureField.get(registryService);
+ if (oldFuture != null) {
+ oldFuture.cancel(true);
+ }
+ lifeKeeperFutureField.set(registryService, null);
+
+ // reset lifeKeeper
+ Field lifeKeeperField =
EtcdRegistryServiceImpl.class.getDeclaredField("lifeKeeper");
+ lifeKeeperField.setAccessible(true);
+ lifeKeeperField.set(registryService, null);
+
+ // reset static leaseId
+ Field leaseIdField =
EtcdRegistryServiceImpl.class.getDeclaredField("leaseId");
+ leaseIdField.setAccessible(true);
+ leaseIdField.set(null, 0L);
+
+ // shutdown old executor and create a fresh one
Field executorServiceField =
EtcdRegistryServiceImpl.class.getDeclaredField("executorService");
executorServiceField.setAccessible(true);
- executorService = spy((ExecutorService)
executorServiceField.get(registryService));
+ ExecutorService oldExecutor = (ExecutorService)
executorServiceField.get(registryService);
+ if (oldExecutor != null && !oldExecutor.isShutdown()) {
+ oldExecutor.shutdownNow();
+ }
+ ExecutorService freshExecutor = new
java.util.concurrent.ThreadPoolExecutor(
+ 2, 2, Integer.MAX_VALUE, TimeUnit.MILLISECONDS, new
LinkedBlockingQueue<>());
+ executorService = spy(freshExecutor);
executorServiceField.set(registryService, executorService);
+ // clear internal maps
+ Field clusterAddressMapField =
EtcdRegistryServiceImpl.class.getDeclaredField("clusterAddressMap");
+ clusterAddressMapField.setAccessible(true);
+ clusterAddressMapField.set(registryService, new ConcurrentHashMap<>());
+ Field listenerMapField =
EtcdRegistryServiceImpl.class.getDeclaredField("listenerMap");
+ listenerMapField.setAccessible(true);
+ listenerMapField.set(registryService, new ConcurrentHashMap<>());
+ Field watcherMapField =
EtcdRegistryServiceImpl.class.getDeclaredField("watcherMap");
+ watcherMapField.setAccessible(true);
+ watcherMapField.set(registryService, new ConcurrentHashMap<>());
+
// inject mock client
Field clientField =
EtcdRegistryServiceImpl.class.getDeclaredField("client");
clientField.setAccessible(true);
clientField.set(registryService, mockClient);
}
+ @org.junit.jupiter.api.AfterEach
+ public void tearDown() throws Exception {
+ Field executorServiceField =
EtcdRegistryServiceImpl.class.getDeclaredField("executorService");
+ executorServiceField.setAccessible(true);
+ ExecutorService executor = (ExecutorService)
executorServiceField.get(registryService);
+ if (executor != null && !executor.isShutdown()) {
+ executor.shutdownNow();
+ }
+ }
+
@BeforeAll
public static void beforeClass() {
String endPoint = String.format("http://%s:%s";, HOST, PORT);
diff --git
a/extensions/rpc/seata-brpc/src/main/java/org/apache/seata/integration/brpc/TransactionPropagationClientInterceptor.java
b/extensions/rpc/seata-brpc/src/main/java/org/apache/seata/integration/brpc/TransactionPropagationClientInterceptor.java
index 123be97b68..913bffcc22 100644
---
a/extensions/rpc/seata-brpc/src/main/java/org/apache/seata/integration/brpc/TransactionPropagationClientInterceptor.java
+++
b/extensions/rpc/seata-brpc/src/main/java/org/apache/seata/integration/brpc/TransactionPropagationClientInterceptor.java
@@ -20,55 +20,33 @@ import com.baidu.brpc.interceptor.AbstractInterceptor;
import com.baidu.brpc.interceptor.InterceptorChain;
import com.baidu.brpc.protocol.Request;
import com.baidu.brpc.protocol.Response;
-import org.apache.seata.integration.rpc.core.ConsumerRpcFilter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.seata.integration.rpc.core.TransactionPropagationHandler;
import java.util.HashMap;
import java.util.Map;
-/**
- * load SEATA xid for brpc request
- *
- */
-public class TransactionPropagationClientInterceptor extends
AbstractInterceptor implements ConsumerRpcFilter<Request> {
-
- private static final Logger LOGGER =
LoggerFactory.getLogger(TransactionPropagationClientInterceptor.class);
+public class TransactionPropagationClientInterceptor extends
AbstractInterceptor {
@Override
public void aroundProcess(Request brpcRequest, Response brpcResponse,
InterceptorChain chain) throws Exception {
-
- Map<String, String> rootContexts = getRootContexts();
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("SEATA-BRPC context:{}",
getJsonContext(rootContexts));
+ Map<String, String> context =
TransactionPropagationHandler.getTransactionPropagationContext();
+ if (!context.isEmpty()) {
+ Map<String, Object> kvAttachment = brpcRequest.getKvAttachment();
+ if (kvAttachment == null) {
+ kvAttachment = new HashMap<>();
+ brpcRequest.setKvAttachment(kvAttachment);
+ }
+ kvAttachment.putAll(context);
}
-
- if (null != getXidFromRootContexts(rootContexts)) {
- bindContextsToRequest(brpcRequest, rootContexts);
- }
-
try {
chain.intercept(brpcRequest, brpcResponse);
} finally {
- cleanRequestContexts(brpcRequest, rootContexts);
- }
- }
-
- @Override
- public void bindContextToRequest(Request rpcRequest, String key, String
value) {
- Map<String, Object> kvAttachment = rpcRequest.getKvAttachment();
- if (null == kvAttachment) {
- kvAttachment = new HashMap<>();
- rpcRequest.setKvAttachment(kvAttachment);
- }
- kvAttachment.put(key, value);
- }
-
- @Override
- public void cleanRequestContext(Request rpcRequest, String key) {
- Map<String, Object> requestAttachment = rpcRequest.getKvAttachment();
- if (null != requestAttachment) {
- requestAttachment.remove(key);
+ Map<String, Object> kvAttachment = brpcRequest.getKvAttachment();
+ if (kvAttachment != null) {
+ for (String key : context.keySet()) {
+ kvAttachment.remove(key);
+ }
+ }
}
}
}
diff --git
a/extensions/rpc/seata-brpc/src/main/java/org/apache/seata/integration/brpc/TransactionPropagationServerInterceptor.java
b/extensions/rpc/seata-brpc/src/main/java/org/apache/seata/integration/brpc/TransactionPropagationServerInterceptor.java
index 2cb2f37de6..efdd56e0e8 100644
---
a/extensions/rpc/seata-brpc/src/main/java/org/apache/seata/integration/brpc/TransactionPropagationServerInterceptor.java
+++
b/extensions/rpc/seata-brpc/src/main/java/org/apache/seata/integration/brpc/TransactionPropagationServerInterceptor.java
@@ -21,70 +21,40 @@ import com.baidu.brpc.interceptor.InterceptorChain;
import com.baidu.brpc.protocol.Request;
import com.baidu.brpc.protocol.Response;
import org.apache.seata.core.context.RootContext;
-import org.apache.seata.integration.rpc.core.ProviderRpcFilter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.seata.integration.rpc.core.TransactionPropagationHandler;
import java.util.Map;
-/**
- * <p>1. load SEATA xid from brpc request in handleRequest</p>
- * <p>2. clear SEATA xid when brpc request done in aroundProcess</p>
- *
- */
-public class TransactionPropagationServerInterceptor extends
AbstractInterceptor implements ProviderRpcFilter<Request> {
-
- private static final Logger LOGGER =
LoggerFactory.getLogger(TransactionPropagationServerInterceptor.class);
-
- @Override
- public boolean handleRequest(Request request) {
-
- Map<String, String> rpcContexts = getRpcContexts(request);
- String xid = RootContext.getXID();
- String rpcXid = getXidFromContexts(rpcContexts);
- if (null == xid) {
- if (null != rpcXid) {
- bindRequestToContexts(rpcContexts);
- if (LOGGER.isInfoEnabled()) {
- LOGGER.info("SEATA-BRPC bind {} to RootContext",
getJsonContext(rpcContexts));
- }
- }
- }
-
- return super.handleRequest(request);
- }
+public class TransactionPropagationServerInterceptor extends
AbstractInterceptor {
@Override
public void aroundProcess(Request brpcRequest, Response brpcResponse,
InterceptorChain chain) throws Exception {
-
+ String rpcXid = getXidFromRequest(brpcRequest);
+ String rpcBranchType = getAttachment(brpcRequest,
RootContext.KEY_BRANCH_TYPE);
+ boolean bound =
TransactionPropagationHandler.bindProviderContext(rpcXid, rpcBranchType);
try {
chain.intercept(brpcRequest, brpcResponse);
} finally {
- Map<String, String> rootContexts = cleanRootContexts();
- Map<String, String> rpcContexts = getRpcContexts(brpcRequest);
- String rpcXid = getXidFromContexts(rpcContexts);
- String xid = getXidFromContexts(rootContexts);
- if (LOGGER.isInfoEnabled()) {
- LOGGER.info("SEATA-BRPC unbind {} from RootContext",
getJsonContext(rootContexts));
- }
- if (null != rpcXid && !rpcXid.equalsIgnoreCase(xid)) {
- if (LOGGER.isWarnEnabled()) {
- LOGGER.warn(
- "SEATA-BRPC context changed during RPC from {} to
{},will be reset.",
- getJsonContext(rpcContexts),
- getJsonContext(rootContexts));
- }
- resetRootContexts(rootContexts);
+ if (bound) {
+ TransactionPropagationHandler.unbindProviderContext(rpcXid);
}
}
}
- @Override
- public String getRpcContext(Request rpcContext, String key) {
- if (null == rpcContext.getKvAttachment()) {
+ private String getXidFromRequest(Request request) {
+ String xid = getAttachment(request, RootContext.KEY_XID);
+ if (xid == null) {
+ xid = getAttachment(request, RootContext.KEY_XID.toLowerCase());
+ }
+ return xid;
+ }
+
+ private String getAttachment(Request request, String key) {
+ Map<String, Object> kvAttachment = request.getKvAttachment();
+ if (kvAttachment == null) {
return null;
}
- Object value = rpcContext.getKvAttachment().get(key);
+ Object value = kvAttachment.get(key);
return value == null ? null : value.toString();
}
}
diff --git a/extensions/rpc/seata-dubbo-alibaba/pom.xml
b/extensions/rpc/seata-dubbo-alibaba/pom.xml
index 17bbec54ea..91b7e05cf3 100644
--- a/extensions/rpc/seata-dubbo-alibaba/pom.xml
+++ b/extensions/rpc/seata-dubbo-alibaba/pom.xml
@@ -35,7 +35,7 @@
<dependencies>
<dependency>
<groupId>${project.groupId}</groupId>
- <artifactId>seata-tm</artifactId>
+ <artifactId>seata-rpc-core</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
diff --git
a/extensions/rpc/seata-dubbo-alibaba/src/main/java/org/apache/seata/integration/dubbo/alibaba/AlibabaDubboTransactionConsumerFilter.java
b/extensions/rpc/seata-dubbo-alibaba/src/main/java/org/apache/seata/integration/dubbo/alibaba/AlibabaDubboTransactionConsumerFilter.java
index 4549dfd5dc..6cabcf3dd9 100644
---
a/extensions/rpc/seata-dubbo-alibaba/src/main/java/org/apache/seata/integration/dubbo/alibaba/AlibabaDubboTransactionConsumerFilter.java
+++
b/extensions/rpc/seata-dubbo-alibaba/src/main/java/org/apache/seata/integration/dubbo/alibaba/AlibabaDubboTransactionConsumerFilter.java
@@ -24,58 +24,32 @@ import com.alibaba.dubbo.rpc.Result;
import com.alibaba.dubbo.rpc.RpcContext;
import com.alibaba.dubbo.rpc.RpcException;
import org.apache.seata.core.constants.DubboConstants;
-import org.apache.seata.core.context.RootContext;
-import org.apache.seata.core.model.BranchType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.seata.integration.rpc.core.TransactionPropagationHandler;
+
+import java.util.Map;
-/**
- * The type Alibaba dubbo transaction consumer filter.
- */
@Activate(
group = {DubboConstants.CONSUMER},
order = 100)
public class AlibabaDubboTransactionConsumerFilter implements Filter {
- private static final Logger LOGGER =
LoggerFactory.getLogger(AlibabaDubboTransactionConsumerFilter.class);
-
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws
RpcException {
if (!DubboConstants.ALIBABADUBBO) {
return invoker.invoke(invocation);
}
- return doInvoke(invoker, invocation);
- }
-
- private Result doInvoke(Invoker<?> invoker, Invocation invocation) throws
RpcException {
- String xid = RootContext.getXID();
- BranchType branchType = RootContext.getBranchType();
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("consumer xid in RootContext[{}], branchType in
RootContext[{}]", xid, branchType);
+ Map<String, String> context =
TransactionPropagationHandler.getTransactionPropagationContext();
+ if (!context.isEmpty()) {
+ for (Map.Entry<String, String> entry : context.entrySet()) {
+ RpcContext.getContext().setAttachment(entry.getKey(),
entry.getValue());
+ }
}
try {
- propagateTransactionContext(xid, branchType);
return invoker.invoke(invocation);
} finally {
- clearTransactionContext();
- }
- }
-
- private void propagateTransactionContext(String xid, BranchType
branchType) {
- if (xid != null) {
- RpcContext.getContext().setAttachment(RootContext.KEY_XID, xid);
- RpcContext.getContext().setAttachment(RootContext.KEY_BRANCH_TYPE,
branchType.name());
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("transaction context propagated: xid={},
branchType={}", xid, branchType);
+ for (String key : context.keySet()) {
+ RpcContext.getContext().removeAttachment(key);
}
}
}
-
- private void clearTransactionContext() {
- RpcContext.getContext().removeAttachment(RootContext.KEY_XID);
- RpcContext.getContext().removeAttachment(RootContext.KEY_BRANCH_TYPE);
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("transaction context cleared");
- }
- }
}
diff --git
a/extensions/rpc/seata-dubbo-alibaba/src/main/java/org/apache/seata/integration/dubbo/alibaba/AlibabaDubboTransactionProviderFilter.java
b/extensions/rpc/seata-dubbo-alibaba/src/main/java/org/apache/seata/integration/dubbo/alibaba/AlibabaDubboTransactionProviderFilter.java
index 7c8a4fddf9..89f5d84ac8 100644
---
a/extensions/rpc/seata-dubbo-alibaba/src/main/java/org/apache/seata/integration/dubbo/alibaba/AlibabaDubboTransactionProviderFilter.java
+++
b/extensions/rpc/seata-dubbo-alibaba/src/main/java/org/apache/seata/integration/dubbo/alibaba/AlibabaDubboTransactionProviderFilter.java
@@ -23,127 +23,34 @@ import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.Result;
import com.alibaba.dubbo.rpc.RpcContext;
import com.alibaba.dubbo.rpc.RpcException;
-import org.apache.seata.common.util.StringUtils;
import org.apache.seata.core.constants.DubboConstants;
import org.apache.seata.core.context.RootContext;
-import org.apache.seata.core.model.BranchType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.seata.integration.rpc.core.TransactionPropagationHandler;
-/**
- * The type Alibaba dubbo transaction provider filter.
- */
@Activate(
group = {DubboConstants.PROVIDER},
order = 100)
public class AlibabaDubboTransactionProviderFilter implements Filter {
- private static final Logger LOGGER =
LoggerFactory.getLogger(AlibabaDubboTransactionProviderFilter.class);
-
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws
RpcException {
if (!DubboConstants.ALIBABADUBBO) {
return invoker.invoke(invocation);
}
-
- return doInvoke(invoker, invocation);
- }
-
- private Result doInvoke(Invoker<?> invoker, Invocation invocation) throws
RpcException {
String rpcXid = getRpcXid();
String rpcBranchType =
RpcContext.getContext().getAttachment(RootContext.KEY_BRANCH_TYPE);
-
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("xid in RpcContext[{}], branchType in
RpcContext[{}]", rpcXid, rpcBranchType);
- }
-
- TransactionContextBinding binding = bindTransactionContext(rpcXid,
rpcBranchType);
-
+ boolean bound =
TransactionPropagationHandler.bindProviderContext(rpcXid, rpcBranchType);
try {
return invoker.invoke(invocation);
} finally {
- unbindTransactionContext(binding);
- clearServerContextAttachments();
- }
- }
-
- private TransactionContextBinding bindTransactionContext(String rpcXid,
String rpcBranchType) {
- TransactionContextBinding binding = new TransactionContextBinding();
-
- if (rpcXid != null) {
- RootContext.bind(rpcXid);
- binding.wasBound = true;
- binding.bindXid = rpcXid;
-
- if (StringUtils.equals(BranchType.TCC.name(), rpcBranchType)) {
- RootContext.bindBranchType(BranchType.TCC);
- binding.wasBranchTypeBound = true;
- }
-
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("bind xid [{}] branchType [{}] to RootContext",
rpcXid, rpcBranchType);
+ if (bound) {
+ TransactionPropagationHandler.unbindProviderContext(rpcXid);
}
+
RpcContext.getServerContext().removeAttachment(RootContext.KEY_XID);
+
RpcContext.getServerContext().removeAttachment(RootContext.KEY_BRANCH_TYPE);
}
-
- return binding;
}
- private void unbindTransactionContext(TransactionContextBinding binding) {
- if (!binding.wasBound) {
- return;
- }
-
- BranchType previousBranchType = RootContext.getBranchType();
- String unbindXid = RootContext.unbind();
- binding.unbindXid = unbindXid;
- binding.unbindBranchType = previousBranchType;
-
- if (binding.wasBranchTypeBound && BranchType.TCC ==
previousBranchType) {
- RootContext.unbindBranchType();
- }
-
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("unbind xid [{}] branchType [{}] from RootContext",
unbindXid, previousBranchType);
- }
-
- handleXidChange(binding);
- }
-
- private void handleXidChange(TransactionContextBinding binding) {
- if (!binding.bindXid.equalsIgnoreCase(binding.unbindXid)) {
- LOGGER.warn(
- "xid in change during RPC from {} to {},branchType from {}
to {}",
- binding.bindXid,
- binding.unbindXid,
- binding.bindBranchType != null ? binding.bindBranchType :
BranchType.AT,
- binding.unbindBranchType);
-
- if (binding.unbindXid != null) {
- restoreTransactionContext(binding);
- }
- }
- }
-
- private void restoreTransactionContext(TransactionContextBinding binding) {
- RootContext.bind(binding.unbindXid);
- LOGGER.warn("bind xid [{}] back to RootContext", binding.unbindXid);
-
- if (BranchType.TCC == binding.unbindBranchType) {
- RootContext.bindBranchType(BranchType.TCC);
- LOGGER.warn("bind branchType [{}] back to RootContext",
binding.unbindBranchType);
- }
- }
-
- private void clearServerContextAttachments() {
- RpcContext.getServerContext().removeAttachment(RootContext.KEY_XID);
-
RpcContext.getServerContext().removeAttachment(RootContext.KEY_BRANCH_TYPE);
- }
-
- /**
- * get rpc xid
- *
- * @return
- */
private String getRpcXid() {
String rpcXid =
RpcContext.getContext().getAttachment(RootContext.KEY_XID);
if (rpcXid == null) {
@@ -151,31 +58,4 @@ public class AlibabaDubboTransactionProviderFilter
implements Filter {
}
return rpcXid;
}
-
- private static class TransactionContextBinding {
- /**
- * The Was bound.
- */
- boolean wasBound = false;
- /**
- * The Was branch type bound.
- */
- boolean wasBranchTypeBound = false;
- /**
- * The Bind xid.
- */
- String bindXid;
- /**
- * The Bind branch type.
- */
- String bindBranchType;
- /**
- * The Unbind xid.
- */
- String unbindXid;
- /**
- * The Unbind branch type.
- */
- BranchType unbindBranchType;
- }
}
diff --git a/extensions/rpc/seata-grpc/pom.xml
b/extensions/rpc/seata-grpc/pom.xml
index 072f64afb4..ac78a50f68 100644
--- a/extensions/rpc/seata-grpc/pom.xml
+++ b/extensions/rpc/seata-grpc/pom.xml
@@ -35,7 +35,7 @@
<dependencies>
<dependency>
<groupId>${project.groupId}</groupId>
- <artifactId>seata-tm</artifactId>
+ <artifactId>seata-rpc-core</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
diff --git
a/extensions/rpc/seata-grpc/src/main/java/org/apache/seata/integration/grpc/interceptor/client/ClientTransactionInterceptor.java
b/extensions/rpc/seata-grpc/src/main/java/org/apache/seata/integration/grpc/interceptor/client/ClientTransactionInterceptor.java
index 51d7220c44..9843b76dc5 100644
---
a/extensions/rpc/seata-grpc/src/main/java/org/apache/seata/integration/grpc/interceptor/client/ClientTransactionInterceptor.java
+++
b/extensions/rpc/seata-grpc/src/main/java/org/apache/seata/integration/grpc/interceptor/client/ClientTransactionInterceptor.java
@@ -26,6 +26,9 @@ import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import org.apache.seata.core.context.RootContext;
import org.apache.seata.integration.grpc.interceptor.GrpcHeaderKey;
+import org.apache.seata.integration.rpc.core.TransactionPropagationHandler;
+
+import java.util.Map;
public class ClientTransactionInterceptor implements ClientInterceptor {
@@ -33,16 +36,20 @@ public class ClientTransactionInterceptor implements
ClientInterceptor {
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
MethodDescriptor<ReqT, RespT> method, CallOptions callOptions,
Channel next) {
- String xid = RootContext.getXID();
+ Map<String, String> context =
TransactionPropagationHandler.getTransactionPropagationContext();
return new ForwardingClientCall.SimpleForwardingClientCall<ReqT,
RespT>(next.newCall(method, callOptions)) {
@Override
public void start(Listener<RespT> responseListener, Metadata
headers) {
- if (xid != null) {
- headers.put(GrpcHeaderKey.XID_HEADER_KEY, xid);
- headers.put(
- GrpcHeaderKey.BRANCH_HEADER_KEY,
- RootContext.getBranchType().name());
+ if (!context.isEmpty()) {
+ String xid = context.get(RootContext.KEY_XID);
+ if (xid != null) {
+ headers.put(GrpcHeaderKey.XID_HEADER_KEY, xid);
+ }
+ String branchType =
context.get(RootContext.KEY_BRANCH_TYPE);
+ if (branchType != null) {
+ headers.put(GrpcHeaderKey.BRANCH_HEADER_KEY,
branchType);
+ }
}
super.start(
new
ForwardingClientCallListener.SimpleForwardingClientCallListener<RespT>(responseListener)
{
diff --git
a/extensions/rpc/seata-grpc/src/main/java/org/apache/seata/integration/grpc/interceptor/server/ServerListenerProxy.java
b/extensions/rpc/seata-grpc/src/main/java/org/apache/seata/integration/grpc/interceptor/server/ServerListenerProxy.java
index 408f4b21a8..71810188b4 100644
---
a/extensions/rpc/seata-grpc/src/main/java/org/apache/seata/integration/grpc/interceptor/server/ServerListenerProxy.java
+++
b/extensions/rpc/seata-grpc/src/main/java/org/apache/seata/integration/grpc/interceptor/server/ServerListenerProxy.java
@@ -17,56 +17,32 @@
package org.apache.seata.integration.grpc.interceptor.server;
import io.grpc.ServerCall;
-import org.apache.seata.common.util.StringUtils;
-import org.apache.seata.core.context.RootContext;
-import org.apache.seata.core.model.BranchType;
+import org.apache.seata.integration.rpc.core.TransactionPropagationHandler;
-import java.util.Map;
import java.util.Objects;
public class ServerListenerProxy<ReqT> extends ServerCall.Listener<ReqT> {
- private ServerCall.Listener<ReqT> target;
- private final String xid;
- private final Map<String, String> context;
+ private final ServerCall.Listener<ReqT> target;
+ private final String rpcXid;
+ private final String rpcBranchType;
- /**
- * Constructs a ServerListenerProxy.
- *
- * @param xid the global transaction id to bind
- * @param context the context map containing metadata such as branch type
- * @param target the original ServerCall.Listener to delegate calls to
- */
- public ServerListenerProxy(String xid, Map<String, String> context,
ServerCall.Listener<ReqT> target) {
- super();
+ public ServerListenerProxy(String rpcXid, String rpcBranchType,
ServerCall.Listener<ReqT> target) {
Objects.requireNonNull(target);
this.target = target;
- this.xid = xid;
- this.context = context;
+ this.rpcXid = rpcXid;
+ this.rpcBranchType = rpcBranchType;
}
- /**
- * Delegates onMessage call to the target listener.
- */
@Override
public void onMessage(ReqT message) {
target.onMessage(message);
}
- /**
- * Cleans up previous transaction context and binds new XID and branch
type (if applicable)
- * before delegating onHalfClose call to the target listener.
- */
@Override
public void onHalfClose() {
- cleanContext();
- if (StringUtils.isNotBlank(xid)) {
- RootContext.bind(xid);
- String branchType = context.get(RootContext.KEY_BRANCH_TYPE);
- if (StringUtils.equals(BranchType.TCC.name(), branchType)) {
- RootContext.bindBranchType(BranchType.TCC);
- }
- }
+ TransactionPropagationHandler.unbindProviderContext(null);
+ TransactionPropagationHandler.bindProviderContext(rpcXid,
rpcBranchType);
target.onHalfClose();
}
@@ -84,16 +60,4 @@ public class ServerListenerProxy<ReqT> extends
ServerCall.Listener<ReqT> {
public void onReady() {
target.onReady();
}
-
- /**
- * Cleans up the transaction context from RootContext to avoid thread
context pollution.
- * Unbinds XID and branch type if previously set.
- */
- private void cleanContext() {
- RootContext.unbind();
- BranchType previousBranchType = RootContext.getBranchType();
- if (BranchType.TCC == previousBranchType) {
- RootContext.unbindBranchType();
- }
- }
}
diff --git
a/extensions/rpc/seata-grpc/src/main/java/org/apache/seata/integration/grpc/interceptor/server/ServerTransactionInterceptor.java
b/extensions/rpc/seata-grpc/src/main/java/org/apache/seata/integration/grpc/interceptor/server/ServerTransactionInterceptor.java
index 39cea8fd7c..e242bab676 100644
---
a/extensions/rpc/seata-grpc/src/main/java/org/apache/seata/integration/grpc/interceptor/server/ServerTransactionInterceptor.java
+++
b/extensions/rpc/seata-grpc/src/main/java/org/apache/seata/integration/grpc/interceptor/server/ServerTransactionInterceptor.java
@@ -20,57 +20,21 @@ import io.grpc.Metadata;
import io.grpc.ServerCall;
import io.grpc.ServerCallHandler;
import io.grpc.ServerInterceptor;
-import org.apache.seata.core.context.RootContext;
import org.apache.seata.integration.grpc.interceptor.GrpcHeaderKey;
+import org.apache.seata.integration.rpc.core.TransactionPropagationHandler;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * ServerTransactionInterceptor intercepts incoming gRPC calls on the server
side
- * to extract global transaction context information (XID and branch type)
from request metadata,
- * and injects this context into a ServerListenerProxy to manage transaction
context lifecycle.
- */
public class ServerTransactionInterceptor implements ServerInterceptor {
- /**
- * Intercepts a gRPC call to extract transaction context and wrap the
ServerCall.Listener.
- *
- * @param serverCall the gRPC ServerCall object
- * @param metadata the request metadata (headers)
- * @param serverCallHandler the next handler in the interceptor chain
- * @param <ReqT> the request type
- * @param <RespT> the response type
- * @return a wrapped ServerCall.Listener that manages transaction context
- */
@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
ServerCall<ReqT, RespT> serverCall, Metadata metadata,
ServerCallHandler<ReqT, RespT> serverCallHandler) {
- String xid = getRpcXid(metadata);
- String branchName = getBranchName(metadata);
- Map<String, String> context = new HashMap<>();
- context.put(RootContext.KEY_BRANCH_TYPE, branchName);
- return new ServerListenerProxy<>(
- xid, Collections.unmodifiableMap(context),
serverCallHandler.startCall(serverCall, metadata));
+ String rpcXid = getRpcXid(metadata);
+ String rpcBranchType = metadata.get(GrpcHeaderKey.BRANCH_HEADER_KEY);
+ return new ServerListenerProxy<>(rpcXid, rpcBranchType,
serverCallHandler.startCall(serverCall, metadata));
}
- /**
- * Extracts the global transaction ID (XID) from metadata headers,
- * supporting both uppercase and lowercase keys.
- */
private String getRpcXid(Metadata metadata) {
- String rpcXid = metadata.get(GrpcHeaderKey.XID_HEADER_KEY);
- if (rpcXid == null) {
- rpcXid = metadata.get(GrpcHeaderKey.XID_HEADER_KEY_LOWERCASE);
- }
- return rpcXid;
- }
-
- /**
- * Extracts the branch transaction type name from metadata headers.
- */
- private String getBranchName(Metadata metadata) {
- return metadata.get(GrpcHeaderKey.BRANCH_HEADER_KEY);
+ return TransactionPropagationHandler.resolveXid(
+ metadata.get(GrpcHeaderKey.XID_HEADER_KEY),
metadata.get(GrpcHeaderKey.XID_HEADER_KEY_LOWERCASE));
}
}
diff --git
a/extensions/rpc/seata-grpc/src/test/java/org/apache/seata/integration/grpc/interceptor/server/ServerListenerProxyTest.java
b/extensions/rpc/seata-grpc/src/test/java/org/apache/seata/integration/grpc/interceptor/server/ServerListenerProxyTest.java
index 9495733fc2..9c59188a23 100644
---
a/extensions/rpc/seata-grpc/src/test/java/org/apache/seata/integration/grpc/interceptor/server/ServerListenerProxyTest.java
+++
b/extensions/rpc/seata-grpc/src/test/java/org/apache/seata/integration/grpc/interceptor/server/ServerListenerProxyTest.java
@@ -24,9 +24,6 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import java.util.HashMap;
-import java.util.Map;
-
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -61,18 +58,13 @@ class ServerListenerProxyTest {
@Test
void testOnHalfClose_withNonEmptyXid_andTCCBranchType_shouldBindContext() {
String xid = "test-xid";
- Map<String, String> context = new HashMap<>();
- context.put(RootContext.KEY_BRANCH_TYPE, BranchType.TCC.name());
-
- ServerListenerProxy<String> proxy = new ServerListenerProxy<>(xid,
context, target);
+ ServerListenerProxy<String> proxy = new ServerListenerProxy<>(xid,
BranchType.TCC.name(), target);
- // Pre-bind some context to test cleanup
RootContext.bind("old-xid");
RootContext.bindBranchType(BranchType.AT);
proxy.onHalfClose();
- // Verify RootContext binding updated
Assertions.assertEquals(xid, RootContext.getXID());
Assertions.assertEquals(BranchType.TCC, RootContext.getBranchType());
@@ -81,15 +73,13 @@ class ServerListenerProxyTest {
@Test
void testOnHalfClose_withEmptyXid_shouldOnlyCleanContext_andCallTarget() {
- ServerListenerProxy<String> proxy = new ServerListenerProxy<>(null,
new HashMap<>(), target);
+ ServerListenerProxy<String> proxy = new ServerListenerProxy<>(null,
null, target);
- // Pre-bind some context to test cleanup
RootContext.bind("old-xid");
RootContext.bindBranchType(BranchType.TCC);
proxy.onHalfClose();
- // Context should be cleaned (unbind XID and branch type)
Assertions.assertNull(RootContext.getXID());
Assertions.assertNull(RootContext.getBranchType());
diff --git
a/extensions/rpc/seata-grpc/src/test/java/org/apache/seata/integration/grpc/interceptor/server/ServerTransactionInterceptorTest.java
b/extensions/rpc/seata-grpc/src/test/java/org/apache/seata/integration/grpc/interceptor/server/ServerTransactionInterceptorTest.java
index bfc07a07d2..f96bd1a1c9 100644
---
a/extensions/rpc/seata-grpc/src/test/java/org/apache/seata/integration/grpc/interceptor/server/ServerTransactionInterceptorTest.java
+++
b/extensions/rpc/seata-grpc/src/test/java/org/apache/seata/integration/grpc/interceptor/server/ServerTransactionInterceptorTest.java
@@ -46,19 +46,16 @@ class ServerTransactionInterceptorTest {
@Test
void testInterceptCall_shouldExtractXidAndBranchTypeAndWrapListener() {
- // Ready
Metadata metadata = new Metadata();
metadata.put(GrpcHeaderKey.XID_HEADER_KEY, "test-xid");
metadata.put(GrpcHeaderKey.BRANCH_HEADER_KEY, "TCC");
- // Mocks
ServerCall<String, String> serverCall = mock(ServerCall.class);
ServerCallHandler<String, String> serverCallHandler =
mock(ServerCallHandler.class);
Listener<String> originalListener = mock(Listener.class);
when(serverCallHandler.startCall(serverCall,
metadata)).thenReturn(originalListener);
- // Call interceptor
ServerCall.Listener<String> listener =
interceptor.interceptCall(serverCall, metadata, serverCallHandler);
assertNotNull(listener);
@@ -78,17 +75,4 @@ class ServerTransactionInterceptorTest {
metadataLower.put(GrpcHeaderKey.XID_HEADER_KEY_LOWERCASE, "lower-xid");
assertEquals("lower-xid", getRpcXidMethod.invoke(interceptor,
metadataLower));
}
-
- @Test
- void testGetBranchName_shouldReturnCorrectValue() throws Exception {
- Metadata metadata = new Metadata();
- metadata.put(GrpcHeaderKey.BRANCH_HEADER_KEY, "branch-type");
-
- Method getBranchNameMethod =
interceptor.getClass().getDeclaredMethod("getBranchName", Metadata.class);
- getBranchNameMethod.setAccessible(true);
-
- String branchName = (String) getBranchNameMethod.invoke(interceptor,
metadata);
-
- assertEquals("branch-type", branchName);
- }
}
diff --git a/extensions/rpc/seata-hsf/pom.xml b/extensions/rpc/seata-hsf/pom.xml
index 95d4aa6940..452ccd2643 100644
--- a/extensions/rpc/seata-hsf/pom.xml
+++ b/extensions/rpc/seata-hsf/pom.xml
@@ -35,7 +35,7 @@
<dependencies>
<dependency>
<groupId>${project.groupId}</groupId>
- <artifactId>seata-tm</artifactId>
+ <artifactId>seata-rpc-core</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
diff --git
a/extensions/rpc/seata-hsf/src/main/java/org/apache/seata/integration/hsf/HsfTransactionConsumerFilter.java
b/extensions/rpc/seata-hsf/src/main/java/org/apache/seata/integration/hsf/HsfTransactionConsumerFilter.java
index 418b5a0865..fa0ed20219 100644
---
a/extensions/rpc/seata-hsf/src/main/java/org/apache/seata/integration/hsf/HsfTransactionConsumerFilter.java
+++
b/extensions/rpc/seata-hsf/src/main/java/org/apache/seata/integration/hsf/HsfTransactionConsumerFilter.java
@@ -22,79 +22,29 @@ import com.taobao.hsf.invocation.InvocationHandler;
import com.taobao.hsf.invocation.RPCResult;
import com.taobao.hsf.invocation.filter.ClientFilter;
import com.taobao.hsf.util.concurrent.ListenableFuture;
-import org.apache.seata.core.context.RootContext;
-import org.apache.seata.core.model.BranchType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.seata.integration.rpc.core.TransactionPropagationHandler;
-/**
- * The type Hsf transaction consumer filter.
- */
-public class HsfTransactionConsumerFilter implements ClientFilter {
+import java.util.Map;
- private static final Logger LOGGER =
LoggerFactory.getLogger(HsfTransactionConsumerFilter.class);
+public class HsfTransactionConsumerFilter implements ClientFilter {
@Override
public ListenableFuture<RPCResult> invoke(InvocationHandler nextHandler,
Invocation invocation) throws Throwable {
- return doInvoke(nextHandler, invocation);
- }
-
- private ListenableFuture<RPCResult> doInvoke(InvocationHandler
nextHandler, Invocation invocation)
- throws Throwable {
- TransactionContext context = extractTransactionContext();
-
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("xid in RootContext[{}], branchType in
RootContext[{}]", context.xid, context.branchType);
+ Map<String, String> context =
TransactionPropagationHandler.getTransactionPropagationContext();
+ if (!context.isEmpty()) {
+ for (Map.Entry<String, String> entry : context.entrySet()) {
+ RPCContext.getClientContext().putAttachment(entry.getKey(),
entry.getValue());
+ }
}
-
try {
- propagateTransactionContext(context);
return nextHandler.invoke(invocation);
} finally {
- clearTransactionContext();
- }
- }
-
- private TransactionContext extractTransactionContext() {
- TransactionContext context = new TransactionContext();
- context.xid = RootContext.getXID();
- context.branchType = RootContext.getBranchType();
- return context;
- }
-
- private void propagateTransactionContext(TransactionContext context) {
- if (context.xid != null) {
- RPCContext.getClientContext().putAttachment(RootContext.KEY_XID,
context.xid);
-
RPCContext.getClientContext().putAttachment(RootContext.KEY_BRANCH_TYPE,
context.branchType.name());
-
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("transaction context propagated: xid={},
branchType={}", context.xid, context.branchType);
+ for (String key : context.keySet()) {
+ RPCContext.getClientContext().removeAttachment(key);
}
}
}
- private void clearTransactionContext() {
- RPCContext.getClientContext().removeAttachment(RootContext.KEY_XID);
-
RPCContext.getClientContext().removeAttachment(RootContext.KEY_BRANCH_TYPE);
-
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("transaction context cleared");
- }
- }
-
@Override
- public void onResponse(Invocation invocation, RPCResult rpcResult) {
- // No operation needed
- }
-
- private static class TransactionContext {
- /**
- * The Xid.
- */
- String xid;
- /**
- * The Branch type.
- */
- BranchType branchType;
- }
+ public void onResponse(Invocation invocation, RPCResult rpcResult) {}
}
diff --git
a/extensions/rpc/seata-hsf/src/main/java/org/apache/seata/integration/hsf/HsfTransactionProviderFilter.java
b/extensions/rpc/seata-hsf/src/main/java/org/apache/seata/integration/hsf/HsfTransactionProviderFilter.java
index b87d59a475..c17dcb70c5 100644
---
a/extensions/rpc/seata-hsf/src/main/java/org/apache/seata/integration/hsf/HsfTransactionProviderFilter.java
+++
b/extensions/rpc/seata-hsf/src/main/java/org/apache/seata/integration/hsf/HsfTransactionProviderFilter.java
@@ -22,165 +22,40 @@ import com.taobao.hsf.invocation.InvocationHandler;
import com.taobao.hsf.invocation.RPCResult;
import com.taobao.hsf.invocation.filter.ServerFilter;
import com.taobao.hsf.util.concurrent.ListenableFuture;
-import org.apache.seata.common.util.StringUtils;
import org.apache.seata.core.context.RootContext;
-import org.apache.seata.core.model.BranchType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.seata.integration.rpc.core.TransactionPropagationHandler;
-/**
- * The type Hsf transaction provider filter.
- */
public class HsfTransactionProviderFilter implements ServerFilter {
- private static final Logger LOGGER =
LoggerFactory.getLogger(HsfTransactionProviderFilter.class);
-
@Override
public ListenableFuture<RPCResult> invoke(InvocationHandler nextHandler,
Invocation invocation) throws Throwable {
- return doInvoke(nextHandler, invocation);
- }
-
- private ListenableFuture<RPCResult> doInvoke(InvocationHandler
nextHandler, Invocation invocation)
- throws Throwable {
- RpcTransactionContext rpcContext = extractRpcTransactionContext();
-
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug(
- "xid in RpcContext[{}], branchType in RpcContext[{}]",
rpcContext.rpcXid, rpcContext.rpcBranchType);
- }
-
- TransactionContextBinding binding = bindTransactionContext(rpcContext);
-
+ String rpcXid = getRpcXid();
+ String rpcBranchType =
getStringAttachment(RootContext.KEY_BRANCH_TYPE);
+ boolean bound =
TransactionPropagationHandler.bindProviderContext(rpcXid, rpcBranchType);
try {
return nextHandler.invoke(invocation);
} finally {
- unbindTransactionContext(binding);
- clearServerContextAttachments();
- }
- }
-
- private RpcTransactionContext extractRpcTransactionContext() {
- RpcTransactionContext context = new RpcTransactionContext();
- context.rpcXid =
RPCContext.getServerContext().getAttachment(RootContext.KEY_XID);
- context.rpcBranchType =
RPCContext.getServerContext().getAttachment(RootContext.KEY_BRANCH_TYPE);
- return context;
- }
-
- private TransactionContextBinding
bindTransactionContext(RpcTransactionContext rpcContext) {
- TransactionContextBinding binding = new TransactionContextBinding();
-
- if (rpcContext.rpcXid != null) {
- String xidStr = rpcContext.rpcXid.toString();
- RootContext.bind(xidStr);
- binding.wasBound = true;
- binding.bindXid = xidStr;
-
- if (rpcContext.rpcBranchType != null
- && StringUtils.equals(BranchType.TCC.name(),
rpcContext.rpcBranchType.toString())) {
- RootContext.bindBranchType(BranchType.TCC);
- binding.wasBranchTypeBound = true;
- }
-
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug(
- "bind xid [{}] branchType [{}] to RootContext",
rpcContext.rpcXid, rpcContext.rpcBranchType);
- }
- }
-
- return binding;
- }
-
- private void unbindTransactionContext(TransactionContextBinding binding) {
- if (!binding.wasBound) {
- return;
- }
-
- BranchType previousBranchType = RootContext.getBranchType();
- String unbindXid = RootContext.unbind();
- binding.unbindXid = unbindXid;
- binding.unbindBranchType = previousBranchType;
-
- if (binding.wasBranchTypeBound && BranchType.TCC ==
previousBranchType) {
- RootContext.unbindBranchType();
- }
-
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("unbind xid [{}] branchType [{}] from RootContext",
unbindXid, previousBranchType);
- }
-
- handleXidChange(binding);
- }
-
- private void handleXidChange(TransactionContextBinding binding) {
- if (!binding.bindXid.equalsIgnoreCase(binding.unbindXid)) {
- LOGGER.warn(
- "xid in change during RPC from {} to {},branchType from {}
to {}",
- binding.bindXid,
- binding.unbindXid,
- binding.bindBranchType != null ? binding.bindBranchType :
"AT",
- binding.unbindBranchType);
-
- if (binding.unbindXid != null) {
- restoreTransactionContext(binding);
+ if (bound) {
+ TransactionPropagationHandler.unbindProviderContext(rpcXid);
}
+
RPCContext.getServerContext().removeAttachment(RootContext.KEY_XID);
+
RPCContext.getServerContext().removeAttachment(RootContext.KEY_BRANCH_TYPE);
}
}
- private void restoreTransactionContext(TransactionContextBinding binding) {
- RootContext.bind(binding.unbindXid);
- LOGGER.warn("bind xid [{}] back to RootContext", binding.unbindXid);
-
- if (BranchType.TCC == binding.unbindBranchType) {
- RootContext.bindBranchType(BranchType.TCC);
- LOGGER.warn("bind branchType [{}] back to RootContext",
binding.unbindBranchType);
+ private String getRpcXid() {
+ String rpcXid = getStringAttachment(RootContext.KEY_XID);
+ if (rpcXid == null) {
+ rpcXid = getStringAttachment(RootContext.KEY_XID.toLowerCase());
}
+ return rpcXid;
}
- private void clearServerContextAttachments() {
- RPCContext.getServerContext().removeAttachment(RootContext.KEY_XID);
-
RPCContext.getServerContext().removeAttachment(RootContext.KEY_BRANCH_TYPE);
+ private String getStringAttachment(String key) {
+ Object value = RPCContext.getServerContext().getAttachment(key);
+ return value == null ? null : value.toString();
}
@Override
- public void onResponse(Invocation invocation, RPCResult rpcResult) {
- // No operation needed
- }
-
- private static class RpcTransactionContext {
- /**
- * The Rpc xid.
- */
- Object rpcXid;
- /**
- * The Rpc branch type.
- */
- Object rpcBranchType;
- }
-
- private static class TransactionContextBinding {
- /**
- * The Was bound.
- */
- boolean wasBound = false;
- /**
- * The Was branch type bound.
- */
- boolean wasBranchTypeBound = false;
- /**
- * The Bind xid.
- */
- String bindXid;
- /**
- * The Bind branch type.
- */
- String bindBranchType;
- /**
- * The Unbind xid.
- */
- String unbindXid;
- /**
- * The Unbind branch type.
- */
- BranchType unbindBranchType;
- }
+ public void onResponse(Invocation invocation, RPCResult rpcResult) {}
}
diff --git
a/extensions/rpc/seata-hsf/src/test/java/org/apache/seata/integration/hsf/HsfTransactionFilterTest.java
b/extensions/rpc/seata-hsf/src/test/java/org/apache/seata/integration/hsf/HsfTransactionFilterTest.java
new file mode 100644
index 0000000000..e192086f6a
--- /dev/null
+++
b/extensions/rpc/seata-hsf/src/test/java/org/apache/seata/integration/hsf/HsfTransactionFilterTest.java
@@ -0,0 +1,190 @@
+/*
+ * 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.hsf;
+
+import com.taobao.hsf.context.RPCContext;
+import com.taobao.hsf.invocation.Invocation;
+import com.taobao.hsf.invocation.InvocationHandler;
+import com.taobao.hsf.invocation.RPCResult;
+import com.taobao.hsf.util.concurrent.ListenableFuture;
+import org.apache.seata.core.context.RootContext;
+import org.apache.seata.core.model.BranchType;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+class HsfTransactionFilterTest {
+
+ private static final String DEFAULT_XID = "127.0.0.1:8091:12345678";
+
+ @AfterEach
+ void cleanup() {
+ RootContext.unbind();
+ RootContext.unbindBranchType();
+ }
+
+ @Test
+ void testConsumerFilter_propagatesXid() throws Throwable {
+ RootContext.bind(DEFAULT_XID);
+ RootContext.bindBranchType(BranchType.TCC);
+
+ Map<String, Object> clientAttachments = new HashMap<>();
+ RPCContext mockClientContext = mock(RPCContext.class);
+ when(mockClientContext.putAttachment(
+ org.mockito.ArgumentMatchers.anyString(),
org.mockito.ArgumentMatchers.any()))
+ .thenAnswer(inv -> {
+ clientAttachments.put(inv.getArgument(0),
inv.getArgument(1));
+ return mockClientContext;
+ });
+
when(mockClientContext.removeAttachment(org.mockito.ArgumentMatchers.anyString()))
+ .thenAnswer(inv -> {
+ clientAttachments.remove((String) inv.getArgument(0));
+ return mockClientContext;
+ });
+
+ InvocationHandler mockHandler = mock(InvocationHandler.class);
+ @SuppressWarnings("unchecked")
+ ListenableFuture<RPCResult> mockFuture = mock(ListenableFuture.class);
+
when(mockHandler.invoke(org.mockito.ArgumentMatchers.any())).thenAnswer(inv -> {
+ assertThat(clientAttachments).containsEntry(RootContext.KEY_XID,
DEFAULT_XID);
+
assertThat(clientAttachments).containsEntry(RootContext.KEY_BRANCH_TYPE,
BranchType.TCC.name());
+ return mockFuture;
+ });
+
+ HsfTransactionConsumerFilter filter = new
HsfTransactionConsumerFilter();
+ try (MockedStatic<RPCContext> rpcContextMock =
mockStatic(RPCContext.class)) {
+
rpcContextMock.when(RPCContext::getClientContext).thenReturn(mockClientContext);
+ filter.invoke(mockHandler, new Invocation());
+ }
+
+ assertThat(clientAttachments).isEmpty();
+ }
+
+ @Test
+ void testConsumerFilter_noXid() throws Throwable {
+ Map<String, Object> clientAttachments = new HashMap<>();
+ RPCContext mockClientContext = mock(RPCContext.class);
+ when(mockClientContext.putAttachment(
+ org.mockito.ArgumentMatchers.anyString(),
org.mockito.ArgumentMatchers.any()))
+ .thenAnswer(inv -> {
+ clientAttachments.put(inv.getArgument(0),
inv.getArgument(1));
+ return mockClientContext;
+ });
+
+ InvocationHandler mockHandler = mock(InvocationHandler.class);
+ @SuppressWarnings("unchecked")
+ ListenableFuture<RPCResult> mockFuture = mock(ListenableFuture.class);
+
when(mockHandler.invoke(org.mockito.ArgumentMatchers.any())).thenReturn(mockFuture);
+
+ HsfTransactionConsumerFilter filter = new
HsfTransactionConsumerFilter();
+ try (MockedStatic<RPCContext> rpcContextMock =
mockStatic(RPCContext.class)) {
+
rpcContextMock.when(RPCContext::getClientContext).thenReturn(mockClientContext);
+ filter.invoke(mockHandler, new Invocation());
+ }
+
+ assertThat(clientAttachments).isEmpty();
+ }
+
+ @Test
+ void testProviderFilter_bindsXid() throws Throwable {
+ Map<String, Object> serverAttachments = new HashMap<>();
+ serverAttachments.put(RootContext.KEY_XID, DEFAULT_XID);
+ serverAttachments.put(RootContext.KEY_BRANCH_TYPE,
BranchType.TCC.name());
+
+ RPCContext mockServerContext = mock(RPCContext.class);
+
when(mockServerContext.getAttachment(RootContext.KEY_XID)).thenReturn(DEFAULT_XID);
+
when(mockServerContext.getAttachment(RootContext.KEY_BRANCH_TYPE)).thenReturn(BranchType.TCC.name());
+
when(mockServerContext.removeAttachment(org.mockito.ArgumentMatchers.anyString()))
+ .thenReturn(mockServerContext);
+
+ InvocationHandler mockHandler = mock(InvocationHandler.class);
+ @SuppressWarnings("unchecked")
+ ListenableFuture<RPCResult> mockFuture = mock(ListenableFuture.class);
+
when(mockHandler.invoke(org.mockito.ArgumentMatchers.any())).thenAnswer(inv -> {
+ assertThat(RootContext.getXID()).isEqualTo(DEFAULT_XID);
+ assertThat(RootContext.getBranchType()).isEqualTo(BranchType.TCC);
+ return mockFuture;
+ });
+
+ HsfTransactionProviderFilter filter = new
HsfTransactionProviderFilter();
+ try (MockedStatic<RPCContext> rpcContextMock =
mockStatic(RPCContext.class)) {
+
rpcContextMock.when(RPCContext::getServerContext).thenReturn(mockServerContext);
+ filter.invoke(mockHandler, new Invocation());
+ }
+
+ assertThat(RootContext.getXID()).isNull();
+ }
+
+ @Test
+ void testProviderFilter_noXid() throws Throwable {
+ RPCContext mockServerContext = mock(RPCContext.class);
+
when(mockServerContext.getAttachment(org.mockito.ArgumentMatchers.anyString()))
+ .thenReturn(null);
+
when(mockServerContext.removeAttachment(org.mockito.ArgumentMatchers.anyString()))
+ .thenReturn(mockServerContext);
+
+ InvocationHandler mockHandler = mock(InvocationHandler.class);
+ @SuppressWarnings("unchecked")
+ ListenableFuture<RPCResult> mockFuture = mock(ListenableFuture.class);
+
when(mockHandler.invoke(org.mockito.ArgumentMatchers.any())).thenAnswer(inv -> {
+ assertThat(RootContext.getXID()).isNull();
+ return mockFuture;
+ });
+
+ HsfTransactionProviderFilter filter = new
HsfTransactionProviderFilter();
+ try (MockedStatic<RPCContext> rpcContextMock =
mockStatic(RPCContext.class)) {
+
rpcContextMock.when(RPCContext::getServerContext).thenReturn(mockServerContext);
+ filter.invoke(mockHandler, new Invocation());
+ }
+
+ assertThat(RootContext.getXID()).isNull();
+ }
+
+ @Test
+ void testProviderFilter_xidFromLowercaseKey() throws Throwable {
+ RPCContext mockServerContext = mock(RPCContext.class);
+
when(mockServerContext.getAttachment(RootContext.KEY_XID)).thenReturn(null);
+
when(mockServerContext.getAttachment(RootContext.KEY_XID.toLowerCase())).thenReturn(DEFAULT_XID);
+
when(mockServerContext.getAttachment(RootContext.KEY_BRANCH_TYPE)).thenReturn(null);
+
when(mockServerContext.removeAttachment(org.mockito.ArgumentMatchers.anyString()))
+ .thenReturn(mockServerContext);
+
+ InvocationHandler mockHandler = mock(InvocationHandler.class);
+ @SuppressWarnings("unchecked")
+ ListenableFuture<RPCResult> mockFuture = mock(ListenableFuture.class);
+
when(mockHandler.invoke(org.mockito.ArgumentMatchers.any())).thenAnswer(inv -> {
+ assertThat(RootContext.getXID()).isEqualTo(DEFAULT_XID);
+ return mockFuture;
+ });
+
+ HsfTransactionProviderFilter filter = new
HsfTransactionProviderFilter();
+ try (MockedStatic<RPCContext> rpcContextMock =
mockStatic(RPCContext.class)) {
+
rpcContextMock.when(RPCContext::getServerContext).thenReturn(mockServerContext);
+ filter.invoke(mockHandler, new Invocation());
+ }
+
+ assertThat(RootContext.getXID()).isNull();
+ }
+}
diff --git a/extensions/rpc/seata-http-jakarta/pom.xml
b/extensions/rpc/seata-http-jakarta/pom.xml
index 1b9fb5abc3..a80da7d594 100644
--- a/extensions/rpc/seata-http-jakarta/pom.xml
+++ b/extensions/rpc/seata-http-jakarta/pom.xml
@@ -50,5 +50,11 @@
<artifactId>jakarta.servlet-api</artifactId>
<scope>provided</scope>
</dependency>
+
+ <dependency>
+ <groupId>javax.servlet</groupId>
+ <artifactId>javax.servlet-api</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
\ No newline at end of file
diff --git
a/extensions/rpc/seata-http-jakarta/src/test/java/org/apache/seata/integration/http/JakartaSeataWebMvcConfigurerTest.java
b/extensions/rpc/seata-http-jakarta/src/test/java/org/apache/seata/integration/http/JakartaSeataWebMvcConfigurerTest.java
new file mode 100644
index 0000000000..5e3f98ae3a
--- /dev/null
+++
b/extensions/rpc/seata-http-jakarta/src/test/java/org/apache/seata/integration/http/JakartaSeataWebMvcConfigurerTest.java
@@ -0,0 +1,42 @@
+/*
+ * 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.http;
+
+import
org.apache.seata.integration.http.jakarta.JakartaTransactionPropagationInterceptor;
+import org.junit.jupiter.api.Test;
+import
org.springframework.web.servlet.config.annotation.InterceptorRegistration;
+import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class JakartaSeataWebMvcConfigurerTest {
+
+ @Test
+ void testAddInterceptors() {
+ JakartaSeataWebMvcConfigurer configurer = new
JakartaSeataWebMvcConfigurer();
+ InterceptorRegistry registry = mock(InterceptorRegistry.class);
+ InterceptorRegistration registration =
mock(InterceptorRegistration.class);
+ when(registry.addInterceptor(any())).thenReturn(registration);
+
+ configurer.addInterceptors(registry);
+
+
verify(registry).addInterceptor(any(JakartaTransactionPropagationInterceptor.class));
+ }
+}
diff --git
a/extensions/rpc/seata-http-jakarta/src/test/java/org/apache/seata/integration/http/jakarta/JakartaTransactionPropagationInterceptorTest.java
b/extensions/rpc/seata-http-jakarta/src/test/java/org/apache/seata/integration/http/jakarta/JakartaTransactionPropagationInterceptorTest.java
new file mode 100644
index 0000000000..42d4125c34
--- /dev/null
+++
b/extensions/rpc/seata-http-jakarta/src/test/java/org/apache/seata/integration/http/jakarta/JakartaTransactionPropagationInterceptorTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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.http.jakarta;
+
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import org.apache.seata.core.context.RootContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class JakartaTransactionPropagationInterceptorTest {
+
+ private static final String DEFAULT_XID = "127.0.0.1:8091:12345678";
+
+ @AfterEach
+ void cleanup() {
+ RootContext.unbind();
+ }
+
+ @Test
+ void testPreHandle_bindsXid() throws Exception {
+ JakartaTransactionPropagationInterceptor interceptor = new
JakartaTransactionPropagationInterceptor();
+ HttpServletRequest request = mock(HttpServletRequest.class);
+ HttpServletResponse response = mock(HttpServletResponse.class);
+ when(request.getHeader(RootContext.KEY_XID)).thenReturn(DEFAULT_XID);
+
+ boolean result = interceptor.preHandle(request, response, new
Object());
+
+ assertThat(result).isTrue();
+ assertThat(RootContext.getXID()).isEqualTo(DEFAULT_XID);
+ }
+
+ @Test
+ void testPreHandle_noXid() throws Exception {
+ JakartaTransactionPropagationInterceptor interceptor = new
JakartaTransactionPropagationInterceptor();
+ HttpServletRequest request = mock(HttpServletRequest.class);
+ HttpServletResponse response = mock(HttpServletResponse.class);
+ when(request.getHeader(RootContext.KEY_XID)).thenReturn(null);
+
+ boolean result = interceptor.preHandle(request, response, new
Object());
+
+ assertThat(result).isTrue();
+ assertThat(RootContext.getXID()).isNull();
+ }
+
+ @Test
+ void testPreHandle_doesNotOverrideExistingXid() throws Exception {
+ RootContext.bind("existing-xid");
+ JakartaTransactionPropagationInterceptor interceptor = new
JakartaTransactionPropagationInterceptor();
+ HttpServletRequest request = mock(HttpServletRequest.class);
+ HttpServletResponse response = mock(HttpServletResponse.class);
+ when(request.getHeader(RootContext.KEY_XID)).thenReturn(DEFAULT_XID);
+
+ boolean result = interceptor.preHandle(request, response, new
Object());
+
+ assertThat(result).isTrue();
+ assertThat(RootContext.getXID()).isEqualTo("existing-xid");
+ }
+
+ @Test
+ void testAfterCompletion_unbindsXid() throws Exception {
+ RootContext.bind(DEFAULT_XID);
+ JakartaTransactionPropagationInterceptor interceptor = new
JakartaTransactionPropagationInterceptor();
+ HttpServletRequest request = mock(HttpServletRequest.class);
+ HttpServletResponse response = mock(HttpServletResponse.class);
+ when(request.getHeader(RootContext.KEY_XID)).thenReturn(DEFAULT_XID);
+
+ interceptor.afterCompletion(request, response, new Object(), null);
+
+ assertThat(RootContext.getXID()).isNull();
+ }
+
+ @Test
+ void testAfterCompletion_noXidInContext() throws Exception {
+ JakartaTransactionPropagationInterceptor interceptor = new
JakartaTransactionPropagationInterceptor();
+ HttpServletRequest request = mock(HttpServletRequest.class);
+ HttpServletResponse response = mock(HttpServletResponse.class);
+ when(request.getHeader(RootContext.KEY_XID)).thenReturn(DEFAULT_XID);
+
+ interceptor.afterCompletion(request, response, new Object(), null);
+
+ assertThat(RootContext.getXID()).isNull();
+ }
+
+ @Test
+ void testFullLifecycle() throws Exception {
+ JakartaTransactionPropagationInterceptor interceptor = new
JakartaTransactionPropagationInterceptor();
+ HttpServletRequest request = mock(HttpServletRequest.class);
+ HttpServletResponse response = mock(HttpServletResponse.class);
+ when(request.getHeader(RootContext.KEY_XID)).thenReturn(DEFAULT_XID);
+
+ interceptor.preHandle(request, response, new Object());
+ assertThat(RootContext.getXID()).isEqualTo(DEFAULT_XID);
+
+ interceptor.afterCompletion(request, response, new Object(), null);
+ assertThat(RootContext.getXID()).isNull();
+ }
+}
diff --git a/extensions/rpc/seata-motan/pom.xml
b/extensions/rpc/seata-motan/pom.xml
index b9a50146c8..158624e63e 100644
--- a/extensions/rpc/seata-motan/pom.xml
+++ b/extensions/rpc/seata-motan/pom.xml
@@ -35,7 +35,7 @@
<dependencies>
<dependency>
<groupId>${project.groupId}</groupId>
- <artifactId>seata-tm</artifactId>
+ <artifactId>seata-rpc-core</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
diff --git
a/extensions/rpc/seata-motan/src/main/java/org/apache/seata/integration/motan/MotanTransactionConsumerFilter.java
b/extensions/rpc/seata-motan/src/main/java/org/apache/seata/integration/motan/MotanTransactionConsumerFilter.java
new file mode 100644
index 0000000000..4ac2ff9d7d
--- /dev/null
+++
b/extensions/rpc/seata-motan/src/main/java/org/apache/seata/integration/motan/MotanTransactionConsumerFilter.java
@@ -0,0 +1,51 @@
+/*
+ * 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.motan;
+
+import com.weibo.api.motan.common.MotanConstants;
+import com.weibo.api.motan.core.extension.Activation;
+import com.weibo.api.motan.core.extension.Scope;
+import com.weibo.api.motan.core.extension.Spi;
+import com.weibo.api.motan.filter.Filter;
+import com.weibo.api.motan.rpc.Caller;
+import com.weibo.api.motan.rpc.Request;
+import com.weibo.api.motan.rpc.Response;
+import org.apache.seata.integration.rpc.core.TransactionPropagationHandler;
+
+import java.util.Map;
+
+@Spi(scope = Scope.SINGLETON)
+@Activation(
+ key = {MotanConstants.NODE_TYPE_REFERER},
+ sequence = 100)
+public class MotanTransactionConsumerFilter implements Filter {
+
+ @Override
+ public Response filter(final Caller<?> caller, final Request request) {
+ Map<String, String> context =
TransactionPropagationHandler.getTransactionPropagationContext();
+ if (!context.isEmpty()) {
+ request.getAttachments().putAll(context);
+ }
+ try {
+ return caller.call(request);
+ } finally {
+ for (String key : context.keySet()) {
+ request.getAttachments().remove(key);
+ }
+ }
+ }
+}
diff --git
a/extensions/rpc/seata-motan/src/main/java/org/apache/seata/integration/motan/MotanTransactionFilter.java
b/extensions/rpc/seata-motan/src/main/java/org/apache/seata/integration/motan/MotanTransactionFilter.java
deleted file mode 100644
index d91ded6810..0000000000
---
a/extensions/rpc/seata-motan/src/main/java/org/apache/seata/integration/motan/MotanTransactionFilter.java
+++ /dev/null
@@ -1,114 +0,0 @@
-/*
- * 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.motan;
-
-import com.weibo.api.motan.common.MotanConstants;
-import com.weibo.api.motan.core.extension.Activation;
-import com.weibo.api.motan.core.extension.Scope;
-import com.weibo.api.motan.core.extension.Spi;
-import com.weibo.api.motan.filter.Filter;
-import com.weibo.api.motan.rpc.Caller;
-import com.weibo.api.motan.rpc.Request;
-import com.weibo.api.motan.rpc.Response;
-import org.apache.seata.common.util.StringUtils;
-import org.apache.seata.core.context.RootContext;
-import org.apache.seata.core.model.BranchType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@Spi(scope = Scope.SINGLETON)
-@Activation(
- key = {MotanConstants.NODE_TYPE_SERVICE,
MotanConstants.NODE_TYPE_REFERER},
- sequence = 100)
-public class MotanTransactionFilter implements Filter {
- private static final Logger LOGGER =
LoggerFactory.getLogger(MotanTransactionFilter.class);
-
- public MotanTransactionFilter() {}
-
- @Override
- public Response filter(final Caller<?> caller, final Request request) {
- String currentXid = RootContext.getXID();
- BranchType branchType = RootContext.getBranchType();
- String requestXid = getRpcXid(request);
- String rpcBranchType = getBranchType(request);
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug(
- "context in RootContext[{},{}], context in
RpcContext[{},{}]",
- currentXid,
- branchType,
- requestXid,
- rpcBranchType);
- }
- boolean bind = false;
- if (currentXid != null) {
- request.getAttachments().put(RootContext.KEY_XID, currentXid);
- request.getAttachments().put(RootContext.KEY_BRANCH_TYPE,
branchType.name());
-
- } else if (requestXid != null) {
- RootContext.bind(requestXid);
- if (StringUtils.equals(BranchType.TCC.name(), rpcBranchType)) {
- RootContext.bindBranchType(BranchType.TCC);
- }
- bind = true;
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("bind [{}] to RootContext", requestXid);
- }
- }
- try {
- return caller.call(request);
- } finally {
- if (bind) {
- BranchType previousBranchType = RootContext.getBranchType();
- String unbindXid = RootContext.unbind();
- if (BranchType.TCC == previousBranchType) {
- RootContext.unbindBranchType();
- }
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("unbind xid [{}] branchType [{}] from
RootContext", unbindXid, previousBranchType);
- }
- if (!requestXid.equalsIgnoreCase(unbindXid)) {
- LOGGER.warn("xid has changed, during RPC from [{}] to
[{}]", requestXid, unbindXid);
- if (unbindXid != null) {
- RootContext.bind(unbindXid);
- LOGGER.warn("bind [{}}] back to RootContext",
unbindXid);
- if (BranchType.TCC == previousBranchType) {
- RootContext.bindBranchType(BranchType.TCC);
- LOGGER.warn("bind branchType [{}] back to
RootContext", previousBranchType);
- }
- }
- }
- }
- }
- }
-
- /**
- * get rpc xid
- * @param request
- * @return
- */
- private String getRpcXid(Request request) {
- String rpcXid = request.getAttachments().get(RootContext.KEY_XID);
- if (rpcXid == null) {
- rpcXid =
request.getAttachments().get(RootContext.KEY_XID.toLowerCase());
- }
- return rpcXid;
- }
-
- private String getBranchType(Request request) {
- return request.getAttachments().get(RootContext.KEY_BRANCH_TYPE);
- }
-}
diff --git
a/extensions/rpc/seata-motan/src/main/java/org/apache/seata/integration/motan/MotanTransactionProviderFilter.java
b/extensions/rpc/seata-motan/src/main/java/org/apache/seata/integration/motan/MotanTransactionProviderFilter.java
new file mode 100644
index 0000000000..ac9e46045d
--- /dev/null
+++
b/extensions/rpc/seata-motan/src/main/java/org/apache/seata/integration/motan/MotanTransactionProviderFilter.java
@@ -0,0 +1,57 @@
+/*
+ * 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.motan;
+
+import com.weibo.api.motan.common.MotanConstants;
+import com.weibo.api.motan.core.extension.Activation;
+import com.weibo.api.motan.core.extension.Scope;
+import com.weibo.api.motan.core.extension.Spi;
+import com.weibo.api.motan.filter.Filter;
+import com.weibo.api.motan.rpc.Caller;
+import com.weibo.api.motan.rpc.Request;
+import com.weibo.api.motan.rpc.Response;
+import org.apache.seata.core.context.RootContext;
+import org.apache.seata.integration.rpc.core.TransactionPropagationHandler;
+
+@Spi(scope = Scope.SINGLETON)
+@Activation(
+ key = {MotanConstants.NODE_TYPE_SERVICE},
+ sequence = 100)
+public class MotanTransactionProviderFilter implements Filter {
+
+ @Override
+ public Response filter(final Caller<?> caller, final Request request) {
+ String rpcXid = getRpcXid(request);
+ String rpcBranchType =
request.getAttachments().get(RootContext.KEY_BRANCH_TYPE);
+ boolean bound =
TransactionPropagationHandler.bindProviderContext(rpcXid, rpcBranchType);
+ try {
+ return caller.call(request);
+ } finally {
+ if (bound) {
+ TransactionPropagationHandler.unbindProviderContext(rpcXid);
+ }
+ }
+ }
+
+ private String getRpcXid(Request request) {
+ String rpcXid = request.getAttachments().get(RootContext.KEY_XID);
+ if (rpcXid == null) {
+ rpcXid =
request.getAttachments().get(RootContext.KEY_XID.toLowerCase());
+ }
+ return rpcXid;
+ }
+}
diff --git
a/extensions/rpc/seata-motan/src/main/resources/META-INF/services/com.weibo.api.motan.filter.Filter
b/extensions/rpc/seata-motan/src/main/resources/META-INF/services/com.weibo.api.motan.filter.Filter
index 14920bb6fa..f94ea2b3a8 100644
---
a/extensions/rpc/seata-motan/src/main/resources/META-INF/services/com.weibo.api.motan.filter.Filter
+++
b/extensions/rpc/seata-motan/src/main/resources/META-INF/services/com.weibo.api.motan.filter.Filter
@@ -14,4 +14,5 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-org.apache.seata.integration.motan.MotanTransactionFilter
\ No newline at end of file
+org.apache.seata.integration.motan.MotanTransactionConsumerFilter
+org.apache.seata.integration.motan.MotanTransactionProviderFilter
\ No newline at end of file
diff --git
a/extensions/rpc/seata-rpc-core/src/main/java/org/apache/seata/integration/rpc/core/BaseRpcFilter.java
b/extensions/rpc/seata-rpc-core/src/main/java/org/apache/seata/integration/rpc/core/BaseRpcFilter.java
deleted file mode 100644
index 71cc7b423e..0000000000
---
a/extensions/rpc/seata-rpc-core/src/main/java/org/apache/seata/integration/rpc/core/BaseRpcFilter.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * 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.rpc.core;
-
-import org.apache.seata.common.util.CollectionUtils;
-import org.apache.seata.common.util.StringUtils;
-import org.apache.seata.core.context.RootContext;
-
-import java.util.Map;
-
-public interface BaseRpcFilter<T> {
- String[] TRX_CONTEXT_KEYS = new String[] {RootContext.KEY_XID,
RootContext.KEY_BRANCH_TYPE};
-
- default String getValueFromMap(Map<String, String> rpcContextMap, String
key) {
- return rpcContextMap.get(key);
- }
-
- default void assertNotNull(Object obj, String errMsg) {
- if (obj == null) {
- throw new IllegalStateException(errMsg);
- }
- }
-
- default String getJsonContext(Map<String, String> contextMap) {
- if (CollectionUtils.isEmpty(contextMap)) {
- return StringUtils.EMPTY;
- }
- StringBuilder sb = new StringBuilder("{");
- for (int i = 0; i < TRX_CONTEXT_KEYS.length; i++) {
- String contextValue = contextMap.get(TRX_CONTEXT_KEYS[i]);
- if (i > 0) {
- sb.append(",");
- }
- sb.append("\"").append(TRX_CONTEXT_KEYS[i]).append("\"");
- sb.append(":");
- if (null == contextValue) {
- sb.append("null");
- } else {
- sb.append("\"").append(contextValue).append("\"");
- }
- }
- sb.append("}");
- return sb.toString();
- }
-}
diff --git
a/extensions/rpc/seata-rpc-core/src/main/java/org/apache/seata/integration/rpc/core/ConsumerRpcFilter.java
b/extensions/rpc/seata-rpc-core/src/main/java/org/apache/seata/integration/rpc/core/ConsumerRpcFilter.java
deleted file mode 100644
index f5bc73a371..0000000000
---
a/extensions/rpc/seata-rpc-core/src/main/java/org/apache/seata/integration/rpc/core/ConsumerRpcFilter.java
+++ /dev/null
@@ -1,91 +0,0 @@
-/*
- * 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.rpc.core;
-
-import org.apache.seata.common.util.StringUtils;
-import org.apache.seata.core.context.RootContext;
-
-import java.util.HashMap;
-import java.util.Map;
-
-public interface ConsumerRpcFilter<T> extends BaseRpcFilter<T> {
-
- /**
- * get contexts from RootContext
- *
- * @return
- */
- default Map<String, String> getRootContexts() {
- Map<String, String> contextMap = new HashMap<>();
- if (RootContext.inGlobalTransaction()) {
- for (int i = 0; i < TRX_CONTEXT_KEYS.length; i++) {
- switch (TRX_CONTEXT_KEYS[i]) {
- case RootContext.KEY_XID:
- assertNotNull(RootContext.getXID(), "xid is null");
- contextMap.put(RootContext.KEY_XID,
RootContext.getXID());
- break;
- case RootContext.KEY_BRANCH_TYPE:
- contextMap.put(
- RootContext.KEY_BRANCH_TYPE,
- RootContext.getBranchType().name());
- break;
- default:
- throw new IllegalArgumentException("wrong context: " +
TRX_CONTEXT_KEYS[i]);
- }
- }
- }
- return contextMap;
- }
-
- default String getXidFromRootContexts(Map<String, String> rootContextMap) {
- return getValueFromMap(rootContextMap, RootContext.KEY_XID);
- }
-
- /**
- * bind contexts to RpcRequest
- *
- * @param rpcRequest
- * @param contextMap
- */
- default void bindContextsToRequest(T rpcRequest, Map<String, String>
contextMap) {
- for (int i = 0; i < TRX_CONTEXT_KEYS.length; i++) {
- String contextValue = contextMap.get(TRX_CONTEXT_KEYS[i]);
- if (StringUtils.isNotBlank(contextValue)) {
- bindContextToRequest(rpcRequest, TRX_CONTEXT_KEYS[i],
contextValue);
- }
- }
- }
-
- void bindContextToRequest(T rpcRequest, String key, String value);
-
- /**
- * clean contexts to RpcRequest
- *
- * @param rpcRequest
- * @param contextMap
- */
- default void cleanRequestContexts(T rpcRequest, Map<String, String>
contextMap) {
- for (int i = 0; i < TRX_CONTEXT_KEYS.length; i++) {
- String contextValue = contextMap.get(TRX_CONTEXT_KEYS[i]);
- if (StringUtils.isNotBlank(contextValue)) {
- cleanRequestContext(rpcRequest, TRX_CONTEXT_KEYS[i]);
- }
- }
- }
-
- void cleanRequestContext(T rpcRequest, String key);
-}
diff --git
a/extensions/rpc/seata-rpc-core/src/main/java/org/apache/seata/integration/rpc/core/ProviderRpcFilter.java
b/extensions/rpc/seata-rpc-core/src/main/java/org/apache/seata/integration/rpc/core/ProviderRpcFilter.java
deleted file mode 100644
index 0e297c6038..0000000000
---
a/extensions/rpc/seata-rpc-core/src/main/java/org/apache/seata/integration/rpc/core/ProviderRpcFilter.java
+++ /dev/null
@@ -1,110 +0,0 @@
-/*
- * 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.rpc.core;
-
-import org.apache.seata.common.util.StringUtils;
-import org.apache.seata.core.context.RootContext;
-import org.apache.seata.core.model.BranchType;
-
-import java.util.HashMap;
-import java.util.Map;
-
-public interface ProviderRpcFilter<T> extends BaseRpcFilter<T> {
-
- String[] TRX_CONTEXT_KEYS =
- new String[] {RootContext.KEY_XID,
RootContext.KEY_XID.toLowerCase(), RootContext.KEY_BRANCH_TYPE};
-
- String LOW_KEY_XID = "tx_xid";
-
- /**
- * get contexts from RpcRequest
- *
- * @param rpcRequest
- * @return
- */
- default Map<String, String> getRpcContexts(T rpcRequest) {
- Map<String, String> contextMap = new HashMap<>();
- for (int i = 0; i < TRX_CONTEXT_KEYS.length; i++) {
- String contextValue = getRpcContext(rpcRequest,
TRX_CONTEXT_KEYS[i]);
- if (StringUtils.isNotBlank(contextValue)) {
- contextMap.put(TRX_CONTEXT_KEYS[i], contextValue);
- }
- }
- return contextMap;
- }
-
- String getRpcContext(T rpcContext, String key);
-
- default String getXidFromContexts(Map<String, String> rpcContextMap) {
- String xid = getValueFromMap(rpcContextMap, RootContext.KEY_XID);
- if (StringUtils.isBlank(xid)) {
- return getValueFromMap(rpcContextMap,
RootContext.KEY_XID.toLowerCase());
- }
- return xid;
- }
-
- default void bindRequestToContexts(Map<String, String> contextMap) {
- for (int i = 0; i < TRX_CONTEXT_KEYS.length; i++) {
- String contextValue = contextMap.get(TRX_CONTEXT_KEYS[i]);
- if (StringUtils.isNotBlank(contextValue)) {
- switch (TRX_CONTEXT_KEYS[i]) {
- case RootContext.KEY_XID:
- case LOW_KEY_XID:
- RootContext.bind(contextValue);
- break;
- case RootContext.KEY_BRANCH_TYPE:
- if
(BranchType.TCC.name().equalsIgnoreCase(contextValue)) {
- RootContext.bindBranchType(BranchType.TCC);
- }
- break;
- default:
- throw new IllegalArgumentException("wrong context:" +
TRX_CONTEXT_KEYS[i]);
- }
- }
- }
- }
-
- default Map<String, String> cleanRootContexts() {
- Map<String, String> contextMap = new HashMap<>();
- for (int i = 0; i < TRX_CONTEXT_KEYS.length; i++) {
- switch (TRX_CONTEXT_KEYS[i]) {
- case RootContext.KEY_XID:
- String xid = RootContext.unbind();
- contextMap.put(RootContext.KEY_XID, xid);
- break;
- case LOW_KEY_XID:
- break;
- case RootContext.KEY_BRANCH_TYPE:
- BranchType contextValue = RootContext.getBranchType();
- if (BranchType.TCC == contextValue) {
- RootContext.unbindBranchType();
- }
- if (null != contextValue) {
- contextMap.put(RootContext.KEY_BRANCH_TYPE,
contextValue.name());
- }
- break;
- default:
- throw new IllegalArgumentException("wrong context:" +
TRX_CONTEXT_KEYS[i]);
- }
- }
- return contextMap;
- }
-
- default void resetRootContexts(Map<String, String> contextMap) {
- bindRequestToContexts(contextMap);
- }
-}
diff --git
a/extensions/rpc/seata-rpc-core/src/main/java/org/apache/seata/integration/rpc/core/TransactionPropagationHandler.java
b/extensions/rpc/seata-rpc-core/src/main/java/org/apache/seata/integration/rpc/core/TransactionPropagationHandler.java
new file mode 100644
index 0000000000..fa0ba68bc9
--- /dev/null
+++
b/extensions/rpc/seata-rpc-core/src/main/java/org/apache/seata/integration/rpc/core/TransactionPropagationHandler.java
@@ -0,0 +1,77 @@
+/*
+ * 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.rpc.core;
+
+import org.apache.seata.core.context.RootContext;
+import org.apache.seata.core.model.BranchType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public final class TransactionPropagationHandler {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(TransactionPropagationHandler.class);
+
+ private TransactionPropagationHandler() {}
+
+ public static Map<String, String> getTransactionPropagationContext() {
+ Map<String, String> context = new HashMap<>();
+ String xid = RootContext.getXID();
+ if (xid != null) {
+ context.put(RootContext.KEY_XID, xid);
+ BranchType branchType = RootContext.getBranchType();
+ if (branchType != null) {
+ context.put(RootContext.KEY_BRANCH_TYPE, branchType.name());
+ }
+ }
+ return context;
+ }
+
+ public static boolean bindProviderContext(String rpcXid, String
rpcBranchType) {
+ if (rpcXid == null) {
+ return false;
+ }
+ RootContext.bind(rpcXid);
+ if (BranchType.TCC.name().equalsIgnoreCase(rpcBranchType)) {
+ RootContext.bindBranchType(BranchType.TCC);
+ }
+ return true;
+ }
+
+ public static void unbindProviderContext(String rpcXid) {
+ BranchType previousBranchType = RootContext.getBranchType();
+ String unbindXid = RootContext.unbind();
+ if (BranchType.TCC == previousBranchType) {
+ RootContext.unbindBranchType();
+ }
+ if (rpcXid != null && !rpcXid.equalsIgnoreCase(unbindXid)) {
+ LOGGER.warn("xid changed during RPC from {} to {}", rpcXid,
unbindXid);
+ if (unbindXid != null) {
+ RootContext.bind(unbindXid);
+ if (BranchType.TCC == previousBranchType) {
+ RootContext.bindBranchType(BranchType.TCC);
+ }
+ }
+ }
+ }
+
+ public static String resolveXid(String xid, String xidLowercase) {
+ return xid != null ? xid : xidLowercase;
+ }
+}
diff --git
a/extensions/rpc/seata-rpc-core/src/test/java/org/apache/seata/integration/rpc/core/TransactionPropagationHandlerTest.java
b/extensions/rpc/seata-rpc-core/src/test/java/org/apache/seata/integration/rpc/core/TransactionPropagationHandlerTest.java
new file mode 100644
index 0000000000..5470e8bd7e
--- /dev/null
+++
b/extensions/rpc/seata-rpc-core/src/test/java/org/apache/seata/integration/rpc/core/TransactionPropagationHandlerTest.java
@@ -0,0 +1,140 @@
+/*
+ * 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.rpc.core;
+
+import org.apache.seata.core.context.RootContext;
+import org.apache.seata.core.model.BranchType;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class TransactionPropagationHandlerTest {
+
+ private static final String DEFAULT_XID = "127.0.0.1:8091:12345678";
+
+ @AfterEach
+ void cleanup() {
+ RootContext.unbind();
+ RootContext.unbindBranchType();
+ }
+
+ @Test
+ void testGetTransactionPropagationContext_noTransaction() {
+ Map<String, String> context =
TransactionPropagationHandler.getTransactionPropagationContext();
+ assertThat(context).isEmpty();
+ }
+
+ @Test
+ void testGetTransactionPropagationContext_withXid() {
+ RootContext.bind(DEFAULT_XID);
+ Map<String, String> context =
TransactionPropagationHandler.getTransactionPropagationContext();
+ assertThat(context).containsEntry(RootContext.KEY_XID, DEFAULT_XID);
+ assertThat(context).containsEntry(RootContext.KEY_BRANCH_TYPE,
BranchType.AT.name());
+ }
+
+ @Test
+ void testGetTransactionPropagationContext_withXidAndBranchType() {
+ RootContext.bind(DEFAULT_XID);
+ RootContext.bindBranchType(BranchType.TCC);
+ Map<String, String> context =
TransactionPropagationHandler.getTransactionPropagationContext();
+ assertThat(context).containsEntry(RootContext.KEY_XID, DEFAULT_XID);
+ assertThat(context).containsEntry(RootContext.KEY_BRANCH_TYPE,
BranchType.TCC.name());
+ }
+
+ @Test
+ void testBindProviderContext_nullXid() {
+ boolean bound =
TransactionPropagationHandler.bindProviderContext(null, null);
+ assertThat(bound).isFalse();
+ assertThat(RootContext.getXID()).isNull();
+ }
+
+ @Test
+ void testBindProviderContext_withXid() {
+ boolean bound =
TransactionPropagationHandler.bindProviderContext(DEFAULT_XID, null);
+ assertThat(bound).isTrue();
+ assertThat(RootContext.getXID()).isEqualTo(DEFAULT_XID);
+ assertThat(RootContext.getBranchType()).isEqualTo(BranchType.AT);
+ }
+
+ @Test
+ void testBindProviderContext_withTCC() {
+ boolean bound =
TransactionPropagationHandler.bindProviderContext(DEFAULT_XID,
BranchType.TCC.name());
+ assertThat(bound).isTrue();
+ assertThat(RootContext.getXID()).isEqualTo(DEFAULT_XID);
+ assertThat(RootContext.getBranchType()).isEqualTo(BranchType.TCC);
+ }
+
+ @Test
+ void testBindProviderContext_withNonTCCBranchType() {
+ boolean bound =
TransactionPropagationHandler.bindProviderContext(DEFAULT_XID,
BranchType.AT.name());
+ assertThat(bound).isTrue();
+ assertThat(RootContext.getXID()).isEqualTo(DEFAULT_XID);
+ assertThat(RootContext.getBranchType()).isEqualTo(BranchType.AT);
+ }
+
+ @Test
+ void testUnbindProviderContext_normalFlow() {
+ RootContext.bind(DEFAULT_XID);
+ TransactionPropagationHandler.unbindProviderContext(DEFAULT_XID);
+ assertThat(RootContext.getXID()).isNull();
+ }
+
+ @Test
+ void testUnbindProviderContext_withTCCBranchType() {
+ RootContext.bind(DEFAULT_XID);
+ RootContext.bindBranchType(BranchType.TCC);
+ TransactionPropagationHandler.unbindProviderContext(DEFAULT_XID);
+ assertThat(RootContext.getXID()).isNull();
+ assertThat(RootContext.getBranchType()).isNull();
+ }
+
+ @Test
+ void testUnbindProviderContext_xidChanged() {
+ String changedXid = "127.0.0.1:8091:99999999";
+ RootContext.bind(changedXid);
+ TransactionPropagationHandler.unbindProviderContext(DEFAULT_XID);
+ assertThat(RootContext.getXID()).isEqualTo(changedXid);
+ }
+
+ @Test
+ void testUnbindProviderContext_xidChangedWithTCC() {
+ String changedXid = "127.0.0.1:8091:99999999";
+ RootContext.bind(changedXid);
+ RootContext.bindBranchType(BranchType.TCC);
+ TransactionPropagationHandler.unbindProviderContext(DEFAULT_XID);
+ assertThat(RootContext.getXID()).isEqualTo(changedXid);
+ assertThat(RootContext.getBranchType()).isEqualTo(BranchType.TCC);
+ }
+
+ @Test
+ void testResolveXid_xidNotNull() {
+ assertThat(TransactionPropagationHandler.resolveXid("xid1",
"xid2")).isEqualTo("xid1");
+ }
+
+ @Test
+ void testResolveXid_xidNull() {
+ assertThat(TransactionPropagationHandler.resolveXid(null,
"xid2")).isEqualTo("xid2");
+ }
+
+ @Test
+ void testResolveXid_bothNull() {
+ assertThat(TransactionPropagationHandler.resolveXid(null,
null)).isNull();
+ }
+}
diff --git a/extensions/rpc/seata-sofa-rpc/pom.xml
b/extensions/rpc/seata-sofa-rpc/pom.xml
index be1e9b225c..338e9a8474 100644
--- a/extensions/rpc/seata-sofa-rpc/pom.xml
+++ b/extensions/rpc/seata-sofa-rpc/pom.xml
@@ -35,7 +35,7 @@
<dependencies>
<dependency>
<groupId>${project.groupId}</groupId>
- <artifactId>seata-tm</artifactId>
+ <artifactId>seata-rpc-core</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
diff --git
a/extensions/rpc/seata-sofa-rpc/src/main/java/org/apache/seata/integration/sofa/rpc/TransactionContextConsumerFilter.java
b/extensions/rpc/seata-sofa-rpc/src/main/java/org/apache/seata/integration/sofa/rpc/TransactionContextConsumerFilter.java
index 95132320af..af650d0095 100644
---
a/extensions/rpc/seata-sofa-rpc/src/main/java/org/apache/seata/integration/sofa/rpc/TransactionContextConsumerFilter.java
+++
b/extensions/rpc/seata-sofa-rpc/src/main/java/org/apache/seata/integration/sofa/rpc/TransactionContextConsumerFilter.java
@@ -16,7 +16,6 @@
*/
package org.apache.seata.integration.sofa.rpc;
-import com.alipay.sofa.rpc.context.RpcInternalContext;
import com.alipay.sofa.rpc.core.exception.SofaRpcException;
import com.alipay.sofa.rpc.core.request.SofaRequest;
import com.alipay.sofa.rpc.core.response.SofaResponse;
@@ -24,100 +23,28 @@ import com.alipay.sofa.rpc.ext.Extension;
import com.alipay.sofa.rpc.filter.AutoActive;
import com.alipay.sofa.rpc.filter.Filter;
import com.alipay.sofa.rpc.filter.FilterInvoker;
-import org.apache.seata.common.util.StringUtils;
-import org.apache.seata.core.context.RootContext;
-import org.apache.seata.core.model.BranchType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.seata.integration.rpc.core.TransactionPropagationHandler;
+
+import java.util.Map;
-/**
- * TransactionContext on consumer side.
- *
- * @since 0.6.0
- */
@Extension(value = "transactionContextConsumer")
@AutoActive(consumerSide = true)
public class TransactionContextConsumerFilter extends Filter {
- /**
- * Logger for this class
- */
- private static final Logger LOGGER =
LoggerFactory.getLogger(TransactionContextConsumerFilter.class);
-
@Override
public SofaResponse invoke(FilterInvoker filterInvoker, SofaRequest
sofaRequest) throws SofaRpcException {
- String xid = RootContext.getXID();
- String rpcXid = getRpcXid();
- BranchType branchType = RootContext.getBranchType();
- String rpcBranchType = getBranchType();
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug(
- "context in RootContext[{},{}], context in
RpcContext[{},{}]",
- xid,
- branchType,
- rpcXid,
- rpcBranchType);
- }
- boolean bind = false;
- if (xid != null) {
- sofaRequest.addRequestProp(RootContext.KEY_XID, xid);
- sofaRequest.addRequestProp(RootContext.KEY_BRANCH_TYPE,
branchType.name());
- } else {
- if (rpcXid != null) {
- RootContext.bind(rpcXid);
- if (StringUtils.equals(BranchType.TCC.name(), rpcBranchType)) {
- RootContext.bindBranchType(BranchType.TCC);
- }
- bind = true;
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("bind[{}] to RootContext", rpcXid);
- }
+ Map<String, String> context =
TransactionPropagationHandler.getTransactionPropagationContext();
+ if (!context.isEmpty()) {
+ for (Map.Entry<String, String> entry : context.entrySet()) {
+ sofaRequest.addRequestProp(entry.getKey(), entry.getValue());
}
}
try {
return filterInvoker.invoke(sofaRequest);
} finally {
- if (bind) {
- String unbindXid = RootContext.unbind();
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("unbind[{}] from RootContext", unbindXid);
- }
- BranchType previousBranchType = RootContext.getBranchType();
- if (BranchType.TCC == previousBranchType) {
- RootContext.unbindBranchType();
- }
- if (!rpcXid.equalsIgnoreCase(unbindXid)) {
- if (LOGGER.isWarnEnabled()) {
- LOGGER.warn("xid in change during RPC from [{}] to
[{}]", rpcXid, unbindXid);
- }
- if (unbindXid != null) {
- RootContext.bind(unbindXid);
- if (LOGGER.isWarnEnabled()) {
- LOGGER.warn("bind [{}] back to RootContext",
unbindXid);
- }
- if (BranchType.TCC == previousBranchType) {
- RootContext.bindBranchType(BranchType.TCC);
- LOGGER.warn("bind branchType [{}] back to
RootContext", previousBranchType);
- }
- }
- }
+ for (String key : context.keySet()) {
+ sofaRequest.removeRequestProp(key);
}
}
}
-
- /**
- * get rpc xid
- * @return
- */
- private String getRpcXid() {
- String rpcXid = (String)
RpcInternalContext.getContext().getAttachment(RootContext.HIDDEN_KEY_XID);
- if (rpcXid == null) {
- rpcXid = (String)
RpcInternalContext.getContext().getAttachment(RootContext.HIDDEN_KEY_XID.toLowerCase());
- }
- return rpcXid;
- }
-
- private String getBranchType() {
- return (String)
RpcInternalContext.getContext().getAttachment(RootContext.HIDDEN_KEY_BRANCH_TYPE);
- }
}
diff --git
a/extensions/rpc/seata-sofa-rpc/src/main/java/org/apache/seata/integration/sofa/rpc/TransactionContextProviderFilter.java
b/extensions/rpc/seata-sofa-rpc/src/main/java/org/apache/seata/integration/sofa/rpc/TransactionContextProviderFilter.java
index e2d0579f9e..9fa2b9a50d 100644
---
a/extensions/rpc/seata-sofa-rpc/src/main/java/org/apache/seata/integration/sofa/rpc/TransactionContextProviderFilter.java
+++
b/extensions/rpc/seata-sofa-rpc/src/main/java/org/apache/seata/integration/sofa/rpc/TransactionContextProviderFilter.java
@@ -16,7 +16,6 @@
*/
package org.apache.seata.integration.sofa.rpc;
-import com.alipay.sofa.rpc.context.RpcInternalContext;
import com.alipay.sofa.rpc.core.exception.SofaRpcException;
import com.alipay.sofa.rpc.core.request.SofaRequest;
import com.alipay.sofa.rpc.core.response.SofaResponse;
@@ -24,95 +23,27 @@ import com.alipay.sofa.rpc.ext.Extension;
import com.alipay.sofa.rpc.filter.AutoActive;
import com.alipay.sofa.rpc.filter.Filter;
import com.alipay.sofa.rpc.filter.FilterInvoker;
-import org.apache.seata.common.util.StringUtils;
import org.apache.seata.core.context.RootContext;
-import org.apache.seata.core.model.BranchType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.seata.integration.rpc.core.TransactionPropagationHandler;
-/**
- * TransactionContext on provider side.
- *
- * @since 0.6.0
- */
@Extension(value = "transactionContextProvider")
@AutoActive(providerSide = true)
public class TransactionContextProviderFilter extends Filter {
- /**
- * Logger for this class
- */
- private static final Logger LOGGER =
LoggerFactory.getLogger(TransactionContextProviderFilter.class);
-
@Override
public SofaResponse invoke(FilterInvoker filterInvoker, SofaRequest
sofaRequest) throws SofaRpcException {
- String xid = RootContext.getXID();
String rpcXid = getRpcXid(sofaRequest);
- BranchType branchType = RootContext.getBranchType();
- String rpcBranchType = getBranchType(sofaRequest);
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug(
- "context in RootContext[{},{}], context in
RpcContext[{},{}]",
- xid,
- branchType,
- rpcXid,
- rpcBranchType);
- }
- boolean bind = false;
- if (xid != null) {
-
RpcInternalContext.getContext().setAttachment(RootContext.HIDDEN_KEY_XID, xid);
-
RpcInternalContext.getContext().setAttachment(RootContext.HIDDEN_KEY_BRANCH_TYPE,
branchType.name());
- } else {
- if (null != rpcXid) {
- RootContext.bind(rpcXid);
- if (StringUtils.equals(BranchType.TCC.name(), rpcBranchType)) {
- RootContext.bindBranchType(BranchType.TCC);
- }
- bind = true;
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("bind[{}] to RootContext", rpcXid);
- }
- }
- }
+ String rpcBranchType = (String)
sofaRequest.getRequestProp(RootContext.KEY_BRANCH_TYPE);
+ boolean bound =
TransactionPropagationHandler.bindProviderContext(rpcXid, rpcBranchType);
try {
return filterInvoker.invoke(sofaRequest);
} finally {
- if (xid != null) {
-
RpcInternalContext.getContext().removeAttachment(RootContext.HIDDEN_KEY_XID);
-
RpcInternalContext.getContext().removeAttachment(RootContext.HIDDEN_KEY_BRANCH_TYPE);
- }
- if (bind) {
- String unbindXid = RootContext.unbind();
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("unbind[{}] from RootContext", unbindXid);
- }
- BranchType previousBranchType = RootContext.getBranchType();
- if (BranchType.TCC == previousBranchType) {
- RootContext.unbindBranchType();
- }
- if (!rpcXid.equalsIgnoreCase(unbindXid)) {
- if (LOGGER.isWarnEnabled()) {
- LOGGER.warn("xid in change during RPC from [{}] to
[{}]", rpcXid, unbindXid);
- }
- if (unbindXid != null) {
- RootContext.bind(unbindXid);
- if (LOGGER.isWarnEnabled()) {
- LOGGER.warn("bind [{}] back to RootContext",
unbindXid);
- }
- if (BranchType.TCC == previousBranchType) {
- RootContext.bindBranchType(BranchType.TCC);
- LOGGER.warn("bind branchType [{}] back to
RootContext", previousBranchType);
- }
- }
- }
+ if (bound) {
+ TransactionPropagationHandler.unbindProviderContext(rpcXid);
}
}
}
- /**
- * get rpc xid
- * @return
- */
private String getRpcXid(SofaRequest sofaRequest) {
String rpcXid = (String)
sofaRequest.getRequestProp(RootContext.KEY_XID);
if (rpcXid == null) {
@@ -120,8 +51,4 @@ public class TransactionContextProviderFilter extends Filter
{
}
return rpcXid;
}
-
- private String getBranchType(SofaRequest sofaRequest) {
- return (String)
sofaRequest.getRequestProp(RootContext.KEY_BRANCH_TYPE);
- }
}
diff --git
a/rm-datasource/src/test/java/org/apache/seata/rm/datasource/ConnectionProxyTest.java
b/rm-datasource/src/test/java/org/apache/seata/rm/datasource/ConnectionProxyTest.java
index 6f82a14c32..323ac00403 100644
---
a/rm-datasource/src/test/java/org/apache/seata/rm/datasource/ConnectionProxyTest.java
+++
b/rm-datasource/src/test/java/org/apache/seata/rm/datasource/ConnectionProxyTest.java
@@ -23,6 +23,7 @@ import
org.apache.seata.core.exception.TransactionExceptionCode;
import org.apache.seata.core.model.BranchStatus;
import org.apache.seata.core.model.BranchType;
import org.apache.seata.core.model.GlobalLockConfig;
+import org.apache.seata.core.model.ResourceManager;
import org.apache.seata.rm.DefaultResourceManager;
import org.apache.seata.rm.datasource.ConnectionProxy.LockRetryPolicy;
import org.apache.seata.rm.datasource.exec.LockConflictException;
@@ -57,12 +58,13 @@ public class ConnectionProxyTest {
private static final String TEST_XID = "testXid";
- private static final String lockKey = "order:123";
+ private static final String LOCK_KEY = "order:123";
private static final String DB_TYPE = "mysql";
private Field branchRollbackFlagField;
private boolean originalBranchRollbackFlag;
+ private ResourceManager originalAtResourceManager;
@BeforeEach
public void initBeforeEach() throws Exception {
@@ -85,16 +87,23 @@ public class ConnectionProxyTest {
null,
TEST_XID,
"{\"autoCommit\":false}",
- lockKey))
+ LOCK_KEY))
.thenThrow(new
TransactionException(TransactionExceptionCode.LockKeyConflict));
DefaultResourceManager defaultResourceManager =
DefaultResourceManager.get();
Assertions.assertNotNull(defaultResourceManager);
+ originalAtResourceManager =
defaultResourceManager.getResourceManager(BranchType.AT);
DefaultResourceManager.mockResourceManager(BranchType.AT, rm);
}
@org.junit.jupiter.api.AfterEach
public void cleanupAfterEach() throws Exception {
- branchRollbackFlagField.set(null, originalBranchRollbackFlag);
+ try {
+ branchRollbackFlagField.set(null, originalBranchRollbackFlag);
+ } finally {
+ if (originalAtResourceManager != null) {
+ DefaultResourceManager.mockResourceManager(BranchType.AT,
originalAtResourceManager);
+ }
+ }
}
@Test
@@ -114,7 +123,7 @@ public class ConnectionProxyTest {
sqlUndoLog.setBeforeImage(beforeImage);
connectionProxy.getContext().appendUndoItem(sqlUndoLog);
connectionProxy.appendUndoLog(new SQLUndoLog());
- connectionProxy.appendLockKey(lockKey);
+ connectionProxy.appendLockKey(LOCK_KEY);
Assertions.assertThrows(LockWaitTimeoutException.class,
connectionProxy::commit);
}
}
@@ -130,7 +139,7 @@ public class ConnectionProxyTest {
ConnectionProxy connectionProxy = new ConnectionProxy(dataSourceProxy,
null);
connectionProxy.bind(TEST_XID);
connectionProxy.appendUndoLog(new SQLUndoLog());
- connectionProxy.appendLockKey(lockKey);
+ connectionProxy.appendLockKey(LOCK_KEY);
SQLUndoLog sqlUndoLog = new SQLUndoLog();
TableRecords beforeImage = new TableRecords();
beforeImage.add(new Row());
@@ -246,7 +255,7 @@ public class ConnectionProxyTest {
beforeImage.add(new Row());
sqlUndoLog.setBeforeImage(beforeImage);
connectionProxy.appendUndoLog(sqlUndoLog);
- connectionProxy.appendLockKey(lockKey);
+ connectionProxy.appendLockKey(LOCK_KEY);
connectionProxy.commit();
@@ -257,7 +266,7 @@ public class ConnectionProxyTest {
Mockito.isNull(),
Mockito.eq(TEST_XID),
Mockito.anyString(),
- Mockito.eq(lockKey));
+ Mockito.eq(LOCK_KEY));
}
}
@@ -273,12 +282,12 @@ public class ConnectionProxyTest {
connectionProxy.setAutoCommit(false);
connectionProxy.setGlobalLockRequire(true);
- connectionProxy.appendLockKey(lockKey);
+ connectionProxy.appendLockKey(LOCK_KEY);
connectionProxy.commit();
Mockito.verify(rm)
- .lockQuery(Mockito.eq(BranchType.AT), Mockito.anyString(),
Mockito.isNull(), Mockito.eq(lockKey));
+ .lockQuery(Mockito.eq(BranchType.AT), Mockito.anyString(),
Mockito.isNull(), Mockito.eq(LOCK_KEY));
}
}
@@ -343,7 +352,7 @@ public class ConnectionProxyTest {
beforeImage.add(new Row());
sqlUndoLog.setBeforeImage(beforeImage);
connectionProxy.appendUndoLog(sqlUndoLog);
- connectionProxy.appendLockKey(lockKey);
+ connectionProxy.appendLockKey(LOCK_KEY);
connectionProxy.commit();
@@ -372,7 +381,7 @@ public class ConnectionProxyTest {
Mockito.isNull(),
Mockito.eq(TEST_XID),
Mockito.anyString(),
- Mockito.eq(lockKey)))
+ Mockito.eq(LOCK_KEY)))
.thenReturn(789L);
DefaultResourceManager.mockResourceManager(BranchType.AT, rm);
@@ -383,7 +392,7 @@ public class ConnectionProxyTest {
beforeImage.add(new Row());
sqlUndoLog.setBeforeImage(beforeImage);
connectionProxy.appendUndoLog(sqlUndoLog);
- connectionProxy.appendLockKey(lockKey);
+ connectionProxy.appendLockKey(LOCK_KEY);
connectionProxy.commit();
@@ -394,7 +403,7 @@ public class ConnectionProxyTest {
Mockito.isNull(),
Mockito.eq(TEST_XID),
Mockito.anyString(),
- Mockito.eq(lockKey));
+ Mockito.eq(LOCK_KEY));
}
}
@@ -407,7 +416,7 @@ public class ConnectionProxyTest {
connectionProxy.setAutoCommit(false);
connectionProxy.bind(TEST_XID);
- connectionProxy.appendLockKey(lockKey);
+ connectionProxy.appendLockKey(LOCK_KEY);
connectionProxy.commit();
@@ -467,7 +476,7 @@ public class ConnectionProxyTest {
beforeImage.add(new Row());
sqlUndoLog.setBeforeImage(beforeImage);
connectionProxy.appendUndoLog(sqlUndoLog);
- connectionProxy.appendLockKey(lockKey);
+ connectionProxy.appendLockKey(LOCK_KEY);
connectionProxy.commit();
@@ -509,7 +518,7 @@ public class ConnectionProxyTest {
beforeImage.add(new Row());
sqlUndoLog.setBeforeImage(beforeImage);
connectionProxy.appendUndoLog(sqlUndoLog);
- connectionProxy.appendLockKey(lockKey);
+ connectionProxy.appendLockKey(LOCK_KEY);
Assertions.assertThrows(SQLException.class, connectionProxy::commit);
@@ -553,7 +562,7 @@ public class ConnectionProxyTest {
beforeImage.add(new Row());
sqlUndoLog.setBeforeImage(beforeImage);
connectionProxy.appendUndoLog(sqlUndoLog);
- connectionProxy.appendLockKey(lockKey);
+ connectionProxy.appendLockKey(LOCK_KEY);
connectionProxy.commit();
@@ -591,7 +600,7 @@ public class ConnectionProxyTest {
beforeImage.add(new Row());
sqlUndoLog.setBeforeImage(beforeImage);
connectionProxy.appendUndoLog(sqlUndoLog);
- connectionProxy.appendLockKey(lockKey);
+ connectionProxy.appendLockKey(LOCK_KEY);
connectionProxy.setAutoCommit(true);
@@ -602,7 +611,7 @@ public class ConnectionProxyTest {
Mockito.isNull(),
Mockito.eq(TEST_XID),
Mockito.anyString(),
- Mockito.eq(lockKey));
+ Mockito.eq(LOCK_KEY));
Assertions.assertTrue(mockConnection.getAutoCommit());
}
}
@@ -619,12 +628,12 @@ public class ConnectionProxyTest {
connectionProxy.setAutoCommit(false);
connectionProxy.setGlobalLockRequire(true);
- connectionProxy.appendLockKey(lockKey);
+ connectionProxy.appendLockKey(LOCK_KEY);
connectionProxy.setAutoCommit(true);
Mockito.verify(rm)
- .lockQuery(Mockito.eq(BranchType.AT), Mockito.anyString(),
Mockito.isNull(), Mockito.eq(lockKey));
+ .lockQuery(Mockito.eq(BranchType.AT), Mockito.anyString(),
Mockito.isNull(), Mockito.eq(LOCK_KEY));
Assertions.assertTrue(mockConnection.getAutoCommit());
}
}
@@ -638,20 +647,20 @@ public class ConnectionProxyTest {
Mockito.eq(BranchType.AT),
Mockito.eq(TEST_RESOURCE_ID),
Mockito.eq(TEST_XID),
- Mockito.eq(lockKey)))
+ Mockito.eq(LOCK_KEY)))
.thenReturn(true);
DefaultResourceManager.mockResourceManager(BranchType.AT, rm);
connectionProxy.bind(TEST_XID);
- connectionProxy.checkLock(lockKey);
+ connectionProxy.checkLock(LOCK_KEY);
Mockito.verify(rm)
.lockQuery(
Mockito.eq(BranchType.AT),
Mockito.eq(TEST_RESOURCE_ID),
Mockito.eq(TEST_XID),
- Mockito.eq(lockKey));
+ Mockito.eq(LOCK_KEY));
}
}
@@ -664,13 +673,13 @@ public class ConnectionProxyTest {
Mockito.eq(BranchType.AT),
Mockito.eq(TEST_RESOURCE_ID),
Mockito.eq(TEST_XID),
- Mockito.eq(lockKey)))
+ Mockito.eq(LOCK_KEY)))
.thenReturn(false);
DefaultResourceManager.mockResourceManager(BranchType.AT, rm);
connectionProxy.bind(TEST_XID);
- Assertions.assertThrows(LockConflictException.class, () ->
connectionProxy.checkLock(lockKey));
+ Assertions.assertThrows(LockConflictException.class, () ->
connectionProxy.checkLock(LOCK_KEY));
}
}
@@ -683,13 +692,13 @@ public class ConnectionProxyTest {
Mockito.eq(BranchType.AT),
Mockito.eq(TEST_RESOURCE_ID),
Mockito.eq(TEST_XID),
- Mockito.eq(lockKey)))
+ Mockito.eq(LOCK_KEY)))
.thenReturn(true);
DefaultResourceManager.mockResourceManager(BranchType.AT, rm);
connectionProxy.bind(TEST_XID);
- boolean result = connectionProxy.lockQuery(lockKey);
+ boolean result = connectionProxy.lockQuery(LOCK_KEY);
Assertions.assertTrue(result);
}
@@ -704,13 +713,13 @@ public class ConnectionProxyTest {
Mockito.eq(BranchType.AT),
Mockito.eq(TEST_RESOURCE_ID),
Mockito.eq(TEST_XID),
- Mockito.eq(lockKey)))
+ Mockito.eq(LOCK_KEY)))
.thenReturn(false);
DefaultResourceManager.mockResourceManager(BranchType.AT, rm);
connectionProxy.bind(TEST_XID);
- boolean result = connectionProxy.lockQuery(lockKey);
+ boolean result = connectionProxy.lockQuery(LOCK_KEY);
Assertions.assertFalse(result);
}
@@ -772,14 +781,14 @@ public class ConnectionProxyTest {
Mockito.eq(BranchType.AT),
Mockito.eq(TEST_RESOURCE_ID),
Mockito.eq(TEST_XID),
- Mockito.eq(lockKey)))
+ Mockito.eq(LOCK_KEY)))
.thenThrow(new
TransactionException(TransactionExceptionCode.LockKeyConflict, "lock
conflict"));
DefaultResourceManager.mockResourceManager(BranchType.AT, rm);
connectionProxy.bind(TEST_XID);
LockConflictException exception =
- Assertions.assertThrows(LockConflictException.class, () ->
connectionProxy.checkLock(lockKey));
+ Assertions.assertThrows(LockConflictException.class, () ->
connectionProxy.checkLock(LOCK_KEY));
Assertions.assertEquals(TransactionExceptionCode.LockKeyConflict,
exception.getCode());
}
}
@@ -793,7 +802,7 @@ public class ConnectionProxyTest {
Mockito.eq(BranchType.AT),
Mockito.eq(TEST_RESOURCE_ID),
Mockito.eq(TEST_XID),
- Mockito.eq(lockKey)))
+ Mockito.eq(LOCK_KEY)))
.thenThrow(new TransactionException(
TransactionExceptionCode.LockKeyConflictFailFast,
"lock conflict fail fast"));
DefaultResourceManager.mockResourceManager(BranchType.AT, rm);
@@ -801,7 +810,7 @@ public class ConnectionProxyTest {
connectionProxy.bind(TEST_XID);
LockConflictException exception =
- Assertions.assertThrows(LockConflictException.class, () ->
connectionProxy.checkLock(lockKey));
+ Assertions.assertThrows(LockConflictException.class, () ->
connectionProxy.checkLock(LOCK_KEY));
Assertions.assertEquals(TransactionExceptionCode.LockKeyConflictFailFast,
exception.getCode());
}
}
diff --git
a/rm-datasource/src/test/java/org/apache/seata/rm/datasource/DataSourceProxyTest.java
b/rm-datasource/src/test/java/org/apache/seata/rm/datasource/DataSourceProxyTest.java
index ae0311c145..dd362aae73 100644
---
a/rm-datasource/src/test/java/org/apache/seata/rm/datasource/DataSourceProxyTest.java
+++
b/rm-datasource/src/test/java/org/apache/seata/rm/datasource/DataSourceProxyTest.java
@@ -17,6 +17,8 @@
package org.apache.seata.rm.datasource;
import com.alibaba.druid.pool.DruidDataSource;
+import org.apache.seata.core.model.BranchType;
+import org.apache.seata.core.model.ResourceManager;
import org.apache.seata.rm.DefaultResourceManager;
import org.apache.seata.rm.datasource.mock.MockDataSource;
import org.apache.seata.rm.datasource.mock.MockDriver;
@@ -256,13 +258,14 @@ public class DataSourceProxyTest {
DataSourceProxy proxy = getDataSourceProxy(dataSource);
+ ResourceManager atManager =
DefaultResourceManager.get().getResourceManager(BranchType.AT);
+
// Ensure it's registered
- Assertions.assertNotNull(
-
DefaultResourceManager.get().getManagedResources().get(proxy.getResourceId()));
+
Assertions.assertNotNull(atManager.getManagedResources().get(proxy.getResourceId()));
proxy.close();
// Ensure it's unregistered
-
Assertions.assertNull(DefaultResourceManager.get().getManagedResources().get(proxy.getResourceId()));
+
Assertions.assertNull(atManager.getManagedResources().get(proxy.getResourceId()));
}
}
diff --git
a/test-suite/test-new-version/src/test/java/org/apache/seata/core/rpc/netty/TmNettyClientTest.java
b/test-suite/test-new-version/src/test/java/org/apache/seata/core/rpc/netty/TmNettyClientTest.java
index e1cf6e6957..434f289ea2 100644
---
a/test-suite/test-new-version/src/test/java/org/apache/seata/core/rpc/netty/TmNettyClientTest.java
+++
b/test-suite/test-new-version/src/test/java/org/apache/seata/core/rpc/netty/TmNettyClientTest.java
@@ -91,7 +91,7 @@ public class TmNettyClientTest extends BaseNettyClientTest {
@Test
public void testSendMsgWithResponse() throws Exception {
int dynamicPort = getDynamicPort();
- ServerInstance serverInstance = startServerSimple(dynamicPort);
+ ServerInstance serverInstance = startServer(dynamicPort);
try {
configureClient(dynamicPort);
@@ -111,7 +111,7 @@ public class TmNettyClientTest extends BaseNettyClientTest {
request.setXid("127.0.0.1:" + dynamicPort + ":1249853");
GlobalCommitResponse globalCommitResponse = null;
try {
- globalCommitResponse = (GlobalCommitResponse)
tmNettyRemotingClient.sendSyncRequest(request);
+ globalCommitResponse = (GlobalCommitResponse)
tmNettyRemotingClient.sendSyncRequest(channel, request);
} catch (TimeoutException e) {
throw new RuntimeException(e);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
