You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@seata.apache.org by ji...@apache.org on 2026/10/04 16:01:15 UTC
(incubator-seata) branch 2.x updated: test: expand server unit test coverage (#8293)
This is an automated email from the ASF dual-hosted git repository.
slievrly 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 d38c41f355 test: expand server unit test coverage (#8293)
d38c41f355 is described below
commit d38c41f35553e7f0650bde80242f7aad41c4850d
Author: jimin <sl...@163.com>
AuthorDate: Mon Oct 5 00:01:10 2026 +0800
test: expand server unit test coverage (#8293)
---
changes/en-us/2.x.md | 1 +
changes/zh-cn/2.x.md | 1 +
.../seata/server/InboundFailureUnitTest.java | 57 +++++
.../cluster/raft/RaftMetadataSyncUnitTest.java | 97 ++++++++
.../cluster/raft/RaftServerLifecycleUnitTest.java | 101 ++++++++
.../raft/snapshot/StoreSnapshotUnitTest.java | 143 +++++++++++
.../controller/ConsoleControllerUnitTest.java | 99 ++++++++
.../console/entity/vo/GlobalLockViewUnitTest.java | 59 +++++
.../console/impl/redis/RedisConsoleUnitTest.java | 130 ++++++++++
.../coordinator/CoordinatorRecoveryUnitTest.java | 143 +++++++++++
.../coordinator/CoreBranchOutcomeUnitTest.java | 214 ++++++++++++++++
.../seata/server/env/PortConfigUnitTest.java | 57 +++++
.../server/filter/RaftRequestFilterUnitTest.java | 128 ++++++++++
.../seata/server/filter/XssRequestUnitTest.java | 69 ++++++
.../instance/GeneralInstanceStrategyUnitTest.java | 102 ++++++++
.../ratelimit/RateLimitConfigurationUnitTest.java | 84 +++++++
.../server/metrics/MetricsSubscriberUnitTest.java | 141 +++++++++++
.../server/session/SessionHelperUnitTest.java | 115 +++++++++
.../session/SessionHolderRecoveryUnitTest.java | 111 +++++++++
.../server/session/SessionLifecycleUnitTest.java | 114 +++++++++
.../spring/listener/ServerEnvironmentUnitTest.java | 76 ++++++
.../db/lock/DataBaseDistributedLockerUnitTest.java | 183 ++++++++++++++
.../storage/db/lock/LockStoreDaoUnitTest.java | 172 +++++++++++++
.../storage/db/store/DatabaseManagersUnitTest.java | 115 +++++++++
.../storage/db/store/LogStoreDaoUnitTest.java | 120 +++++++++
.../storage/db/store/VGroupMappingDaoUnitTest.java | 95 +++++++
.../file/session/FileSessionReplayUnitTest.java | 134 ++++++++++
.../file/store/FileMappingFailureUnitTest.java | 49 ++++
.../file/store/FileStoreFailureUnitTest.java | 139 +++++++++++
.../server/storage/redis/JedisPoolUnitTest.java | 142 +++++++++++
.../storage/redis/RedisFactoriesUnitTest.java | 103 ++++++++
.../storage/redis/lock/RedisLockerUnitTest.java | 156 ++++++++++++
.../storage/redis/lock/RedisLuaLockerUnitTest.java | 125 ++++++++++
.../redis/session/RedisSessionManagerUnitTest.java | 80 ++++++
.../storage/redis/store/RedisLuaStoreUnitTest.java | 148 +++++++++++
.../redis/store/RedisTransactionStoreUnitTest.java | 273 +++++++++++++++++++++
.../redis/store/RedisVGroupMappingUnitTest.java | 68 +++++
.../server/transaction/at/ATLockUnitTest.java | 80 ++++++
.../server/transaction/saga/SagaCoreUnitTest.java | 204 +++++++++++++++
39 files changed, 4428 insertions(+)
diff --git a/changes/en-us/2.x.md b/changes/en-us/2.x.md
index 4c581c54f6..54c2aedfa5 100644
--- a/changes/en-us/2.x.md
+++ b/changes/en-us/2.x.md
@@ -58,6 +58,7 @@ Add changes here for all PR submitted to the 2.x branch.
- [[#8290](https://github.com/apache/incubator-seata/pull/8290)] cover map cloning and equality, collection shrink recovery and self-references, and duration parsing failures
- [[#8289](https://github.com/apache/incubator-seata/pull/8289)] verify non-negative atomic counter results across integer overflow
+- [[#8293](https://github.com/apache/incubator-seata/pull/8293)] add server transaction, storage, Raft, and console unit tests
- [[#8255](https://github.com/apache/incubator-seata/pull/8255)] initialize H2 session tests without connecting to the default MySQL store
- [[#8288](https://github.com/apache/incubator-seata/pull/8288)] cover Spring fence transactions, scanner proxy behavior and datasource proxy registration
- [[#8283](https://github.com/apache/incubator-seata/pull/8283)] cover cycle dependency handler cleanup, nested calls, thread isolation, and identity tracking
diff --git a/changes/zh-cn/2.x.md b/changes/zh-cn/2.x.md
index 7e5dba0daf..7b23b56695 100644
--- a/changes/zh-cn/2.x.md
+++ b/changes/zh-cn/2.x.md
@@ -57,6 +57,7 @@
- [[#8290](https://github.com/apache/incubator-seata/pull/8290)] 补充 Map 克隆与相等判断、集合缩短恢复与自引用以及时长解析异常测试
- [[#8289](https://github.com/apache/incubator-seata/pull/8289)] 验证原子计数器跨整数溢出边界后的非负结果
+- [[#8293](https://github.com/apache/incubator-seata/pull/8293)] 补充服务端事务、存储、Raft 和控制台单元测试
- [[#8255](https://github.com/apache/incubator-seata/pull/8255)] 避免 H2 会话测试初始化默认 MySQL 存储并使建表失败显式失败
- [[#8288](https://github.com/apache/incubator-seata/pull/8288)] 补充 Spring Fence 事务、扫描器代理行为及数据源代理注册单元测试
- [[#8283](https://github.com/apache/incubator-seata/pull/8283)] 补充循环依赖处理器的清理、嵌套调用、线程隔离和对象身份跟踪测试
diff --git a/server/src/test/java/org/apache/seata/server/InboundFailureUnitTest.java b/server/src/test/java/org/apache/seata/server/InboundFailureUnitTest.java
new file mode 100644
index 0000000000..f7d4ed0a60
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/InboundFailureUnitTest.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.server;
+
+import org.apache.seata.common.exception.StoreException;
+import org.apache.seata.core.model.GlobalStatus;
+import org.apache.seata.core.protocol.transaction.*;
+import org.apache.seata.server.session.*;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class InboundFailureUnitTest extends BaseSpringBootTest {
+ @Test
+ void commitAndRollbackFailuresReportPersistedStatus() throws Exception {
+ AbstractTCInboundHandler handler = mock(AbstractTCInboundHandler.class, CALLS_REAL_METHODS);
+ GlobalCommitRequest commit = new GlobalCommitRequest();
+ commit.setXid("xid");
+ GlobalRollbackRequest rollback = new GlobalRollbackRequest();
+ rollback.setXid("xid");
+ doThrow(new StoreException("offline")).when(handler).doGlobalCommit(eq(commit), any(), isNull());
+ doThrow(new StoreException("offline")).when(handler).doGlobalRollback(eq(rollback), any(), isNull());
+ GlobalSession session = mock(GlobalSession.class);
+ when(session.getStatus()).thenReturn(GlobalStatus.CommitRetrying);
+ try (MockedStatic<SessionHolder> holder = mockStatic(SessionHolder.class)) {
+ holder.when(() -> SessionHolder.findGlobalSession("xid", false)).thenReturn(session);
+ assertEquals(
+ GlobalStatus.CommitRetrying, handler.handle(commit, null).getGlobalStatus());
+ assertEquals(
+ GlobalStatus.CommitRetrying, handler.handle(rollback, null).getGlobalStatus());
+ holder.when(() -> SessionHolder.findGlobalSession("xid", false)).thenReturn(null);
+ assertEquals(GlobalStatus.Finished, handler.handle(commit, null).getGlobalStatus());
+ assertEquals(GlobalStatus.Finished, handler.handle(rollback, null).getGlobalStatus());
+ holder.when(() -> SessionHolder.findGlobalSession("xid", false))
+ .thenThrow(new IllegalStateException("read failed"));
+ assertEquals(GlobalStatus.Committing, handler.handle(commit, null).getGlobalStatus());
+ assertEquals(
+ GlobalStatus.Rollbacking, handler.handle(rollback, null).getGlobalStatus());
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/cluster/raft/RaftMetadataSyncUnitTest.java b/server/src/test/java/org/apache/seata/server/cluster/raft/RaftMetadataSyncUnitTest.java
new file mode 100644
index 0000000000..23ce0d75b3
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/cluster/raft/RaftMetadataSyncUnitTest.java
@@ -0,0 +1,97 @@
+/*
+ * 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.server.cluster.raft;
+
+import com.alipay.sofa.jraft.RouteTable;
+import com.alipay.sofa.jraft.conf.Configuration;
+import com.alipay.sofa.jraft.entity.PeerId;
+import com.alipay.sofa.jraft.rpc.*;
+import com.alipay.sofa.jraft.rpc.impl.cli.CliClientServiceImpl;
+import org.apache.seata.common.metadata.*;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.cluster.raft.processor.request.PutNodeMetadataRequest;
+import org.apache.seata.server.cluster.raft.processor.response.PutNodeMetadataResponse;
+import org.apache.seata.server.cluster.raft.sync.msg.dto.RaftClusterMetadata;
+import org.junit.jupiter.api.Test;
+import org.mockito.*;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class RaftMetadataSyncUnitTest extends BaseSpringBootTest {
+ @Test
+ void metadataSyncPublishesRoleAndHandlesSuccessRejectionAndTransportFailure() throws Exception {
+ RaftStateMachine machine = mock(RaftStateMachine.class, CALLS_REAL_METHODS);
+ RaftClusterMetadata metadata = new RaftClusterMetadata();
+ Node leader = new Node();
+ leader.setVersion("2.8.0");
+ metadata.setLeader(leader);
+ AtomicBoolean syncing = new AtomicBoolean(true);
+ ScheduledFuture<?> future = mock(ScheduledFuture.class);
+ ReflectionTestUtils.setField(machine, "group", "unit");
+ ReflectionTestUtils.setField(machine, "initSync", syncing);
+ ReflectionTestUtils.setField(machine, "scheduledFuture", future);
+ machine.setRaftLeaderMetadata(metadata);
+ PeerId local = new PeerId("127.0.0.1", 9091), remote = new PeerId("127.0.0.1", 9092);
+ RaftServer server = mock(RaftServer.class);
+ when(server.getServerId()).thenReturn(local);
+ CliClientServiceImpl cli = mock(CliClientServiceImpl.class);
+ RpcClient client = mock(RpcClient.class);
+ when(cli.getRpcClient()).thenReturn(client);
+ RouteTable routes = mock(RouteTable.class);
+ when(routes.getConfiguration("unit")).thenReturn(new Configuration(Collections.singletonList(local)));
+ try (MockedStatic<RaftServerManager> servers = mockStatic(RaftServerManager.class);
+ MockedStatic<RouteTable> table = mockStatic(RouteTable.class)) {
+ servers.when(() -> RaftServerManager.getRaftServer("unit")).thenReturn(server);
+ servers.when(RaftServerManager::getCliClientServiceInstance).thenReturn(cli);
+ table.when(RouteTable::getInstance).thenReturn(routes);
+ ReflectionTestUtils.invokeMethod(machine, "syncCurrentNodeInfo", remote);
+ ArgumentCaptor<InvokeCallback> callback = ArgumentCaptor.forClass(InvokeCallback.class);
+ ArgumentCaptor<PutNodeMetadataRequest> request = ArgumentCaptor.forClass(PutNodeMetadataRequest.class);
+ verify(client)
+ .invokeAsync(
+ eq(remote.getEndpoint()),
+ request.capture(),
+ any(InvokeContext.class),
+ callback.capture(),
+ eq(30000L));
+ assertEquals(ClusterRole.FOLLOWER, request.getValue().getNode().getRole());
+ PutNodeMetadataResponse response = new PutNodeMetadataResponse(true);
+ response.setSuccess(true);
+ callback.getValue().complete(response, null);
+ verify(future).cancel(true);
+ response.setSuccess(false);
+ callback.getValue().complete(response, null);
+ assertFalse(syncing.get());
+ syncing.set(true);
+ callback.getValue().complete(null, new IllegalStateException("offline"));
+ assertFalse(syncing.get());
+ metadata.setLeader(null);
+ syncing.set(true);
+ ReflectionTestUtils.invokeMethod(machine, "syncCurrentNodeInfo", remote);
+ assertFalse(syncing.get());
+ when(routes.selectLeader("unit")).thenReturn(null);
+ ReflectionTestUtils.invokeMethod(machine, "syncCurrentNodeInfo", "unit");
+ assertFalse(syncing.get());
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/cluster/raft/RaftServerLifecycleUnitTest.java b/server/src/test/java/org/apache/seata/server/cluster/raft/RaftServerLifecycleUnitTest.java
new file mode 100644
index 0000000000..c8d49d8c63
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/cluster/raft/RaftServerLifecycleUnitTest.java
@@ -0,0 +1,101 @@
+/*
+ * 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.server.cluster.raft;
+
+import com.alipay.sofa.jraft.*;
+import com.alipay.sofa.jraft.entity.PeerId;
+import com.alipay.sofa.jraft.option.NodeOptions;
+import com.alipay.sofa.jraft.rpc.RpcServer;
+import org.apache.seata.config.Configuration;
+import org.apache.seata.config.ConfigurationFactory;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.*;
+
+import java.nio.file.Path;
+import java.util.*;
+
+import static org.apache.seata.common.ConfigurationKeys.SERVER_RAFT_SSL_ENABLED;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class RaftServerLifecycleUnitTest extends BaseSpringBootTest {
+ @TempDir
+ Path directory;
+
+ @Test
+ void startupConfiguresStorageSslAndShutdownJoinsGroup() throws Exception {
+ Class.forName(RaftStateMachine.class.getName());
+ String[] names = {
+ "bolt.server.ssl.enable",
+ "bolt.server.ssl.clientAuth",
+ "bolt.client.ssl.enable",
+ "bolt.server.ssl.keystore",
+ "bolt.server.ssl.keystore.password",
+ "bolt.server.ssl.keystore.type",
+ "bolt.server.ssl.kmf.algorithm",
+ "bolt.client.ssl.keystore",
+ "bolt.client.ssl.keystore.password",
+ "bolt.client.ssl.keystore.type",
+ "bolt.client.ssl.tmf.algorithm"
+ };
+ Map<String, String> previous = new HashMap<>();
+ for (String name : names) {
+ previous.put(name, System.getProperty(name));
+ }
+ Configuration config = mock(Configuration.class);
+ when(config.getBoolean(eq(SERVER_RAFT_SSL_ENABLED), anyBoolean())).thenReturn(true);
+ when(config.getConfig(anyString())).thenReturn("fixture");
+ NodeOptions options = new NodeOptions();
+ Node node = mock(Node.class);
+ when(node.getOptions()).thenReturn(options);
+ RouteTable routes = mock(RouteTable.class);
+ try (MockedStatic<ConfigurationFactory> factory = mockStatic(ConfigurationFactory.class);
+ MockedStatic<RouteTable> table = mockStatic(RouteTable.class);
+ MockedConstruction<RaftStateMachine> machines = mockConstruction(RaftStateMachine.class);
+ MockedConstruction<RaftGroupService> groups =
+ mockConstruction(RaftGroupService.class, (group, context) -> when(group.start(false))
+ .thenReturn(node))) {
+ factory.when(ConfigurationFactory::getInstance).thenReturn(config);
+ table.when(RouteTable::getInstance).thenReturn(routes);
+ RaftServer server = new RaftServer(
+ directory.toString(), "group", new PeerId("127.0.0.1", 9091), options, mock(RpcServer.class));
+ server.start();
+ assertTrue(options.getLogUri().endsWith("group/log"));
+ assertTrue(options.getSnapshotUri().endsWith("group/snapshot"));
+ assertSame(node, server.getNode());
+ assertEquals("true", System.getProperty("bolt.server.ssl.clientAuth"));
+ assertEquals("fixture", System.getProperty("bolt.client.ssl.keystore"));
+ server.close();
+ verify(groups.constructed().get(0)).shutdown();
+ verify(groups.constructed().get(0)).join();
+ doThrow(new InterruptedException("interrupted"))
+ .when(groups.constructed().get(0))
+ .join();
+ assertDoesNotThrow(server::destroy);
+ } finally {
+ previous.forEach((key, value) -> {
+ if (value == null) {
+ System.clearProperty(key);
+ } else {
+ System.setProperty(key, value);
+ }
+ });
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/cluster/raft/snapshot/StoreSnapshotUnitTest.java b/server/src/test/java/org/apache/seata/server/cluster/raft/snapshot/StoreSnapshotUnitTest.java
new file mode 100644
index 0000000000..c0b77b8d2b
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/cluster/raft/snapshot/StoreSnapshotUnitTest.java
@@ -0,0 +1,143 @@
+/*
+ * 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.server.cluster.raft.snapshot;
+
+import com.alipay.sofa.jraft.entity.LocalFileMetaOutter;
+import com.alipay.sofa.jraft.entity.RaftOutter;
+import com.alipay.sofa.jraft.storage.snapshot.SnapshotReader;
+import com.alipay.sofa.jraft.storage.snapshot.SnapshotWriter;
+import org.apache.seata.core.store.MappingDO;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.cluster.raft.*;
+import org.apache.seata.server.cluster.raft.snapshot.metadata.LeaderMetadataSnapshotFile;
+import org.apache.seata.server.cluster.raft.snapshot.session.SessionSnapshotFile;
+import org.apache.seata.server.cluster.raft.snapshot.vgroup.VGroupSnapshotFile;
+import org.apache.seata.server.cluster.raft.sync.msg.dto.RaftClusterMetadata;
+import org.apache.seata.server.lock.*;
+import org.apache.seata.server.session.*;
+import org.apache.seata.server.storage.raft.session.RaftSessionManager;
+import org.apache.seata.server.storage.raft.store.RaftVGroupMappingStoreManager;
+import org.junit.jupiter.api.*;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockedStatic;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class StoreSnapshotUnitTest extends BaseSpringBootTest {
+ @TempDir
+ Path directory;
+
+ private SnapshotWriter writer;
+ private SnapshotReader reader;
+
+ @BeforeEach
+ void open() {
+ writer = mock(SnapshotWriter.class);
+ reader = mock(SnapshotReader.class);
+ when(writer.getPath()).thenReturn(directory.toString());
+ when(reader.getPath()).thenReturn(directory.toString());
+ when(writer.addFile(anyString())).thenReturn(true);
+ when(reader.getFileMeta(anyString())).thenReturn(LocalFileMetaOutter.LocalFileMeta.getDefaultInstance());
+ when(reader.load())
+ .thenReturn(RaftOutter.SnapshotMeta.newBuilder()
+ .setLastIncludedIndex(3)
+ .setLastIncludedTerm(1)
+ .build());
+ }
+
+ @Test
+ void vgroupRestoreReplacesOnlyTargetGroup() throws Exception {
+ RaftVGroupMappingStoreManager manager = mock(RaftVGroupMappingStoreManager.class);
+ MappingDO mapping = new MappingDO();
+ mapping.setVGroup("payments");
+ mapping.setNamespace("tenant");
+ mapping.setCluster("east");
+ Map<String, MappingDO> mappings = new HashMap<>(Collections.singletonMap("payments", mapping));
+ when(manager.loadVGroupsByUnit("group")).thenReturn(mappings);
+ try (MockedStatic<SessionHolder> holder = mockStatic(SessionHolder.class)) {
+ holder.when(SessionHolder::getRootVGroupMappingManager).thenReturn(manager);
+ VGroupSnapshotFile file = spy(new VGroupSnapshotFile("group"));
+ assertTrue(file.save(writer).isOk());
+ doReturn(mappings).when(file).load(anyString());
+ assertTrue(file.load(reader));
+ verify(manager).clear("group");
+ verify(manager)
+ .localAddVGroups(
+ argThat(m -> "tenant".equals(m.get("payments").getNamespace())), eq("group"));
+ checkFailures(file);
+ }
+ }
+
+ @Test
+ void sessionRoundTripClearsOldSessionsAndLocks() throws Exception {
+ RaftSessionManager manager = mock(RaftSessionManager.class);
+ LockManager locks = mock(LockManager.class);
+ GlobalSession session = GlobalSession.createGlobalSession("app", "group", "transaction", 1000);
+ Map<String, GlobalSession> sessions = new HashMap<>();
+ sessions.put(session.getXid(), session);
+ when(manager.getSessionMap()).thenReturn(sessions);
+ try (MockedStatic<SessionHolder> holder = mockStatic(SessionHolder.class);
+ MockedStatic<LockerManagerFactory> factory = mockStatic(LockerManagerFactory.class)) {
+ holder.when(() -> SessionHolder.getRootSessionManager("group")).thenReturn(manager);
+ factory.when(LockerManagerFactory::getLockManager).thenReturn(locks);
+ SessionSnapshotFile file = new SessionSnapshotFile("group");
+ assertTrue(file.save(writer).isOk());
+ sessions.put("stale", new GlobalSession());
+ assertTrue(file.load(reader));
+ assertEquals(Collections.singleton(session.getXid()), sessions.keySet());
+ assertEquals("transaction", sessions.get(session.getXid()).getTransactionName());
+ verify(locks).cleanAllLocks();
+ checkFailures(file);
+ }
+ }
+
+ @Test
+ void leaderMetadataRoundTripUpdatesStateMachine() throws Exception {
+ RaftServer server = mock(RaftServer.class);
+ RaftStateMachine machine = mock(RaftStateMachine.class);
+ when(server.getRaftStateMachine()).thenReturn(machine);
+ RaftClusterMetadata metadata = new RaftClusterMetadata();
+ when(machine.getRaftLeaderMetadata()).thenReturn(metadata);
+ try (MockedStatic<RaftServerManager> manager = mockStatic(RaftServerManager.class)) {
+ manager.when(() -> RaftServerManager.getRaftServer("group")).thenReturn(server);
+ LeaderMetadataSnapshotFile file = new LeaderMetadataSnapshotFile("group");
+ assertTrue(file.save(writer).isOk());
+ assertTrue(file.load(reader));
+ verify(machine).setRaftLeaderMetadata(any(RaftClusterMetadata.class));
+ checkFailures(file);
+ }
+ }
+
+ private void checkFailures(StoreSnapshotFile original) throws Exception {
+ when(writer.addFile(anyString())).thenReturn(false);
+ assertFalse(original.save(writer).isOk());
+ StoreSnapshotFile file = mockingDetails(original).isSpy() ? original : spy(original);
+ doReturn(false).when(file).save(any(RaftSnapshot.class), anyString());
+ assertFalse(file.save(writer).isOk());
+ doThrow(new IOException("disk full")).when(file).save(any(RaftSnapshot.class), anyString());
+ assertFalse(file.save(writer).isOk());
+ doThrow(new IOException("corrupt snapshot")).when(file).load(anyString());
+ assertFalse(file.load(reader));
+ when(reader.getFileMeta(anyString())).thenReturn(null);
+ assertFalse(original.load(reader));
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/console/controller/ConsoleControllerUnitTest.java b/server/src/test/java/org/apache/seata/server/console/controller/ConsoleControllerUnitTest.java
new file mode 100644
index 0000000000..f87f88e829
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/console/controller/ConsoleControllerUnitTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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.server.console.controller;
+
+import org.apache.seata.common.result.*;
+import org.apache.seata.server.console.entity.param.*;
+import org.apache.seata.server.console.service.*;
+import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class ConsoleControllerUnitTest {
+ @Test
+ void globalSessionServiceRoutesCommands() {
+ GlobalSessionService service = mock(GlobalSessionService.class);
+ GlobalSessionController controller = new GlobalSessionController();
+ ReflectionTestUtils.setField(controller, "globalSessionService", service);
+ SingleResult<Void> result = SingleResult.success();
+ when(service.deleteGlobalSession("xid")).thenReturn(result);
+ assertSame(result, controller.deleteGlobalSession("xid"));
+ verify(service).deleteGlobalSession("xid");
+ when(service.forceDeleteGlobalSession("xid")).thenReturn(result);
+ assertSame(result, controller.forceDeleteGlobalSession("xid"));
+ verify(service).forceDeleteGlobalSession("xid");
+ when(service.stopGlobalRetry("xid")).thenReturn(result);
+ assertSame(result, controller.stopGlobalSession("xid"));
+ verify(service).stopGlobalRetry("xid");
+ when(service.startGlobalRetry("xid")).thenReturn(result);
+ assertSame(result, controller.startGlobalSession("xid"));
+ verify(service).startGlobalRetry("xid");
+ when(service.sendCommitOrRollback("xid")).thenReturn(result);
+ assertSame(result, controller.sendCommitOrRollback("xid"));
+ verify(service).sendCommitOrRollback("xid");
+ when(service.changeGlobalStatus("xid")).thenReturn(result);
+ assertSame(result, controller.changeGlobalStatus("xid"));
+ verify(service).changeGlobalStatus("xid");
+ }
+
+ @Test
+ void branchSessionServiceRoutesCommands() {
+ BranchSessionService service = mock(BranchSessionService.class);
+ BranchSessionController controller = new BranchSessionController();
+ ReflectionTestUtils.setField(controller, "branchSessionService", service);
+ SingleResult<Void> result = SingleResult.success();
+ when(service.deleteBranchSession("xid", "7")).thenReturn(result);
+ assertSame(result, controller.deleteBranchSession("xid", "7"));
+ verify(service).deleteBranchSession("xid", "7");
+ when(service.forceDeleteBranchSession("xid", "7")).thenReturn(result);
+ assertSame(result, controller.forceDeleteBranchSession("xid", "7"));
+ verify(service).forceDeleteBranchSession("xid", "7");
+ when(service.stopBranchRetry("xid", "7")).thenReturn(result);
+ assertSame(result, controller.stopBranchSession("xid", "7"));
+ verify(service).stopBranchRetry("xid", "7");
+ when(service.startBranchRetry("xid", "7")).thenReturn(result);
+ assertSame(result, controller.startBranchRetry("xid", "7"));
+ verify(service).startBranchRetry("xid", "7");
+ }
+
+ @Test
+ void globalQueryPropagatesServiceFailure() {
+ GlobalSessionService service = mock(GlobalSessionService.class);
+ GlobalSessionController controller = new GlobalSessionController();
+ ReflectionTestUtils.setField(controller, "globalSessionService", service);
+ GlobalSessionParam param = new GlobalSessionParam();
+ PageResult result = PageResult.failure("invalid", "bad request");
+ when(service.query(param)).thenReturn(result);
+ assertSame(result, controller.query(param));
+ }
+
+ @Test
+ void lockControllerPreservesQueriesAndCommands() {
+ GlobalLockService service = mock(GlobalLockService.class);
+ GlobalLockController controller = new GlobalLockController();
+ ReflectionTestUtils.setField(controller, "globalLockService", service);
+ GlobalLockParam param = new GlobalLockParam();
+ PageResult page = PageResult.failure("invalid", "bad request");
+ when(service.query(param)).thenReturn(page);
+ assertSame(page, controller.query(param));
+ SingleResult<Void> deleted = SingleResult.success();
+ when(service.deleteLock(param)).thenReturn(deleted);
+ assertSame(deleted, controller.delete(param));
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/console/entity/vo/GlobalLockViewUnitTest.java b/server/src/test/java/org/apache/seata/server/console/entity/vo/GlobalLockViewUnitTest.java
new file mode 100644
index 0000000000..3ede61fca9
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/console/entity/vo/GlobalLockViewUnitTest.java
@@ -0,0 +1,59 @@
+/*
+ * 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.server.console.entity.vo;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.seata.core.constants.ServerTableColumnsName;
+import org.junit.jupiter.api.Test;
+
+import java.sql.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class GlobalLockViewUnitTest {
+ @Test
+ void sqlConversionSerializesLargeIdsAsStringsAndRetainsTimestamps() throws Exception {
+ ResultSet rs = mock(ResultSet.class);
+ when(rs.getString(ServerTableColumnsName.LOCK_TABLE_XID)).thenReturn("host:8091:9007199254740993");
+ when(rs.getLong(ServerTableColumnsName.LOCK_TABLE_TRANSACTION_ID)).thenReturn(9007199254740993L);
+ when(rs.getLong(ServerTableColumnsName.LOCK_TABLE_BRANCH_ID)).thenReturn(9007199254740995L);
+ when(rs.getString(ServerTableColumnsName.LOCK_TABLE_RESOURCE_ID)).thenReturn("db");
+ when(rs.getString(ServerTableColumnsName.LOCK_TABLE_TABLE_NAME)).thenReturn("orders");
+ when(rs.getString(ServerTableColumnsName.LOCK_TABLE_PK)).thenReturn("7");
+ when(rs.getString(ServerTableColumnsName.LOCK_TABLE_ROW_KEY)).thenReturn("db^^^orders^^^7");
+ when(rs.getTimestamp(ServerTableColumnsName.LOCK_TABLE_GMT_CREATE)).thenReturn(new Timestamp(1000));
+ when(rs.getTimestamp(ServerTableColumnsName.LOCK_TABLE_GMT_MODIFIED)).thenReturn(new Timestamp(2000));
+ GlobalLockVO view = GlobalLockVO.convert(rs);
+ view.setVgroup("payments");
+ JsonNode json = new ObjectMapper().valueToTree(view);
+ assertTrue(json.get("transactionId").isTextual());
+ assertEquals("9007199254740993", json.get("transactionId").asText());
+ assertEquals("9007199254740995", json.get("branchId").asText());
+ assertEquals("payments", json.get("vgroup").asText());
+ assertEquals("orders", json.get("tableName").asText());
+ assertEquals("db", json.get("resourceId").asText());
+ assertEquals("7", json.get("pk").asText());
+ assertEquals(1000, json.get("gmtCreate").asLong());
+ assertEquals(2000, json.get("gmtModified").asLong());
+ assertTrue(view.toString().contains("9007199254740993"));
+ when(rs.getTimestamp(anyString())).thenReturn(null);
+ assertNull(GlobalLockVO.convert(rs).getGmtCreate());
+ assertNull(GlobalLockVO.convert(rs).getGmtModified());
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/console/impl/redis/RedisConsoleUnitTest.java b/server/src/test/java/org/apache/seata/server/console/impl/redis/RedisConsoleUnitTest.java
new file mode 100644
index 0000000000..687fe5ef49
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/console/impl/redis/RedisConsoleUnitTest.java
@@ -0,0 +1,130 @@
+/*
+ * 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.server.console.impl.redis;
+
+import org.apache.seata.common.result.PageResult;
+import org.apache.seata.core.exception.TransactionException;
+import org.apache.seata.core.model.GlobalStatus;
+import org.apache.seata.core.store.BranchTransactionDO;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.console.entity.param.*;
+import org.apache.seata.server.console.entity.vo.*;
+import org.apache.seata.server.console.exception.ConsoleException;
+import org.apache.seata.server.lock.*;
+import org.apache.seata.server.session.*;
+import org.apache.seata.server.storage.redis.JedisPooledFactory;
+import org.apache.seata.server.storage.redis.store.*;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import redis.clients.jedis.Jedis;
+
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class RedisConsoleUnitTest extends BaseSpringBootTest {
+ @Test
+ void globalQueriesDispatchPaginationXidAndStatus() {
+ RedisTransactionStoreManager store = mock(RedisTransactionStoreManager.class);
+ try (MockedStatic<RedisTransactionStoreManagerFactory> factory =
+ mockStatic(RedisTransactionStoreManagerFactory.class)) {
+ factory.when(RedisTransactionStoreManagerFactory::getInstance).thenReturn(store);
+ GlobalSessionRedisServiceImpl service = new GlobalSessionRedisServiceImpl();
+ GlobalSessionParam p = new GlobalSessionParam();
+ p.setPageNum(1);
+ p.setPageSize(10);
+ GlobalSession session = GlobalSession.createGlobalSession("app", "group", "order", 1000);
+ when(store.countByGlobalSessions(any())).thenReturn(1L);
+ when(store.findGlobalSessionByPage(1, 10, false)).thenReturn(Collections.singletonList(session));
+ assertEquals(1, service.query(p).getTotal());
+ p.setXid(session.getXid());
+ when(store.readSession(any(SessionCondition.class))).thenReturn(Collections.singletonList(session));
+ PageResult<GlobalSessionVO> result = service.query(p);
+ assertEquals(session.getXid(), result.getData().get(0).getXid());
+ verify(store)
+ .readSession(argThat(
+ (SessionCondition c) -> c.getXid().equals(session.getXid()) && c.isLazyLoadBranch()));
+ p.setStatus(GlobalStatus.Begin.getCode());
+ assertEquals(1, service.query(p).getTotal());
+ p.setXid(null);
+ when(store.readSessionStatusByPage(p)).thenReturn(Collections.singletonList(session));
+ assertEquals(1, service.query(p).getData().size());
+ p.setTimeStart(1L);
+ assertFalse(service.query(p).isSuccess());
+ }
+ }
+
+ @Test
+ void branchQueryCopiesRecordsAndHandlesEmptyXid() {
+ RedisTransactionStoreManager store = mock(RedisTransactionStoreManager.class);
+ try (MockedStatic<RedisTransactionStoreManagerFactory> factory =
+ mockStatic(RedisTransactionStoreManagerFactory.class)) {
+ factory.when(RedisTransactionStoreManagerFactory::getInstance).thenReturn(store);
+ BranchSessionRedisServiceImpl service = new BranchSessionRedisServiceImpl();
+ assertTrue(service.queryByXid(" ").getData().isEmpty());
+ factory.verifyNoInteractions();
+ assertTrue(service.queryByXid("missing").getData().isEmpty());
+ BranchTransactionDO branch = new BranchTransactionDO();
+ branch.setXid("xid");
+ branch.setBranchId(7L);
+ branch.setResourceId("orders");
+ when(store.findBranchSessionByXid("xid")).thenReturn(Collections.singletonList(branch));
+ BranchSessionVO result = service.queryByXid("xid").getData().get(0);
+ assertEquals("7", result.getBranchId());
+ assertEquals("orders", result.getResourceId());
+ }
+ }
+
+ @Test
+ void lockQueriesResolveIndexesAndRejectIncompleteParameters() throws Exception {
+ Jedis jedis = mock(Jedis.class);
+ LockManager locks = mock(LockManager.class);
+ try (MockedStatic<JedisPooledFactory> factory = mockStatic(JedisPooledFactory.class);
+ MockedStatic<LockerManagerFactory> managers = mockStatic(LockerManagerFactory.class)) {
+ factory.when(JedisPooledFactory::getJedisInstance).thenReturn(jedis);
+ managers.when(LockerManagerFactory::getLockManager).thenReturn(locks);
+ GlobalLockRedisServiceImpl service = new GlobalLockRedisServiceImpl();
+ GlobalLockParam p = new GlobalLockParam();
+ p.setPageNum(1);
+ p.setPageSize(10);
+ assertFalse(service.query(p).isSuccess());
+ p.setXid("xid");
+ assertTrue(service.query(p).getData().isEmpty());
+ Map<String, String> row = new HashMap<>();
+ row.put("xid", "xid");
+ row.put("branchId", "7");
+ row.put("pk", "1");
+ when(jedis.hgetAll("SEATA_GLOBAL_LOCKxid")).thenReturn(Collections.singletonMap("7", "row1;row2"));
+ when(jedis.hgetAll("row1")).thenReturn(row);
+ when(jedis.hgetAll("row2")).thenReturn(row);
+ assertEquals(2, service.query(p).getData().size());
+ p.setXid(null);
+ p.setTableName("orders");
+ p.setPk("1");
+ p.setResourceId("db");
+ when(jedis.hgetAll("SEATA_ROW_LOCK_db^^^orders^^^1")).thenReturn(row);
+ assertEquals(1, service.query(p).getData().size());
+ p.setXid("xid");
+ p.setBranchId("7");
+ assertTrue(service.deleteLock(p).isSuccess());
+ verify(locks).releaseLock(argThat(b -> b.getBranchId() == 7L && "xid".equals(b.getXid())));
+ doThrow(new TransactionException("failed")).when(locks).releaseLock(any(BranchSession.class));
+ assertThrows(ConsoleException.class, () -> service.deleteLock(p));
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/coordinator/CoordinatorRecoveryUnitTest.java b/server/src/test/java/org/apache/seata/server/coordinator/CoordinatorRecoveryUnitTest.java
new file mode 100644
index 0000000000..252e5303de
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/coordinator/CoordinatorRecoveryUnitTest.java
@@ -0,0 +1,143 @@
+/*
+ * 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.server.coordinator;
+
+import io.netty.channel.Channel;
+import org.apache.seata.core.context.RootContext;
+import org.apache.seata.core.exception.TransactionException;
+import org.apache.seata.core.model.GlobalStatus;
+import org.apache.seata.core.protocol.transaction.*;
+import org.apache.seata.core.rpc.RemotingServer;
+import org.apache.seata.core.rpc.netty.ChannelManager;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.session.*;
+import org.junit.jupiter.api.*;
+import org.mockito.MockedStatic;
+import org.slf4j.MDC;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.*;
+import java.util.concurrent.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class CoordinatorRecoveryUnitTest extends BaseSpringBootTest {
+ private DefaultCoordinator coordinator;
+ private DefaultCore core;
+ private SessionManager manager;
+ private ScheduledExecutorService scheduler;
+ private RemotingServer remoting;
+ private MockedStatic<SessionHolder> holder;
+
+ @BeforeEach
+ void open() {
+ coordinator = mock(DefaultCoordinator.class, CALLS_REAL_METHODS);
+ core = mock(DefaultCore.class);
+ manager = mock(SessionManager.class);
+ scheduler = mock(ScheduledThreadPoolExecutor.class);
+ remoting = mock(RemotingServer.class);
+ ReflectionTestUtils.setField(coordinator, "core", core);
+ ReflectionTestUtils.setField(coordinator, "syncProcessing", scheduler);
+ coordinator.setRemotingServer(remoting);
+ holder = mockStatic(SessionHolder.class);
+ holder.when(SessionHolder::getRootSessionManager).thenReturn(manager);
+ holder.when(() -> SessionHolder.lockAndExecute(any(), any()))
+ .thenAnswer(i -> ((GlobalSession.LockCallable<?>) i.getArgument(1)).call());
+ }
+
+ @AfterEach
+ void close() {
+ holder.close();
+ MDC.remove(RootContext.MDC_KEY_XID);
+ }
+
+ private GlobalSession session(GlobalStatus status) {
+ GlobalSession s = mock(GlobalSession.class);
+ when(s.getXid()).thenReturn("host:8091:42");
+ when(s.getStatus()).thenReturn(status);
+ when(s.getBeginTime()).thenReturn(System.currentTimeMillis());
+ when(s.timeToDeadSession()).thenReturn(0L);
+ when(manager.findGlobalSessions(any())).thenReturn(new ArrayList<>(Collections.singletonList(s)));
+ return s;
+ }
+
+ @Test
+ void scheduledRecoveryInvokesCoreAndReschedulesAfterFailure() throws Exception {
+ GlobalSession s = session(GlobalStatus.Committing);
+ coordinator.handleCommittingByScheduled();
+ verify(core).doGlobalCommit(s, true);
+ verify(scheduler).schedule(any(Runnable.class), anyLong(), eq(TimeUnit.MILLISECONDS));
+ doThrow(new TransactionException("failed")).when(core).doGlobalCommit(s, true);
+ assertDoesNotThrow(coordinator::handleCommittingByScheduled);
+ coordinator.handleRollbackingByScheduled();
+ verify(core).doGlobalRollback(s, true);
+ doThrow(new TransactionException("failed")).when(core).doGlobalRollback(s, true);
+ assertDoesNotThrow(coordinator::handleRollbackingByScheduled);
+ coordinator.handleEndStatesByScheduled();
+ verify(scheduler, times(5)).schedule(any(Runnable.class), anyLong(), eq(TimeUnit.MILLISECONDS));
+ }
+
+ @Test
+ void retryAndAsyncRecoveryAttemptCoreAndContainTransactionFailures() throws Exception {
+ GlobalSession s = session(GlobalStatus.CommitRetrying);
+ coordinator.handleRetryCommitting();
+ coordinator.handleRetryRollbacking();
+ coordinator.handleAsyncCommitting();
+ verify(core, times(2)).doGlobalCommit(s, true);
+ verify(core).doGlobalRollback(s, true);
+ doThrow(new TransactionException("failed")).when(core).doGlobalCommit(s, true);
+ doThrow(new TransactionException("failed")).when(core).doGlobalRollback(s, true);
+ assertDoesNotThrow(coordinator::handleRetryCommitting);
+ assertDoesNotThrow(coordinator::handleRetryRollbacking);
+ assertDoesNotThrow(coordinator::handleAsyncCommitting);
+ }
+
+ @Test
+ void commitAndRollbackResponsesReflectCoreResults() throws Exception {
+ GlobalCommitRequest commit = new GlobalCommitRequest();
+ commit.setXid("xid");
+ GlobalCommitResponse committed = new GlobalCommitResponse();
+ when(core.commit("xid")).thenReturn(GlobalStatus.Committed);
+ coordinator.doGlobalCommit(commit, committed, null);
+ assertEquals(GlobalStatus.Committed, committed.getGlobalStatus());
+ GlobalRollbackRequest rollback = new GlobalRollbackRequest();
+ rollback.setXid("xid");
+ GlobalRollbackResponse rolled = new GlobalRollbackResponse();
+ when(core.rollback("xid")).thenReturn(GlobalStatus.Rollbacked);
+ coordinator.doGlobalRollback(rollback, rolled, null);
+ assertEquals(GlobalStatus.Rollbacked, rolled.getGlobalStatus());
+ }
+
+ @Test
+ void undoCleanupTargetsEachResourceAndContinuesAfterTransportFailure() throws Exception {
+ Channel a = mock(Channel.class), b = mock(Channel.class);
+ Map<String, Channel> channels = new LinkedHashMap<>();
+ channels.put("db1", a);
+ channels.put("db2", b);
+ try (MockedStatic<ChannelManager> channelManager = mockStatic(ChannelManager.class)) {
+ channelManager.when(ChannelManager::getRmChannels).thenReturn(channels);
+ doThrow(new IllegalStateException("offline")).when(remoting).sendAsyncRequest(eq(a), any());
+ coordinator.undoLogDelete();
+ verify(remoting)
+ .sendAsyncRequest(
+ eq(b),
+ argThat(r -> r instanceof UndoLogDeleteRequest
+ && "db2".equals(((UndoLogDeleteRequest) r).getResourceId())));
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/coordinator/CoreBranchOutcomeUnitTest.java b/server/src/test/java/org/apache/seata/server/coordinator/CoreBranchOutcomeUnitTest.java
new file mode 100644
index 0000000000..741ff77d2a
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/coordinator/CoreBranchOutcomeUnitTest.java
@@ -0,0 +1,214 @@
+/*
+ * 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.server.coordinator;
+
+import org.apache.seata.core.exception.TransactionException;
+import org.apache.seata.core.model.*;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.metrics.MetricsPublisher;
+import org.apache.seata.server.session.*;
+import org.junit.jupiter.api.*;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.mockito.MockedStatic;
+
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class CoreBranchOutcomeUnitTest extends BaseSpringBootTest {
+ private DefaultCore core;
+ private AbstractCore branchCore;
+ private GlobalSession global;
+ private BranchSession branch;
+ private MockedStatic<MetricsPublisher> metrics;
+ private MockedStatic<SessionHelper> helper;
+
+ @BeforeEach
+ void open() throws Exception {
+ core = mock(DefaultCore.class, CALLS_REAL_METHODS);
+ branchCore = mock(AbstractCore.class);
+ doReturn(branchCore).when(core).getCore(any());
+ global = mock(GlobalSession.class);
+ branch = mock(BranchSession.class);
+ when(branch.getBranchType()).thenReturn(BranchType.TCC);
+ when(branch.getStatus()).thenReturn(BranchStatus.Registered);
+ when(branch.getXid()).thenReturn("xid");
+ when(global.getXid()).thenReturn("xid");
+ when(global.getStatus()).thenReturn(GlobalStatus.Committing);
+ when(global.getSortedBranches()).thenReturn(Collections.singletonList(branch));
+ when(global.getReverseSortedBranches()).thenReturn(Collections.singletonList(branch));
+ metrics = mockStatic(MetricsPublisher.class);
+ helper = mockStatic(SessionHelper.class, CALLS_REAL_METHODS);
+ helper.when(() -> SessionHelper.removeBranch(any(), any(), anyBoolean()))
+ .thenAnswer(i -> null);
+ helper.when(() -> SessionHelper.endCommitted(any(), anyBoolean())).thenAnswer(i -> null);
+ helper.when(() -> SessionHelper.endRollbacked(any(), anyBoolean())).thenAnswer(i -> null);
+ helper.when(() -> SessionHelper.endCommitFailed(any(), anyBoolean())).thenAnswer(i -> null);
+ helper.when(() -> SessionHelper.endRollbackFailed(any(), anyBoolean())).thenAnswer(i -> null);
+ }
+
+ @AfterEach
+ void close() {
+ helper.close();
+ metrics.close();
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = BranchStatus.class,
+ names = {"PhaseTwo_Committed", "PhaseTwo_CommitFailed_Unretryable", "PhaseTwo_CommitFailed_Retryable"})
+ void commitOutcomeDrivesCleanupOrRetry(BranchStatus outcome) throws Exception {
+ when(branchCore.branchCommit(global, branch)).thenReturn(outcome);
+ assertEquals(outcome == BranchStatus.PhaseTwo_Committed, core.doGlobalCommit(global, false));
+ if (outcome == BranchStatus.PhaseTwo_Committed) {
+ helper.verify(() -> SessionHelper.removeBranch(global, branch, true));
+ } else if (outcome == BranchStatus.PhaseTwo_CommitFailed_Unretryable) {
+ helper.verify(() -> SessionHelper.endCommitFailed(global, false));
+ } else {
+ verify(global).queueToRetryCommit();
+ assertFalse(core.doGlobalCommit(global, true));
+ when(global.canBeCommittedAsync()).thenReturn(true);
+ assertTrue(core.doGlobalCommit(global, true));
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = BranchStatus.class,
+ names = {"PhaseTwo_Rollbacked", "PhaseTwo_RollbackFailed_Unretryable", "PhaseTwo_RollbackFailed_Retryable"})
+ void rollbackOutcomeDrivesCleanupOrRetry(BranchStatus outcome) throws Exception {
+ when(branchCore.branchRollback(global, branch)).thenReturn(outcome);
+ assertEquals(outcome == BranchStatus.PhaseTwo_Rollbacked, core.doGlobalRollback(global, false));
+ if (outcome == BranchStatus.PhaseTwo_Rollbacked) {
+ helper.verify(() -> SessionHelper.removeBranch(global, branch, true));
+ } else if (outcome == BranchStatus.PhaseTwo_RollbackFailed_Unretryable) {
+ helper.verify(() -> SessionHelper.endRollbackFailed(global, false));
+ } else {
+ verify(global).queueToRetryRollback();
+ assertFalse(core.doGlobalRollback(global, true));
+ verify(global, times(1)).queueToRetryRollback();
+ }
+ }
+
+ @Test
+ void transportExceptionsQueueOnlyFirstAttempt() throws Exception {
+ when(branchCore.branchCommit(global, branch)).thenThrow(new TransactionException("offline"));
+ assertThrows(TransactionException.class, () -> core.doGlobalCommit(global, false));
+ verify(global).queueToRetryCommit();
+ assertTrue(core.doGlobalCommit(global, true));
+ when(branchCore.branchRollback(global, branch)).thenThrow(new TransactionException("offline"));
+ assertThrows(TransactionException.class, () -> core.doGlobalRollback(global, false));
+ verify(global).queueToRetryRollback();
+ assertThrows(TransactionException.class, () -> core.doGlobalRollback(global, true));
+ verify(global, times(1)).queueToRetryRollback();
+ }
+
+ @Test
+ void deletionUsesBranchSpecificSuccessStatus() throws Exception {
+ for (BranchType type : Arrays.asList(BranchType.AT, BranchType.TCC, BranchType.XA)) {
+ when(branch.getBranchType()).thenReturn(type);
+ when(branchCore.branchDelete(global, branch))
+ .thenReturn(
+ type == BranchType.AT ? BranchStatus.PhaseTwo_Committed : BranchStatus.PhaseTwo_Rollbacked);
+ assertTrue(core.doBranchDelete(global, branch));
+ when(branchCore.branchDelete(global, branch)).thenReturn(BranchStatus.PhaseOne_Done);
+ assertFalse(core.doBranchDelete(global, branch));
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = BranchStatus.class,
+ names = {"PhaseOne_Failed", "PhaseOne_PrepareFailed", "STOP_RETRY"})
+ void failedOrStoppedBranchesDoNotRepeatRemoteWork(BranchStatus status) throws Exception {
+ when(branch.getStatus()).thenReturn(status);
+ assertTrue(core.doGlobalCommit(global, true));
+ assertTrue(core.doGlobalRollback(global, true));
+ verifyNoInteractions(branchCore);
+ if (status == BranchStatus.STOP_RETRY) {
+ helper.verify(() -> SessionHelper.removeBranch(global, branch, false), never());
+ } else {
+ helper.verify(() -> SessionHelper.removeBranch(global, branch, false), times(2));
+ }
+ }
+
+ @Test
+ void readOnlyXaBranchIsRemovedWithoutCommitRequest() throws Exception {
+ when(branch.getBranchType()).thenReturn(BranchType.XA);
+ when(branch.getStatus()).thenReturn(BranchStatus.PhaseOne_RDONLY);
+ assertTrue(core.doGlobalCommit(global, false));
+ verifyNoInteractions(branchCore);
+ helper.verify(() -> SessionHelper.removeBranch(global, branch, true));
+ }
+
+ @Test
+ void commitHandlesMissingTimedOutAndAsynchronousSessions() throws Exception {
+ try (MockedStatic<SessionHolder> holder = mockStatic(SessionHolder.class)) {
+ assertEquals(GlobalStatus.Finished, core.commit("missing"));
+ holder.when(() -> SessionHolder.findGlobalSession("xid")).thenReturn(global);
+ when(global.isTimeout()).thenReturn(true);
+ assertEquals(GlobalStatus.TimeoutRollbacking, core.commit("xid"));
+ verify(global, never()).close();
+ when(global.isTimeout()).thenReturn(false);
+ when(global.getStatus()).thenReturn(GlobalStatus.Begin);
+ when(global.canBeCommittedAsync()).thenReturn(true);
+ doAnswer(i -> {
+ when(global.getStatus()).thenReturn(GlobalStatus.AsyncCommitting);
+ return null;
+ })
+ .when(global)
+ .asyncCommit();
+ holder.when(() -> SessionHolder.lockAndExecute(any(), any()))
+ .thenAnswer(i -> ((GlobalSession.LockCallable<?>) i.getArgument(1)).call());
+ assertEquals(GlobalStatus.Committed, core.commit("xid"));
+ verify(global).close();
+ verify(global).asyncCommit();
+ verify(global).clean();
+ verifyNoInteractions(branchCore);
+ }
+ }
+
+ @Test
+ void successfulSynchronousCommitQueuesRemainingAsyncBranches() throws Exception {
+ try (MockedStatic<SessionHolder> holder = mockStatic(SessionHolder.class)) {
+ holder.when(() -> SessionHolder.findGlobalSession("xid")).thenReturn(global);
+ holder.when(() -> SessionHolder.lockAndExecute(any(), any()))
+ .thenAnswer(i -> ((GlobalSession.LockCallable<?>) i.getArgument(1)).call());
+ when(global.getStatus()).thenReturn(GlobalStatus.Begin);
+ when(global.canBeCommittedAsync()).thenReturn(false, true);
+ when(global.hasBranch()).thenReturn(true);
+ doReturn(true).when(core).doGlobalCommit(global, false);
+ assertEquals(GlobalStatus.Committed, core.commit("xid"));
+ verify(global).close();
+ verify(global).changeGlobalStatus(GlobalStatus.Committing);
+ verify(global).clean();
+ verify(core).doGlobalCommit(global, false);
+ verify(global).asyncCommit();
+ }
+ }
+
+ @Test
+ void unretryableBranchDeletionStopsFurtherRetries() throws Exception {
+ when(branchCore.branchDelete(global, branch)).thenReturn(BranchStatus.PhaseTwo_RollbackFailed_Unretryable);
+ for (BranchType type : Arrays.asList(BranchType.AT, BranchType.TCC, BranchType.XA)) {
+ when(branch.getBranchType()).thenReturn(type);
+ assertTrue(core.doBranchDelete(global, branch));
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/env/PortConfigUnitTest.java b/server/src/test/java/org/apache/seata/server/env/PortConfigUnitTest.java
new file mode 100644
index 0000000000..d6a2c7350a
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/env/PortConfigUnitTest.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.server.env;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class PortConfigUnitTest {
+ @TempDir
+ Path directory;
+
+ @Test
+ void explicitPropertiesAndYamlSupportValidMissingAndMalformedPorts() throws Exception {
+ String previous = System.getProperty("spring.config.location");
+ try {
+ Path file = directory.resolve("application.properties");
+ System.setProperty("spring.config.location", file.toString());
+ Files.write(file, "server.port=9090".getBytes(StandardCharsets.UTF_8));
+ assertEquals(9090, PortHelper.getPortFromConfigFile());
+ Files.write(file, "server.port=invalid".getBytes(StandardCharsets.UTF_8));
+ assertEquals(8080, PortHelper.getPortFromConfigFile());
+ Files.write(file, "other=value".getBytes(StandardCharsets.UTF_8));
+ assertEquals(8080, PortHelper.getPortFromConfigFile());
+ Path yaml = directory.resolve("application.yml");
+ System.setProperty("spring.config.location", yaml.toString());
+ Files.write(yaml, "server:\n port: 9191\n".getBytes(StandardCharsets.UTF_8));
+ assertEquals(9191, PortHelper.getPortFromConfigFile());
+ Files.write(yaml, "other: value".getBytes(StandardCharsets.UTF_8));
+ assertEquals(8080, PortHelper.getPortFromConfigFile());
+ } finally {
+ if (previous == null) {
+ System.clearProperty("spring.config.location");
+ } else {
+ System.setProperty("spring.config.location", previous);
+ }
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/filter/RaftRequestFilterUnitTest.java b/server/src/test/java/org/apache/seata/server/filter/RaftRequestFilterUnitTest.java
new file mode 100644
index 0000000000..fa245bce0b
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/filter/RaftRequestFilterUnitTest.java
@@ -0,0 +1,128 @@
+/*
+ * 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.server.filter;
+
+import io.netty.handler.codec.http.*;
+import io.netty.handler.codec.http2.DefaultHttp2Headers;
+import org.apache.seata.common.store.SessionMode;
+import org.apache.seata.core.exception.HttpRequestFilterException;
+import org.apache.seata.core.rpc.netty.http.SimpleHttp2Request;
+import org.apache.seata.core.rpc.netty.http.filter.*;
+import org.apache.seata.server.cluster.listener.ClusterChangeEvent;
+import org.apache.seata.server.cluster.raft.context.SeataClusterContext;
+import org.apache.seata.server.store.StoreConfig;
+import org.junit.jupiter.api.*;
+import org.mockito.MockedStatic;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.*;
+
+import static org.apache.seata.common.Constants.RAFT_GROUP_HEADER;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class RaftRequestFilterUnitTest {
+ private final RaftRequestFilter filter = new RaftRequestFilter();
+ private Map<String, Boolean> prevent;
+ private Map<String, Boolean> previous;
+
+ @BeforeEach
+ void open() {
+ prevent = (Map<String, Boolean>) ReflectionTestUtils.getField(RaftRequestFilter.class, "GROUP_PREVENT");
+ previous = new HashMap<>(prevent);
+ prevent.clear();
+ }
+
+ @AfterEach
+ void close() {
+ prevent.clear();
+ prevent.putAll(previous);
+ SeataClusterContext.unbindGroup();
+ }
+
+ private HttpFilterContext<?> context(Object request, Map<String, List<String>> params) {
+ HttpRequestParamWrapper wrapper = mock(HttpRequestParamWrapper.class);
+ when(wrapper.getAllParamsAsMultiMap()).thenReturn(params);
+ return new HttpFilterContext<>(request, null, true, "HTTP/1.1", () -> wrapper);
+ }
+
+ @Test
+ void nonTargetRequestsPassThroughWithoutGroupBinding() throws Exception {
+ HttpRequestFilterChain chain = mock(HttpRequestFilterChain.class);
+ for (Object request :
+ Arrays.asList(new Object(), new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/health"))) {
+ HttpFilterContext<?> context = context(request, Collections.emptyMap());
+ filter.doFilter(context, chain);
+ verify(chain).doFilter(context);
+ }
+ }
+
+ @Test
+ void queryGroupTakesPrecedenceAndBindingIsAlwaysCleared() throws Exception {
+ try (MockedStatic<StoreConfig> config = mockStatic(StoreConfig.class)) {
+ config.when(StoreConfig::getSessionMode).thenReturn(SessionMode.RAFT);
+ assertTrue(filter.shouldApply());
+ filter.onApplicationEvent(new ClusterChangeEvent(this, "query", 1, true));
+ DefaultHttpRequest request =
+ new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/vgroup/v1/add");
+ request.headers().set(RAFT_GROUP_HEADER, "header");
+ HttpFilterContext<?> context =
+ context(request, Collections.singletonMap("unit", Collections.singletonList("query")));
+ HttpRequestFilterChain chain = mock(HttpRequestFilterChain.class);
+ doAnswer(i -> {
+ assertEquals("query", SeataClusterContext.getGroup());
+ throw new HttpRequestFilterException("downstream");
+ })
+ .when(chain)
+ .doFilter(context);
+ assertThrows(HttpRequestFilterException.class, () -> filter.doFilter(context, chain));
+ assertNull(SeataClusterContext.getGroup());
+ filter.onApplicationEvent(new ClusterChangeEvent(this, "query", 2, false));
+ reset(chain);
+ assertThrows(HttpRequestFilterException.class, () -> filter.doFilter(context, chain));
+ verifyNoInteractions(chain);
+ assertNull(SeataClusterContext.getGroup());
+ config.when(StoreConfig::getSessionMode).thenReturn(SessionMode.FILE);
+ assertFalse(filter.shouldApply());
+ RaftRequestFilter.setPrevent("ignored", true);
+ assertFalse(prevent.containsKey("ignored"));
+ }
+ }
+
+ @Test
+ void http1AndHttp2HeadersPermitLeadersAndReads() throws Exception {
+ prevent.put("leader", true);
+ HttpRequestFilterChain chain = mock(HttpRequestFilterChain.class);
+ DefaultHttpRequest http1 =
+ new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/api/v1/console/commit");
+ http1.headers().set(RAFT_GROUP_HEADER, "leader");
+ HttpFilterContext<?> c1 = context(http1, Collections.emptyMap());
+ filter.doFilter(c1, chain);
+ verify(chain).doFilter(c1);
+ SimpleHttp2Request http2 = new SimpleHttp2Request(
+ HttpMethod.POST, "/vgroup/v1/add", new DefaultHttp2Headers(false).set(RAFT_GROUP_HEADER, "leader"), "");
+ HttpFilterContext<?> c2 = context(http2, Collections.emptyMap());
+ filter.doFilter(c2, chain);
+ verify(chain).doFilter(c2);
+ HttpFilterContext<?> read = context(
+ new SimpleHttp2Request(HttpMethod.GET, "/vgroup/v1/query", new DefaultHttp2Headers(false), ""),
+ Collections.emptyMap());
+ filter.doFilter(read, chain);
+ verify(chain).doFilter(read);
+ assertNull(SeataClusterContext.getGroup());
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/filter/XssRequestUnitTest.java b/server/src/test/java/org/apache/seata/server/filter/XssRequestUnitTest.java
new file mode 100644
index 0000000000..6640d5179f
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/filter/XssRequestUnitTest.java
@@ -0,0 +1,69 @@
+/*
+ * 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.server.filter;
+
+import org.apache.seata.common.ConfigurationKeys;
+import org.apache.seata.config.ConfigurationCache;
+import org.apache.seata.core.exception.HttpRequestFilterException;
+import org.apache.seata.core.rpc.netty.http.filter.*;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class XssRequestUnitTest extends BaseSpringBootTest {
+ @Test
+ void configuredKeywordsAndEventHandlersAreRejectedBeforeDispatch() throws Exception {
+ String key = ConfigurationKeys.SERVER_HTTP_FILTER_XSS_FILTER_KEYWORDS, previous = System.getProperty(key);
+ Object keywords = ReflectionTestUtils.getField(XSSHttpRequestFilter.class, "xssKeywords");
+ try {
+ System.setProperty(key, "[\"dangerous\",\"dangerous\"]");
+ ConfigurationCache.clear();
+ XSSHttpRequestFilter filter = new XSSHttpRequestFilter();
+ assertTrue(filter.shouldApply());
+ assertEquals(Integer.MIN_VALUE, filter.getOrder());
+ HttpRequestParamWrapper params = mock(HttpRequestParamWrapper.class);
+ HttpFilterContext<?> context = new HttpFilterContext<>(new Object(), null, true, "HTTP/1.1", () -> params);
+ HttpRequestFilterChain chain = mock(HttpRequestFilterChain.class);
+ for (String value : Arrays.asList("D A N G E R O U S", "ononononon", "oncustom='payload'")) {
+ when(params.getAllParamsAsMultiMap())
+ .thenReturn(Collections.singletonMap("input", Collections.singletonList(value)));
+ assertThrows(HttpRequestFilterException.class, () -> filter.doFilter(context, chain));
+ }
+ verifyNoInteractions(chain);
+ when(params.getAllParamsAsMultiMap())
+ .thenReturn(Collections.singletonMap("input", Arrays.asList(null, "normal text")));
+ filter.doFilter(context, chain);
+ verify(chain).doFilter(context);
+ System.setProperty(key, "not-json");
+ ConfigurationCache.clear();
+ assertThrows(IllegalArgumentException.class, XSSHttpRequestFilter::new);
+ } finally {
+ ReflectionTestUtils.setField(XSSHttpRequestFilter.class, "xssKeywords", keywords);
+ if (previous == null) {
+ System.clearProperty(key);
+ } else {
+ System.setProperty(key, previous);
+ }
+ ConfigurationCache.clear();
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/instance/GeneralInstanceStrategyUnitTest.java b/server/src/test/java/org/apache/seata/server/instance/GeneralInstanceStrategyUnitTest.java
new file mode 100644
index 0000000000..9238c3945e
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/instance/GeneralInstanceStrategyUnitTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.server.instance;
+
+import org.apache.seata.common.holder.ObjectHolder;
+import org.apache.seata.common.metadata.Instance;
+import org.apache.seata.common.store.SessionMode;
+import org.apache.seata.common.thread.ThreadPoolExecutorFactory;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.session.SessionHolder;
+import org.apache.seata.server.store.*;
+import org.apache.seata.spring.boot.autoconfigure.properties.registry.*;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedStatic;
+import org.springframework.boot.web.server.autoconfigure.ServerProperties;
+import org.springframework.context.ApplicationContext;
+import org.springframework.core.env.*;
+
+import java.util.*;
+import java.util.concurrent.*;
+
+import static org.apache.seata.common.Constants.OBJECT_KEY_SPRING_CONFIGURABLE_ENVIRONMENT;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class GeneralInstanceStrategyUnitTest extends BaseSpringBootTest {
+ @Test
+ void instanceMetadataAndHeartbeatLifecycle() {
+ Object old = ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_CONFIGURABLE_ENVIRONMENT);
+ ScheduledExecutorService previous = GeneralInstanceStrategy.EXECUTOR_SERVICE;
+ StandardEnvironment environment = new StandardEnvironment();
+ Map<String, Object> values = new HashMap<>();
+ values.put(org.apache.seata.common.ConfigurationKeys.META_PREFIX + "zone", "east");
+ values.put("unrelated", "ignored");
+ environment.getPropertySources().addFirst(new MapPropertySource("test", values));
+ ObjectHolder.INSTANCE.setObject(OBJECT_KEY_SPRING_CONFIGURABLE_ENVIRONMENT, environment);
+ Instance instance = mock(Instance.class);
+ ScheduledThreadPoolExecutor executor = mock(ScheduledThreadPoolExecutor.class);
+ VGroupMappingStoreManager mappings = mock(VGroupMappingStoreManager.class);
+ try (MockedStatic<Instance> instances = mockStatic(Instance.class);
+ MockedStatic<StoreConfig> config = mockStatic(StoreConfig.class);
+ MockedStatic<SessionHolder> holder = mockStatic(SessionHolder.class);
+ MockedStatic<ThreadPoolExecutorFactory> pools = mockStatic(ThreadPoolExecutorFactory.class)) {
+ instances.when(Instance::getInstance).thenReturn(instance);
+ config.when(StoreConfig::getSessionMode).thenReturn(SessionMode.FILE);
+ holder.when(SessionHolder::getRootVGroupMappingManager).thenReturn(mappings);
+ pools.when(() -> ThreadPoolExecutorFactory.newScheduledThreadPoolExecutor("scheduledExecutor", 1, true))
+ .thenReturn(executor);
+ GeneralInstanceStrategy strategy = new GeneralInstanceStrategy();
+ strategy.registryProperties = new RegistryProperties();
+ strategy.registryNamingServerProperties = new RegistryNamingServerProperties();
+ strategy.registryNamingServerProperties.setNamespace("tenant");
+ strategy.registryNamingServerProperties.setCluster("east");
+ strategy.registryNamingServerProperties.setHeartbeatPeriod(1000);
+ ServerProperties properties = new ServerProperties();
+ properties.setPort(8080);
+ strategy.applicationContext = mock(ApplicationContext.class);
+ when(strategy.applicationContext.getBean(ServerProperties.class)).thenReturn(properties);
+ strategy.postConstruct();
+ assertSame(instance, strategy.serverInstanceInit());
+ verify(instance).setNamespace("tenant");
+ verify(instance).setClusterName("east");
+ verify(instance).addMetadata("cluster-type", "default");
+ verify(instance).addMetadata("zone", "east");
+ verify(instance, never()).addMetadata(eq("unrelated"), any());
+ assertEquals(SeataInstanceStrategy.Type.GENERAL, strategy.type());
+ strategy.registryProperties.setType("file");
+ strategy.init();
+ verifyNoInteractions(executor);
+ strategy.registryProperties.setType(org.apache.seata.common.ConfigurationKeys.NAMING_SERVER);
+ strategy.init();
+ strategy.init();
+ ArgumentCaptor<Runnable> task = ArgumentCaptor.forClass(Runnable.class);
+ verify(executor).scheduleAtFixedRate(task.capture(), eq(1000L), eq(1000L), eq(TimeUnit.MILLISECONDS));
+ when(instance.getTerm()).thenReturn(1L);
+ task.getValue().run();
+ verify(mappings).notifyMapping();
+ doThrow(new IllegalStateException("offline")).when(mappings).notifyMapping();
+ assertDoesNotThrow(task.getValue()::run);
+ strategy.destroy();
+ verify(executor).shutdown();
+ } finally {
+ GeneralInstanceStrategy.EXECUTOR_SERVICE = previous;
+ ObjectHolder.INSTANCE.setObject(OBJECT_KEY_SPRING_CONFIGURABLE_ENVIRONMENT, old);
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/limit/ratelimit/RateLimitConfigurationUnitTest.java b/server/src/test/java/org/apache/seata/server/limit/ratelimit/RateLimitConfigurationUnitTest.java
new file mode 100644
index 0000000000..e9574bdb64
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/limit/ratelimit/RateLimitConfigurationUnitTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.server.limit.ratelimit;
+
+import org.apache.seata.common.ConfigurationKeys;
+import org.apache.seata.config.*;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class RateLimitConfigurationUnitTest extends BaseSpringBootTest {
+ @Test
+ void dynamicChangesPreserveLastValidNumbers() {
+ RateLimiter limiter = mock(RateLimiter.class);
+ RateLimiterHandler handler = new RateLimiterHandler(limiter);
+ change(handler, ConfigurationKeys.RATE_LIMIT_ENABLE, "true");
+ change(handler, ConfigurationKeys.RATE_LIMIT_BUCKET_TOKEN_NUM_PER_SECOND, "1");
+ change(handler, ConfigurationKeys.RATE_LIMIT_BUCKET_TOKEN_MAX_NUM, "10");
+ change(handler, ConfigurationKeys.RATE_LIMIT_BUCKET_TOKEN_INITIAL_NUM, "2");
+ change(handler, ConfigurationKeys.RATE_LIMIT_BUCKET_TOKEN_MAX_NUM, "invalid");
+ org.mockito.ArgumentCaptor<RateLimiterHandlerConfig> captured =
+ org.mockito.ArgumentCaptor.forClass(RateLimiterHandlerConfig.class);
+ verify(limiter, times(5)).reInit(captured.capture());
+ RateLimiterHandlerConfig config = captured.getValue();
+ assertTrue(config.isEnable());
+ assertEquals(1, config.getBucketTokenNumPerSecond());
+ assertEquals(10, config.getBucketTokenMaxNum());
+ assertEquals(2, config.getBucketTokenInitialNum());
+ TokenBucketLimiter bucket = new TokenBucketLimiter();
+ bucket.reInit(config);
+ assertTrue(bucket.isEnable());
+ assertTrue(bucket.canPass());
+ config.setEnable(false);
+ bucket.reInit(config);
+ assertFalse(bucket.isEnable());
+ }
+
+ @Test
+ void startupReadsEnabledAndDisabledConfigurations() {
+ Configuration config = mock(Configuration.class);
+ when(config.getBoolean(ConfigurationKeys.RATE_LIMIT_ENABLE)).thenReturn(true);
+ when(config.getConfig(ConfigurationKeys.RATE_LIMIT_BUCKET_TOKEN_NUM_PER_SECOND))
+ .thenReturn("1");
+ when(config.getConfig(ConfigurationKeys.RATE_LIMIT_BUCKET_TOKEN_MAX_NUM))
+ .thenReturn("10");
+ when(config.getConfig(ConfigurationKeys.RATE_LIMIT_BUCKET_TOKEN_INITIAL_NUM))
+ .thenReturn("0");
+ try (MockedStatic<ConfigurationFactory> factory = mockStatic(ConfigurationFactory.class)) {
+ factory.when(ConfigurationFactory::getInstance).thenReturn(config);
+ TokenBucketLimiter limiter = new TokenBucketLimiter();
+ limiter.init();
+ assertTrue(limiter.isEnable());
+ assertEquals(10, limiter.obtainConfig().getBucketTokenMaxNum());
+ assertTrue(limiter.canPass());
+ when(config.getBoolean(ConfigurationKeys.RATE_LIMIT_ENABLE)).thenReturn(false);
+ limiter.init();
+ assertFalse(limiter.isEnable());
+ }
+ }
+
+ private void change(RateLimiterHandler handler, String key, String value) {
+ ConfigurationChangeEvent event = new ConfigurationChangeEvent();
+ event.setDataId(key);
+ event.setNewValue(value);
+ handler.onChangeEvent(event);
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/metrics/MetricsSubscriberUnitTest.java b/server/src/test/java/org/apache/seata/server/metrics/MetricsSubscriberUnitTest.java
new file mode 100644
index 0000000000..3cccc256af
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/metrics/MetricsSubscriberUnitTest.java
@@ -0,0 +1,141 @@
+/*
+ * 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.server.metrics;
+
+import org.apache.seata.core.event.*;
+import org.apache.seata.metrics.*;
+import org.apache.seata.metrics.registry.Registry;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class MetricsSubscriberUnitTest {
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "CommitFailed",
+ "RollbackFailed",
+ "TimeoutRollbackFailed",
+ "CommitRetryTimeout",
+ "RollbackRetryTimeout"
+ })
+ void failuresDecreaseActiveCountAndRecordFailureLatency(String status) {
+ Registry registry = mock(Registry.class);
+ Counter counter = mock(Counter.class);
+ Summary summary = mock(Summary.class);
+ Timer timer = mock(Timer.class);
+ when(registry.getCounter(any())).thenReturn(counter);
+ when(registry.getSummary(any())).thenReturn(summary);
+ when(registry.getTimer(any())).thenReturn(timer);
+ MetricsSubscriber subscriber = new MetricsSubscriber(registry);
+ subscriber.recordGlobalTransactionEventForMetrics(
+ new GlobalTransactionEvent(1, "tc", "order", "app", "group", 10L, 40L, status, false, false));
+ verify(counter).decrease(1);
+ verify(timer).record(30, TimeUnit.MILLISECONDS);
+ verify(summary, atLeastOnce()).increase(1);
+ verify(registry)
+ .getSummary(MeterIdConstants.SUMMARY_FAILED
+ .withTag(IdConstants.APP_ID_KEY, "app")
+ .withTag(IdConstants.GROUP_KEY, "group"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"CommitRetrying", "RollbackRetrying", "TimeoutRollbackRetrying"})
+ void retriesIncludeTransactionNameWithoutCompletingActiveTransaction(String status) {
+ Registry registry = mock(Registry.class);
+ Summary summary = mock(Summary.class);
+ when(registry.getSummary(any())).thenReturn(summary);
+ new MetricsSubscriber(registry)
+ .recordGlobalTransactionEventForMetrics(
+ new GlobalTransactionEvent(1, "tc", "order", "app", "group", 10L, 40L, status, true, true));
+ verify(summary).increase(1);
+ verify(registry, never()).getCounter(any());
+ verify(registry, never()).getTimer(any());
+ }
+
+ @Test
+ void exceptionsRateLimitsAndUnknownStatusesAreHandled() {
+ Registry registry = mock(Registry.class);
+ Summary summary = mock(Summary.class);
+ when(registry.getSummary(any())).thenReturn(summary);
+ MetricsSubscriber subscriber = new MetricsSubscriber(registry);
+ subscriber.exceptionEventForMetrics(new ExceptionEvent("failure"));
+ subscriber.recordRateLimitEventForMetrics(new RateLimitEvent("trace", "token", "app", "host:8091"));
+ verify(summary, times(2)).increase(1);
+ clearInvocations(registry);
+ subscriber.recordGlobalTransactionEventForMetrics(
+ new GlobalTransactionEvent(1, "tc", "order", "app", "group", 10L, 40L, "unknown", false, false));
+ verifyNoInteractions(registry);
+ assertDoesNotThrow(() -> new MetricsSubscriber(null).recordGlobalTransactionEventForMetrics(null));
+ }
+
+ @Test
+ void retryCompletionIsCountedOnlyAfterBothPhasesFinish() {
+ Registry registry = mock(Registry.class);
+ Counter counter = mock(Counter.class);
+ Summary summary = mock(Summary.class);
+ Timer timer = mock(Timer.class);
+ when(registry.getCounter(any())).thenReturn(counter);
+ when(registry.getSummary(any())).thenReturn(summary);
+ when(registry.getTimer(any())).thenReturn(timer);
+ MetricsSubscriber subscriber = new MetricsSubscriber(registry);
+ for (String status : new String[] {"Committed", "Rollbacked"}) {
+ subscriber.recordGlobalTransactionEventForMetrics(
+ new GlobalTransactionEvent(1, "tc", "order", "app", "group", 10L, 40L, status, true, true));
+ }
+ verifyNoInteractions(counter, summary, timer);
+ for (String status : new String[] {
+ IdConstants.STATUS_VALUE_AFTER_COMMITTED_KEY, IdConstants.STATUS_VALUE_AFTER_ROLLBACKED_KEY
+ }) {
+ subscriber.recordGlobalTransactionEventForMetrics(
+ new GlobalTransactionEvent(1, "tc", "order", "app", "group", 10L, 40L, status, true, true));
+ }
+ verify(counter, times(2)).decrease(1);
+ verify(counter, times(2)).increase(1);
+ verify(summary, times(2)).increase(1);
+ verify(timer, times(2)).record(30, TimeUnit.MILLISECONDS);
+ clearInvocations(counter, summary, timer);
+ subscriber.recordGlobalTransactionEventForMetrics(new GlobalTransactionEvent(
+ 1, "tc", "order", "app", "group", 10L, 40L, "TimeoutRollbacked", false, false));
+ verify(counter).decrease(1);
+ verifyNoInteractions(summary, timer);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"Committed", "Rollbacked"})
+ void firstCompletionRecordsSuccessAndLatency(String status) {
+ Registry registry = mock(Registry.class);
+ Counter counter = mock(Counter.class);
+ Summary summary = mock(Summary.class);
+ Timer timer = mock(Timer.class);
+ when(registry.getCounter(any())).thenReturn(counter);
+ when(registry.getSummary(any())).thenReturn(summary);
+ when(registry.getTimer(any())).thenReturn(timer);
+ new MetricsSubscriber(registry)
+ .recordGlobalTransactionEventForMetrics(
+ new GlobalTransactionEvent(1, "tc", "order", "app", "group", 10L, 40L, status, false, false));
+ verify(counter).decrease(1);
+ verify(counter).increase(1);
+ verify(summary).increase(1);
+ verify(timer).record(30, TimeUnit.MILLISECONDS);
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/session/SessionHelperUnitTest.java b/server/src/test/java/org/apache/seata/server/session/SessionHelperUnitTest.java
new file mode 100644
index 0000000000..1fca35823b
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/session/SessionHelperUnitTest.java
@@ -0,0 +1,115 @@
+/*
+ * 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.server.session;
+
+import org.apache.seata.core.context.RootContext;
+import org.apache.seata.core.exception.TransactionException;
+import org.apache.seata.core.model.*;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.cluster.raft.context.SeataClusterContext;
+import org.apache.seata.server.metrics.MetricsPublisher;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.slf4j.MDC;
+
+import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class SessionHelperUnitTest extends BaseSpringBootTest {
+ @Test
+ void branchIterationPreservesResourceOrderingAndPropagatesResults() throws Exception {
+ BranchSession first = SessionHelper.newBranch(BranchType.AT, "xid", 1, "resource", "data");
+ BranchSession second = SessionHelper.newBranch(BranchType.AT, "xid", 2, "resource", "data");
+ List<BranchSession> branches = Arrays.asList(first, second);
+ List<Long> visited = Collections.synchronizedList(new ArrayList<>());
+ BranchSessionHandler handler = b -> {
+ assertEquals(String.valueOf(b.getBranchId()), MDC.get(RootContext.MDC_KEY_BRANCH_ID));
+ visited.add(b.getBranchId());
+ return null;
+ };
+ assertNull(SessionHelper.parallelForEach(branches, handler));
+ assertEquals(Arrays.asList(1L, 2L), visited);
+ assertNull(SessionHelper.singleForEach(Collections.<BranchSession>emptyList(), handler));
+ assertTrue(SessionHelper.forEach(branches, (BranchSessionHandler) b -> true));
+ assertFalse(SessionHelper.singleForEach(branches, (BranchSessionHandler) b -> false));
+ assertFalse(SessionHelper.parallelForEach(branches, (BranchSessionHandler) b -> false));
+ TransactionException failure = new TransactionException("failed");
+ assertSame(
+ failure,
+ assertThrows(
+ TransactionException.class,
+ () -> SessionHelper.parallelForEach(branches, (BranchSessionHandler) b -> {
+ throw failure;
+ })));
+ assertThrows(
+ TransactionException.class,
+ () -> SessionHelper.parallelForEach(branches, (BranchSessionHandler) b -> {
+ throw new IllegalStateException("failed");
+ }));
+ assertNull(MDC.get(RootContext.MDC_KEY_BRANCH_ID));
+ }
+
+ @Test
+ void globalIterationIsolatesFailuresAndClearsContext() {
+ GlobalSession a = new GlobalSession();
+ a.setXid("a");
+ GlobalSession b = new GlobalSession();
+ b.setXid("b");
+ AtomicInteger calls = new AtomicInteger();
+ GlobalSessionHandler handler = s -> {
+ assertEquals(s.getXid(), MDC.get(RootContext.MDC_KEY_XID));
+ calls.incrementAndGet();
+ throw new IllegalStateException("failed");
+ };
+ SessionHelper.singleForEach(Arrays.asList(a, b), handler);
+ assertEquals(2, calls.get());
+ SessionHelper.parallelForEach(Arrays.asList(a, b), handler);
+ SessionHelper.forEach(Arrays.asList(a, b), handler);
+ assertEquals(6, calls.get());
+ assertNull(MDC.get(RootContext.MDC_KEY_XID));
+ assertNull(SeataClusterContext.getGroup());
+ }
+
+ @Test
+ void branchRemovalUnlocksAllBranchesAndFinalStatusClosesSession() throws Exception {
+ GlobalSession global = mock(GlobalSession.class);
+ BranchSession branch = new BranchSession();
+ when(global.getSortedBranches()).thenReturn(Collections.singletonList(branch));
+ SessionHelper.removeAllBranch(global, false);
+ verify(global).removeAndUnlockBranch(branch);
+ when(global.getSortedBranches()).thenReturn(Collections.emptyList());
+ SessionHelper.removeAllBranch(global, false);
+ verify(global, times(1)).removeAndUnlockBranch(branch);
+ try (MockedStatic<MetricsPublisher> metrics = mockStatic(MetricsPublisher.class)) {
+ for (GlobalStatus status : Arrays.asList(
+ GlobalStatus.Committed,
+ GlobalStatus.Finished,
+ GlobalStatus.Rollbacked,
+ GlobalStatus.TimeoutRollbacked)) {
+ reset(global);
+ when(global.getStatus()).thenReturn(status);
+ SessionHelper.processEndState(global);
+ verify(global).end();
+ }
+ when(global.getStatus()).thenReturn(GlobalStatus.Begin);
+ assertThrows(TransactionException.class, () -> SessionHelper.processEndState(global));
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/session/SessionHolderRecoveryUnitTest.java b/server/src/test/java/org/apache/seata/server/session/SessionHolderRecoveryUnitTest.java
new file mode 100644
index 0000000000..4c024e8cab
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/session/SessionHolderRecoveryUnitTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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.server.session;
+
+import org.apache.seata.common.exception.ShouldNeverHappenException;
+import org.apache.seata.common.store.SessionMode;
+import org.apache.seata.core.exception.TransactionException;
+import org.apache.seata.core.model.*;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.junit.jupiter.api.*;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.mockito.MockedStatic;
+
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class SessionHolderRecoveryUnitTest extends BaseSpringBootTest {
+ @ParameterizedTest
+ @EnumSource(
+ value = GlobalStatus.class,
+ names = {"Finished", "UnKnown", "CommitFailed", "RollbackFailed", "TimeoutRollbackFailed"})
+ void terminalErrorsAreRemovedAndStoreFailuresContained(GlobalStatus status) throws Exception {
+ GlobalSession session = mock(GlobalSession.class);
+ when(session.getStatus()).thenReturn(status);
+ SessionManager manager = mock(SessionManager.class);
+ try (MockedStatic<SessionHolder> holder = mockStatic(SessionHolder.class, CALLS_REAL_METHODS)) {
+ holder.when(SessionHolder::getRootSessionManager).thenReturn(manager);
+ SessionHolder.reload(Collections.singletonList(session), SessionMode.FILE, false);
+ verify(manager).removeGlobalSession(session);
+ doThrow(new TransactionException("offline")).when(manager).removeGlobalSession(session);
+ assertDoesNotThrow(() -> SessionHolder.reload(Collections.singletonList(session), SessionMode.FILE, false));
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = GlobalStatus.class,
+ names = {"Committed", "Rollbacked", "TimeoutRollbacked"})
+ void completedTransactionsUseFinalizersAndContainFailures(GlobalStatus status) throws Exception {
+ GlobalSession session = mock(GlobalSession.class);
+ when(session.getStatus()).thenReturn(status);
+ try (MockedStatic<SessionHelper> helper = mockStatic(SessionHelper.class)) {
+ SessionHolder.reload(Collections.singletonList(session), SessionMode.FILE, false);
+ if (status == GlobalStatus.Committed) {
+ helper.verify(() -> SessionHelper.endCommitted(session, true));
+ helper.when(() -> SessionHelper.endCommitted(session, true))
+ .thenThrow(new TransactionException("failed"));
+ } else {
+ helper.verify(() -> SessionHelper.endRollbacked(session, true));
+ helper.when(() -> SessionHelper.endRollbacked(session, true))
+ .thenThrow(new TransactionException("failed"));
+ }
+ assertDoesNotThrow(() -> SessionHolder.reload(Collections.singletonList(session), SessionMode.FILE, false));
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = GlobalStatus.class,
+ names = {"Committing", "CommitRetrying", "AsyncCommitting"})
+ void raftCommitRecoveryCleansLocks(GlobalStatus status) throws Exception {
+ GlobalSession session = mock(GlobalSession.class);
+ when(session.getStatus()).thenReturn(status);
+ SessionHolder.reload(Collections.singletonList(session), SessionMode.RAFT, false);
+ verify(session).clean();
+ doThrow(new TransactionException("failed")).when(session).clean();
+ assertThrows(
+ RuntimeException.class,
+ () -> SessionHolder.reload(Collections.singletonList(session), SessionMode.RAFT, false));
+ }
+
+ @Test
+ void beginRecoveryDistinguishesNewLeaderTransactionsAndReacquiresBranchLocks() throws Exception {
+ GlobalSession session = mock(GlobalSession.class);
+ BranchSession branch = mock(BranchSession.class);
+ when(session.getStatus()).thenReturn(GlobalStatus.Begin);
+ when(session.getSortedBranches()).thenReturn(Collections.singletonList(branch));
+ when(session.getBranchSessions()).thenReturn(Collections.singletonList(branch));
+ when(session.getBeginTime()).thenReturn(0L);
+ SessionHolder.reload(Collections.singletonList(session), SessionMode.RAFT, true);
+ verify(branch).lock();
+ verify(session).changeGlobalStatus(GlobalStatus.RollbackRetrying);
+ when(session.getBeginTime()).thenReturn(Long.MAX_VALUE);
+ SessionHolder.reload(Collections.singletonList(session), SessionMode.RAFT, false);
+ verify(session).setActive(true);
+ when(session.getStatus()).thenReturn(GlobalStatus.Rollbacking);
+ SessionHolder.reload(Collections.singletonList(session), SessionMode.RAFT, true);
+ verify(branch).setLockStatus(LockStatus.Rollbacking);
+ doThrow(new TransactionException("lock failed")).when(branch).lock();
+ assertThrows(
+ ShouldNeverHappenException.class,
+ () -> SessionHolder.reload(Collections.singletonList(session), SessionMode.FILE, true));
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/session/SessionLifecycleUnitTest.java b/server/src/test/java/org/apache/seata/server/session/SessionLifecycleUnitTest.java
new file mode 100644
index 0000000000..fc503710bd
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/session/SessionLifecycleUnitTest.java
@@ -0,0 +1,114 @@
+/*
+ * 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.server.session;
+
+import org.apache.seata.core.exception.*;
+import org.apache.seata.core.model.*;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.cluster.raft.RaftServerManager;
+import org.apache.seata.server.lock.*;
+import org.apache.seata.server.store.TransactionStoreManager;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class SessionLifecycleUnitTest extends BaseSpringBootTest {
+ @Test
+ void listenersReceivePersistedLifecycleAndCanBeRemoved() throws Exception {
+ SessionManager manager = mock(SessionManager.class);
+ LockManager locks = mock(LockManager.class);
+ SessionLifecycleListener listener = mock(SessionLifecycleListener.class);
+ GlobalSession global = new GlobalSession("app", "group", "transaction", 1000);
+ BranchSession branch = mock(BranchSession.class);
+ when(branch.unlock()).thenReturn(true);
+ when(locks.releaseGlobalSessionLock(global)).thenReturn(true);
+ try (MockedStatic<SessionHolder> holder = mockStatic(SessionHolder.class);
+ MockedStatic<LockerManagerFactory> factory = mockStatic(LockerManagerFactory.class);
+ MockedStatic<RaftServerManager> raft = mockStatic(RaftServerManager.class)) {
+ holder.when(SessionHolder::getRootSessionManager).thenReturn(manager);
+ factory.when(LockerManagerFactory::getLockManager).thenReturn(locks);
+ global.addSessionLifecycleListener(listener);
+ global.begin();
+ verify(listener).onBegin(global);
+ global.addBranch(branch);
+ verify(listener).onAddBranch(global, branch);
+ assertTrue(global.getBranchSessions().contains(branch));
+ global.changeBranchStatus(branch, BranchStatus.PhaseOne_Done);
+ verify(listener).onBranchStatusChange(global, branch, BranchStatus.PhaseOne_Done);
+ global.changeGlobalStatus(GlobalStatus.Rollbacking);
+ verify(listener).onStatusChange(global, GlobalStatus.Rollbacking);
+ global.close();
+ verify(listener).onClose(global);
+ global.removeAndUnlockBranch(branch);
+ verify(listener).onRemoveBranch(global, branch);
+ assertTrue(global.getBranchSessions().isEmpty());
+ global.setStatus(GlobalStatus.Committed);
+ global.end();
+ verify(listener).onSuccessEnd(global);
+ global.setStatus(GlobalStatus.CommitFailed);
+ global.end();
+ verify(listener).onFailEnd(global);
+ global.removeSessionLifecycleListener(listener);
+ clearInvocations(listener);
+ global.begin();
+ verifyNoInteractions(listener);
+ when(locks.releaseGlobalSessionLock(global)).thenReturn(false);
+ assertThrows(TransactionException.class, global::clean);
+ when(branch.unlock()).thenReturn(false);
+ assertThrows(TransactionException.class, () -> global.unlockBranch(branch));
+ }
+ }
+
+ @Test
+ void failedPersistenceIsTranslatedToTransactionExceptions() throws Exception {
+ AbstractSessionManager manager = mock(AbstractSessionManager.class, CALLS_REAL_METHODS);
+ TransactionStoreManager store = mock(TransactionStoreManager.class);
+ manager.setTransactionStoreManager(store);
+ GlobalSession global = new GlobalSession("app", "group", "transaction", 1000);
+ BranchSession branch = new BranchSession();
+ assertThrows(GlobalTransactionException.class, () -> manager.addGlobalSession(global));
+ assertThrows(
+ GlobalTransactionException.class,
+ () -> manager.updateGlobalSessionStatus(global, GlobalStatus.Committing));
+ assertThrows(GlobalTransactionException.class, () -> manager.removeGlobalSession(global));
+ assertThrows(BranchTransactionException.class, () -> manager.addBranchSession(global, branch));
+ assertThrows(
+ BranchTransactionException.class,
+ () -> manager.updateBranchSessionStatus(branch, BranchStatus.Registered));
+ assertThrows(BranchTransactionException.class, () -> manager.removeBranchSession(global, branch));
+ manager.destroy();
+ }
+
+ @Test
+ void lazyBranchesAreLoadedOnceFromRootManager() {
+ GlobalSession lazy = new GlobalSession("app", "group", "transaction", 1000, true);
+ lazy.setXid("xid");
+ GlobalSession stored = new GlobalSession("app", "group", "transaction", 1000);
+ stored.add(new BranchSession());
+ SessionManager manager = mock(SessionManager.class);
+ when(manager.findGlobalSession("xid", true)).thenReturn(stored);
+ try (MockedStatic<SessionHolder> holder = mockStatic(SessionHolder.class)) {
+ holder.when(SessionHolder::getRootSessionManager).thenReturn(manager);
+ lazy.loadBranchs();
+ lazy.loadBranchs();
+ assertEquals(1, lazy.getBranchSessions().size());
+ verify(manager, times(1)).findGlobalSession("xid", true);
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/spring/listener/ServerEnvironmentUnitTest.java b/server/src/test/java/org/apache/seata/server/spring/listener/ServerEnvironmentUnitTest.java
new file mode 100644
index 0000000000..aabfe62001
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/spring/listener/ServerEnvironmentUnitTest.java
@@ -0,0 +1,76 @@
+/*
+ * 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.server.spring.listener;
+
+import org.apache.seata.common.holder.ObjectHolder;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.spring.boot.autoconfigure.SeataCoreEnvironmentPostProcessor;
+import org.apache.seata.spring.boot.autoconfigure.SeataServerEnvironmentPostProcessor;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.springframework.boot.context.event.*;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.core.ResolvableType;
+import org.springframework.core.env.StandardEnvironment;
+
+import java.util.*;
+
+import static org.apache.seata.common.Constants.OBJECT_KEY_SPRING_CONFIGURABLE_ENVIRONMENT;
+import static org.apache.seata.core.constants.ConfigurationKeys.SERVER_SERVICE_PORT_CAMEL;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class ServerEnvironmentUnitTest extends BaseSpringBootTest {
+ @Test
+ void commandLinePortPrecedesEnvironmentAndReadyMarksBootCompletion() {
+ Object previous = ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_CONFIGURABLE_ENVIRONMENT);
+ Map<String, String> properties = new HashMap<>();
+ for (String key :
+ Arrays.asList(SERVER_SERVICE_PORT_CAMEL, "production.deploy.output", "ENV_LOG_SYS_BOOT_COMPLETED")) {
+ properties.put(key, System.getProperty(key));
+ }
+ try (MockedStatic<SeataCoreEnvironmentPostProcessor> core =
+ mockStatic(SeataCoreEnvironmentPostProcessor.class);
+ MockedStatic<SeataServerEnvironmentPostProcessor> server =
+ mockStatic(SeataServerEnvironmentPostProcessor.class)) {
+ ServerApplicationListener listener = new ServerApplicationListener();
+ StandardEnvironment env = new StandardEnvironment();
+ ApplicationEnvironmentPreparedEvent event = mock(ApplicationEnvironmentPreparedEvent.class);
+ when(event.getEnvironment()).thenReturn(env);
+ when(event.getArgs()).thenReturn(new String[] {"-p", "9191"});
+ listener.onApplicationEvent(event);
+ assertEquals("9191", System.getProperty(SERVER_SERVICE_PORT_CAMEL));
+ assertSame(env, ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_CONFIGURABLE_ENVIRONMENT));
+ assertTrue(listener.supportsEventType(ResolvableType.forClass(ApplicationReadyEvent.class)));
+ assertFalse(listener.supportsEventType(ResolvableType.forClass(ApplicationEvent.class)));
+ assertFalse(listener.supportsEventType(ResolvableType.NONE));
+ System.setProperty("production.deploy.output", "true");
+ listener.onApplicationEvent(mock(ApplicationReadyEvent.class));
+ assertEquals("true", System.getProperty("ENV_LOG_SYS_BOOT_COMPLETED"));
+ assertDoesNotThrow(() -> listener.onApplicationEvent(mock(ApplicationEvent.class)));
+ } finally {
+ ObjectHolder.INSTANCE.setObject(OBJECT_KEY_SPRING_CONFIGURABLE_ENVIRONMENT, previous);
+ properties.forEach((key, value) -> {
+ if (value == null) {
+ System.clearProperty(key);
+ } else {
+ System.setProperty(key, value);
+ }
+ });
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/db/lock/DataBaseDistributedLockerUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/db/lock/DataBaseDistributedLockerUnitTest.java
new file mode 100644
index 0000000000..2125191897
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/db/lock/DataBaseDistributedLockerUnitTest.java
@@ -0,0 +1,183 @@
+/*
+ * 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.server.storage.db.lock;
+
+import org.apache.seata.common.exception.ShouldNeverHappenException;
+import org.apache.seata.common.loader.EnhancedServiceLoader;
+import org.apache.seata.config.*;
+import org.apache.seata.core.store.DistributedLockDO;
+import org.apache.seata.core.store.db.DataSourceProvider;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.junit.jupiter.api.*;
+import org.mockito.MockedStatic;
+
+import javax.sql.DataSource;
+import java.sql.*;
+
+import static org.apache.seata.core.constants.ConfigurationKeys.*;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class DataBaseDistributedLockerUnitTest extends BaseSpringBootTest {
+ private DataSource dataSource;
+ private Connection connection;
+ private PreparedStatement statement;
+ private ResultSet result;
+ private DataBaseDistributedLocker locker;
+ private Configuration config;
+ private MockedStatic<ConfigurationFactory> configuration;
+ private MockedStatic<EnhancedServiceLoader> loader;
+
+ @BeforeEach
+ void open() throws Exception {
+ config = mock(Configuration.class);
+ when(config.getConfig(DISTRIBUTED_LOCK_DB_TABLE)).thenReturn("distributed_lock");
+ when(config.getConfig(STORE_DB_TYPE)).thenReturn("mysql");
+ when(config.getConfig(STORE_DB_DATASOURCE_TYPE)).thenReturn("test");
+ configuration = mockStatic(ConfigurationFactory.class);
+ configuration.when(ConfigurationFactory::getInstance).thenReturn(config);
+ dataSource = mock(DataSource.class);
+ connection = mock(Connection.class);
+ statement = mock(PreparedStatement.class);
+ result = mock(ResultSet.class);
+ when(dataSource.getConnection()).thenReturn(connection);
+ when(connection.getAutoCommit()).thenReturn(true);
+ when(connection.prepareStatement(anyString())).thenReturn(statement);
+ when(statement.executeQuery()).thenReturn(result);
+ when(statement.executeUpdate()).thenReturn(1);
+ DataSourceProvider provider = mock(DataSourceProvider.class);
+ when(provider.provide()).thenReturn(dataSource);
+ // Intercept provider loading before any real loader invocation can initialize
+ // a static datasource configuration with this test's mocked configuration.
+ loader = mockStatic(EnhancedServiceLoader.class, invocation -> {
+ Object[] args = invocation.getArguments();
+ if ("load".equals(invocation.getMethod().getName())
+ && args.length > 0
+ && args[0] == DataSourceProvider.class) {
+ return provider;
+ }
+ return invocation.callRealMethod();
+ });
+ locker = new DataBaseDistributedLocker();
+ }
+
+ @AfterEach
+ void close() {
+ loader.close();
+ configuration.close();
+ }
+
+ private DistributedLockDO lock() {
+ DistributedLockDO d = new DistributedLockDO();
+ d.setLockKey("lease");
+ d.setLockValue("owner");
+ d.setExpireTime(1000L);
+ return d;
+ }
+
+ @Test
+ void acquisitionInsertsLeaseAndRestoresConnection() throws Exception {
+ DistributedLockDO d = lock();
+ long start = System.currentTimeMillis();
+ assertTrue(locker.acquireLock(d));
+ assertTrue(d.getExpireTime() >= start + 1000);
+ verify(statement, times(2)).setString(1, "lease");
+ verify(statement).setString(2, "owner");
+ verify(connection).commit();
+ verify(connection).setAutoCommit(false);
+ verify(connection).setAutoCommit(true);
+ verify(connection).close();
+ }
+
+ @Test
+ void activeLeaseBlocksButExpiredLeaseCanBeReplaced() throws Exception {
+ when(result.next()).thenReturn(true);
+ when(result.getLong(anyString())).thenReturn(Long.MAX_VALUE);
+ when(result.getString(anyString())).thenReturn("other");
+ assertFalse(locker.acquireLock(lock()));
+ verify(statement, never()).executeUpdate();
+ when(result.getLong(anyString())).thenReturn(0L);
+ assertTrue(locker.acquireLock(lock()));
+ verify(statement).executeUpdate();
+ verify(connection, times(2)).commit();
+ }
+
+ @Test
+ void releasePreservesOtherOwnerAndExpiresOwnedLease() throws Exception {
+ when(result.next()).thenReturn(true);
+ when(result.getLong(anyString())).thenReturn(Long.MAX_VALUE);
+ when(result.getString(anyString())).thenReturn("other");
+ assertTrue(locker.releaseLock(lock()));
+ verify(statement, never()).executeUpdate();
+ when(result.getString(anyString())).thenReturn("owner");
+ DistributedLockDO d = lock();
+ assertTrue(locker.releaseLock(d));
+ assertEquals(0L, d.getExpireTime());
+ assertEquals(" ", d.getLockValue());
+ verify(statement).setLong(2, 0L);
+ verify(connection, times(2)).close();
+ }
+
+ @Test
+ void missingLeaseCannotBeReleased() throws Exception {
+ assertThrows(ShouldNeverHappenException.class, () -> locker.releaseLock(lock()));
+ verify(connection).close();
+ }
+
+ @Test
+ void sqlFailuresRollbackAndCloseIncludingLockWaitTimeout() throws Exception {
+ when(statement.executeQuery()).thenThrow(new SQLException("try restarting transaction", "", 1205));
+ assertFalse(locker.acquireLock(lock()));
+ assertFalse(locker.releaseLock(lock()));
+ verify(connection, times(2)).rollback();
+ verify(connection, times(2)).close();
+ doThrow(new SQLException("broken")).when(statement).executeQuery();
+ doThrow(new SQLException("rollback failed")).when(connection).rollback();
+ assertFalse(locker.acquireLock(lock()));
+ assertFalse(locker.releaseLock(lock()));
+ when(dataSource.getConnection()).thenThrow(new SQLException("offline"));
+ assertFalse(locker.acquireLock(lock()));
+ assertFalse(locker.releaseLock(lock()));
+ }
+
+ @Test
+ void zeroAffectedRowsAndZeroDurationArePreserved() throws Exception {
+ DistributedLockDO d = lock();
+ d.setExpireTime(0L);
+ when(statement.executeUpdate()).thenReturn(0);
+ assertFalse(locker.insertDistribute(connection, d));
+ assertFalse(locker.updateDistributedLock(connection, d));
+ assertEquals(0L, d.getExpireTime());
+ }
+
+ @Test
+ void missingTableTemporarilyDemotesLocking() throws Exception {
+ when(config.getConfig(DISTRIBUTED_LOCK_DB_TABLE)).thenReturn("");
+ DataBaseDistributedLocker demoted = new DataBaseDistributedLocker();
+ assertTrue(demoted.acquireLock(lock()));
+ assertTrue(demoted.releaseLock(lock()));
+ verifyNoInteractions(connection);
+ org.mockito.ArgumentCaptor<ConfigurationChangeListener> listener =
+ org.mockito.ArgumentCaptor.forClass(ConfigurationChangeListener.class);
+ verify(config).addConfigListener(eq(DISTRIBUTED_LOCK_DB_TABLE), listener.capture());
+ ConfigurationChangeEvent event = new ConfigurationChangeEvent();
+ event.setNewValue("distributed_lock");
+ listener.getValue().onChangeEvent(event);
+ assertTrue(demoted.acquireLock(lock()));
+ verify(connection).commit();
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/db/lock/LockStoreDaoUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/db/lock/LockStoreDaoUnitTest.java
new file mode 100644
index 0000000000..de58e53109
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/db/lock/LockStoreDaoUnitTest.java
@@ -0,0 +1,172 @@
+/*
+ * 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.server.storage.db.lock;
+
+import org.apache.seata.common.exception.*;
+import org.apache.seata.config.ConfigurationFactory;
+import org.apache.seata.core.constants.ConfigurationKeys;
+import org.apache.seata.core.lock.RowLock;
+import org.apache.seata.core.model.LockStatus;
+import org.apache.seata.core.store.*;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.junit.jupiter.api.*;
+
+import javax.sql.DataSource;
+import java.sql.*;
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class LockStoreDaoUnitTest extends BaseSpringBootTest {
+ private DataSource source;
+ private Connection connection;
+ private PreparedStatement statement;
+ private LockStoreDataBaseDAO dao;
+ private LockDO lock;
+
+ @BeforeEach
+ void open() throws Exception {
+ ConfigurationFactory.getInstance().putConfig(ConfigurationKeys.STORE_DB_TYPE, "mysql");
+ ConfigurationFactory.getInstance().putConfig(ConfigurationKeys.LOCK_DB_TABLE, "lock_table");
+ source = mock(DataSource.class);
+ connection = mock(Connection.class);
+ statement = mock(PreparedStatement.class);
+ when(source.getConnection()).thenReturn(connection);
+ when(connection.prepareStatement(anyString())).thenReturn(statement);
+ when(statement.executeUpdate()).thenReturn(1);
+ when(statement.executeQuery()).thenReturn(mock(ResultSet.class));
+ dao = new LockStoreDataBaseDAO(source);
+ lock = new LockDO();
+ lock.setXid("host:8091:42");
+ lock.setTransactionId(42L);
+ lock.setBranchId(7L);
+ lock.setResourceId("db");
+ lock.setTableName("orders");
+ lock.setPk("1");
+ lock.setRowKey("db^^^orders^^^1");
+ }
+
+ @Test
+ void unlockOperationsBindTheirOwnScopeAndCloseResources() throws Exception {
+ assertTrue(dao.unLock(lock));
+ assertTrue(dao.unLock(lock.getXid()));
+ assertTrue(dao.unLock(7L));
+ dao.updateLockStatus(lock.getXid(), LockStatus.Rollbacking);
+ verify(statement, times(2)).setString(1, lock.getXid());
+ verify(statement).setLong(1, 7L);
+ verify(statement).setString(2, lock.getRowKey());
+ verify(statement).setInt(1, LockStatus.Rollbacking.getCode());
+ verify(connection, times(4)).close();
+ verify(statement, times(4)).close();
+ }
+
+ @Test
+ void singleInsertDistinguishesConflictFromDatabaseFailure() throws Exception {
+ assertTrue(dao.doAcquireLock(connection, lock));
+ verify(statement).setString(7, lock.getRowKey());
+ verify(statement).setInt(8, LockStatus.Locked.getCode());
+ when(statement.executeUpdate()).thenReturn(0);
+ assertFalse(dao.doAcquireLock(connection, lock));
+ doThrow(new SQLIntegrityConstraintViolationException("duplicate"))
+ .when(statement)
+ .executeUpdate();
+ assertFalse(dao.doAcquireLock(connection, lock));
+ doThrow(new SQLException("offline")).when(statement).executeUpdate();
+ assertThrows(StoreException.class, () -> dao.doAcquireLock(connection, lock));
+ }
+
+ @Test
+ void batchExceptionsDistinguishConstraintViolations() throws Exception {
+ BatchUpdateException withCause = new BatchUpdateException();
+ withCause.initCause(new SQLIntegrityConstraintViolationException("duplicate"));
+ when(statement.executeBatch()).thenThrow(withCause);
+ assertFalse(dao.doAcquireLocks(connection, Collections.singletonList(lock)));
+ BatchUpdateException withNext = new BatchUpdateException();
+ withNext.setNextException(new SQLIntegrityConstraintViolationException("duplicate"));
+ doThrow(withNext).when(statement).executeBatch();
+ assertFalse(dao.doAcquireLocks(connection, Collections.singletonList(lock)));
+ BatchUpdateException other = new BatchUpdateException();
+ other.setNextException(new SQLException("offline"));
+ doThrow(other).when(statement).executeBatch();
+ assertThrows(SQLException.class, () -> dao.doAcquireLocks(connection, Collections.singletonList(lock)));
+ doThrow(new BatchUpdateException()).when(statement).executeBatch();
+ assertThrows(BatchUpdateException.class, () -> dao.doAcquireLocks(connection, Collections.singletonList(lock)));
+ }
+
+ @Test
+ void queryAndUnlockFailuresPropagateWithResourcesClosed() throws Exception {
+ when(connection.prepareStatement(anyString())).thenThrow(new SQLException("offline"));
+ assertThrows(StoreException.class, () -> dao.unLock(lock));
+ assertThrows(StoreException.class, () -> dao.unLock(lock.getXid()));
+ assertThrows(StoreException.class, () -> dao.unLock(7L));
+ assertThrows(DataAccessException.class, () -> dao.updateLockStatus(lock.getXid(), LockStatus.Locked));
+ assertThrows(DataAccessException.class, () -> dao.isLockable(Collections.singletonList(lock)));
+ verify(connection, times(5)).close();
+ }
+
+ @Test
+ void databaseLockerDelegatesAndDistinguishesStoreFailures() {
+ LockStore store = mock(LockStore.class);
+ DataBaseLocker locker = new DataBaseLocker();
+ locker.setLockStore(store);
+ RowLock row = new RowLock();
+ row.setXid(lock.getXid());
+ row.setTransactionId(42L);
+ row.setBranchId(7L);
+ row.setResourceId("db");
+ row.setTableName("orders");
+ row.setPk("1");
+ row.setRowKey(lock.getRowKey());
+ List<RowLock> rows = Collections.singletonList(row);
+ when(store.acquireLock(anyList(), eq(true), eq(false))).thenReturn(true);
+ when(store.unLock(anyList())).thenReturn(true);
+ when(store.unLock(7L)).thenReturn(true);
+ when(store.unLock("xid")).thenReturn(true);
+ when(store.isLockable(anyList())).thenReturn(true);
+ assertTrue(locker.acquireLock(rows));
+ assertTrue(locker.releaseLock(rows));
+ assertTrue(locker.releaseLock("xid", 7L));
+ assertTrue(locker.releaseLock("xid"));
+ assertTrue(locker.isLockable(rows));
+ locker.updateLockStatus("xid", LockStatus.Rollbacking);
+ verify(store).updateLockStatus("xid", LockStatus.Rollbacking);
+ RuntimeException failure = new IllegalStateException("offline");
+ doThrow(failure).when(store).acquireLock(anyList(), anyBoolean(), anyBoolean());
+ doThrow(failure).when(store).unLock(anyList());
+ doThrow(failure).when(store).unLock(anyLong());
+ doThrow(failure).when(store).unLock(anyString());
+ doThrow(failure).when(store).isLockable(anyList());
+ assertFalse(locker.acquireLock(rows));
+ assertFalse(locker.releaseLock(rows));
+ assertFalse(locker.releaseLock("xid", 7L));
+ assertFalse(locker.releaseLock("xid"));
+ assertFalse(locker.isLockable(rows));
+ StoreException se = new StoreException("store");
+ doThrow(se).when(store).acquireLock(anyList(), anyBoolean(), anyBoolean());
+ doThrow(se).when(store).unLock(anyList());
+ doThrow(se).when(store).unLock(anyLong());
+ doThrow(se).when(store).unLock(anyString());
+ assertSame(se, assertThrows(StoreException.class, () -> locker.acquireLock(rows)));
+ assertThrows(StoreException.class, () -> locker.releaseLock(rows));
+ assertThrows(StoreException.class, () -> locker.releaseLock("xid", 7L));
+ assertThrows(StoreException.class, () -> locker.releaseLock("xid"));
+ assertTrue(locker.acquireLock(Collections.emptyList()));
+ assertTrue(locker.releaseLock(Collections.emptyList()));
+ assertTrue(locker.isLockable(Collections.emptyList()));
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/db/store/DatabaseManagersUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/db/store/DatabaseManagersUnitTest.java
new file mode 100644
index 0000000000..0a2f008b07
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/db/store/DatabaseManagersUnitTest.java
@@ -0,0 +1,115 @@
+/*
+ * 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.server.storage.db.store;
+
+import org.apache.seata.common.metadata.Instance;
+import org.apache.seata.core.model.*;
+import org.apache.seata.core.store.*;
+import org.apache.seata.discovery.registry.*;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.session.*;
+import org.apache.seata.server.store.TransactionStoreManager.LogOperation;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class DatabaseManagersUnitTest extends BaseSpringBootTest {
+ @Test
+ void sessionQueriesPreserveLazyLoadingAndExpectedStatus() {
+ DataBaseTransactionStoreManager manager = mock(DataBaseTransactionStoreManager.class, CALLS_REAL_METHODS);
+ LogStore store = mock(LogStore.class);
+ manager.setLogStore(store);
+ manager.setLogQueryLimit(10);
+ GlobalTransactionDO global = new GlobalTransactionDO();
+ global.setXid("host:8091:42");
+ global.setTransactionId(42L);
+ global.setStatus(1);
+ global.setTimeout(1000);
+ global.setBeginTime(1L);
+ when(store.queryGlobalTransactionDO(42L)).thenReturn(global);
+ when(store.queryGlobalTransactionDO(global.getXid())).thenReturn(global);
+ when(store.queryGlobalTransactionDO(any(int[].class), eq(10))).thenReturn(Collections.singletonList(global));
+ BranchTransactionDO branch = new BranchTransactionDO();
+ branch.setBranchId(7L);
+ branch.setXid(global.getXid());
+ branch.setTransactionId(42L);
+ branch.setStatus(1);
+ branch.setBranchType("AT");
+ when(store.queryBranchTransactionDO(global.getXid())).thenReturn(Collections.singletonList(branch));
+ when(store.queryBranchTransactionDO(anyList())).thenReturn(Collections.singletonList(branch));
+ assertEquals(1, manager.readSession(42L).getBranchSessions().size());
+ SessionCondition c = new SessionCondition();
+ c.setTransactionId(42L);
+ assertEquals(1, manager.readSession(c).size());
+ assertTrue(manager.readSession(new GlobalStatus[] {GlobalStatus.Begin}, false)
+ .get(0)
+ .isLazyLoadBranch());
+ assertEquals(
+ 1,
+ manager.readSortByTimeoutBeginSessions(true)
+ .get(0)
+ .getBranchSessions()
+ .size());
+ assertNull(manager.readSession(new SessionCondition()));
+ assertNull(manager.readSession(new SessionCondition("missing")));
+ GlobalSession session = new GlobalSession("app", "group", "order", 1000);
+ session.setExpectedStatusFromCurrent();
+ session.setStatus(GlobalStatus.Committing);
+ when(store.updateGlobalTransactionDO(any(), eq(1))).thenReturn(true);
+ assertTrue(manager.writeSession(LogOperation.GLOBAL_UPDATE, session));
+ verify(store).updateGlobalTransactionDO(argThat(g -> g.getStatus() == 2), eq(1));
+ }
+
+ @Test
+ void mappingsFilterClusterAndNotifyRegistryWithCurrentValues() throws Exception {
+ DataBaseVGroupMappingStoreManager manager = mock(DataBaseVGroupMappingStoreManager.class, CALLS_REAL_METHODS);
+ VGroupMappingDataBaseDAO dao = mock(VGroupMappingDataBaseDAO.class);
+ manager.vGroupMappingDataBaseDAO = dao;
+ MappingDO matching = new MappingDO();
+ matching.setVGroup("payments");
+ matching.setCluster("east");
+ MappingDO other = new MappingDO();
+ other.setVGroup("orders");
+ other.setCluster("west");
+ when(dao.queryMappingDO()).thenReturn(Arrays.asList(matching, other, new MappingDO()));
+ when(dao.insertMappingDO(matching)).thenReturn(true);
+ when(dao.deleteMappingDOByVGroup("payments")).thenReturn(true);
+ Instance instance = mock(Instance.class);
+ when(instance.getClusterName()).thenReturn("east");
+ RegistryService registry = mock(RegistryService.class);
+ try (MockedStatic<Instance> instances = mockStatic(Instance.class);
+ MockedStatic<MultiRegistryFactory> registries = mockStatic(MultiRegistryFactory.class)) {
+ instances.when(Instance::getInstance).thenReturn(instance);
+ registries.when(MultiRegistryFactory::getInstances).thenReturn(Collections.singletonList(registry));
+ assertTrue(manager.addVGroup(matching));
+ assertTrue(manager.removeVGroup("payments"));
+ assertEquals(
+ Collections.singleton("payments"), manager.loadVGroups().keySet());
+ manager.notifyMapping();
+ verify(instance).addMetadata(eq("vGroup"), argThat(m -> ((Map<?, ?>) m).containsKey("payments")));
+ verify(registry).register(any(java.net.InetSocketAddress.class));
+ doThrow(new IllegalStateException("offline"))
+ .when(registry)
+ .register(any(java.net.InetSocketAddress.class));
+ assertThrows(RuntimeException.class, manager::notifyMapping);
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/db/store/LogStoreDaoUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/db/store/LogStoreDaoUnitTest.java
new file mode 100644
index 0000000000..cd2d78d77c
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/db/store/LogStoreDaoUnitTest.java
@@ -0,0 +1,120 @@
+/*
+ * 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.server.storage.db.store;
+
+import org.apache.seata.common.exception.*;
+import org.apache.seata.core.store.*;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.junit.jupiter.api.*;
+
+import javax.sql.DataSource;
+import java.sql.*;
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class LogStoreDaoUnitTest extends BaseSpringBootTest {
+ private DataSource source;
+ private Connection connection;
+ private PreparedStatement statement;
+ private ResultSet result;
+ private LogStoreDataBaseDAO dao;
+
+ @BeforeEach
+ void open() throws Exception {
+ source = mock(DataSource.class);
+ connection = mock(Connection.class);
+ statement = mock(PreparedStatement.class);
+ result = mock(ResultSet.class);
+ when(source.getConnection()).thenReturn(connection);
+ when(connection.prepareStatement(anyString())).thenReturn(statement);
+ when(statement.executeQuery()).thenReturn(result);
+ DatabaseMetaData metadata = mock(DatabaseMetaData.class);
+ when(connection.getMetaData()).thenReturn(metadata);
+ when(metadata.getTables(any(), any(), any(), any())).thenReturn(mock(ResultSet.class));
+ dao = new LogStoreDataBaseDAO(source);
+ dao.setDbType("mysql");
+ clearInvocations(source, connection, statement, result);
+ }
+
+ @Test
+ void maxIdScansGlobalAndBranchTablesWithinBounds() throws Exception {
+ when(result.next()).thenReturn(true, false, true, false);
+ when(result.getLong(1)).thenReturn(12L, 20L);
+ assertEquals(20L, dao.getCurrentMaxSessionId(100, 1));
+ verify(statement, times(2)).setLong(1, 100);
+ verify(statement, times(2)).setLong(2, 1);
+ verify(connection, times(2)).close();
+ verify(result, times(2)).close();
+ }
+
+ @Test
+ void missingQueriesReturnEmptyResultsAndSqlFailuresPropagate() throws Exception {
+ assertNull(dao.queryGlobalTransactionDO("xid"));
+ assertNull(dao.queryGlobalTransactionDO(1L));
+ assertTrue(dao.queryGlobalTransactionDO(new int[] {1}, 10).isEmpty());
+ assertTrue(dao.queryBranchTransactionDO("xid").isEmpty());
+ when(source.getConnection()).thenThrow(new SQLException("offline"));
+ assertThrows(DataAccessException.class, () -> dao.queryGlobalTransactionDO("xid"));
+ assertThrows(DataAccessException.class, () -> dao.queryGlobalTransactionDO(1L));
+ assertThrows(DataAccessException.class, () -> dao.queryGlobalTransactionDO(new int[] {1}, 10));
+ assertThrows(DataAccessException.class, () -> dao.queryBranchTransactionDO("xid"));
+ assertThrows(DataAccessException.class, () -> dao.queryBranchTransactionDO(Collections.singletonList("xid")));
+ assertThrows(DataAccessException.class, () -> dao.getCurrentMaxSessionId(100, 1));
+ GlobalTransactionDO global = new GlobalTransactionDO();
+ global.setStatus(1);
+ global.setTransactionId(1L);
+ global.setTimeout(1000);
+ global.setBeginTime(1L);
+ BranchTransactionDO branch = new BranchTransactionDO();
+ branch.setBranchId(7L);
+ branch.setStatus(1);
+ branch.setTransactionId(1L);
+ assertThrows(StoreException.class, () -> dao.insertGlobalTransactionDO(global));
+ assertThrows(StoreException.class, () -> dao.updateGlobalTransactionDO(global));
+ assertThrows(StoreException.class, () -> dao.updateGlobalTransactionDO(global, 1));
+ assertThrows(StoreException.class, () -> dao.deleteGlobalTransactionDO(global));
+ assertThrows(StoreException.class, () -> dao.insertBranchTransactionDO(branch));
+ assertThrows(StoreException.class, () -> dao.updateBranchTransactionDO(branch));
+ assertThrows(StoreException.class, () -> dao.deleteBranchTransactionDO(branch));
+ }
+
+ @Test
+ void schemasFollowDatabaseDialectAndClosePostgresResources() throws Exception {
+ dao.setDbType("h2");
+ assertNull(org.springframework.test.util.ReflectionTestUtils.invokeMethod(dao, "getSchema", connection));
+ dao.setDbType("sqlserver");
+ when(connection.getSchema()).thenReturn("dbo");
+ assertEquals(
+ "dbo", org.springframework.test.util.ReflectionTestUtils.invokeMethod(dao, "getSchema", connection));
+ dao.setDbType("postgresql");
+ when(result.next()).thenReturn(true);
+ when(result.getString(1)).thenReturn("transactions");
+ assertEquals(
+ "transactions",
+ org.springframework.test.util.ReflectionTestUtils.invokeMethod(dao, "getSchema", connection));
+ verify(connection).prepareStatement("select current_schema");
+ verify(statement).close();
+ verify(result).close();
+ when(statement.executeQuery()).thenThrow(new SQLException("schema unavailable"));
+ assertThrows(
+ StoreException.class,
+ () -> org.springframework.test.util.ReflectionTestUtils.invokeMethod(dao, "getSchema", connection));
+ verify(statement, times(2)).close();
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/db/store/VGroupMappingDaoUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/db/store/VGroupMappingDaoUnitTest.java
new file mode 100644
index 0000000000..9c60d3fb56
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/db/store/VGroupMappingDaoUnitTest.java
@@ -0,0 +1,95 @@
+/*
+ * 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.server.storage.db.store;
+
+import org.apache.seata.common.exception.SeataRuntimeException;
+import org.apache.seata.core.store.MappingDO;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.junit.jupiter.api.*;
+
+import javax.sql.DataSource;
+import java.sql.*;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class VGroupMappingDaoUnitTest extends BaseSpringBootTest {
+ private DataSource source;
+ private Connection connection;
+ private PreparedStatement statement;
+ private ResultSet result;
+ private VGroupMappingDataBaseDAO dao;
+
+ @BeforeEach
+ void open() throws Exception {
+ source = mock(DataSource.class);
+ connection = mock(Connection.class);
+ statement = mock(PreparedStatement.class);
+ result = mock(ResultSet.class);
+ when(source.getConnection()).thenReturn(connection);
+ when(connection.prepareStatement(anyString())).thenReturn(statement);
+ when(statement.executeQuery()).thenReturn(result);
+ when(statement.executeUpdate()).thenReturn(1);
+ dao = new VGroupMappingDataBaseDAO(source);
+ }
+
+ @Test
+ void insertBindsNamespaceAndClusterAndClosesResources() throws Exception {
+ MappingDO mapping = new MappingDO();
+ mapping.setVGroup("payments");
+ mapping.setNamespace("tenant");
+ mapping.setCluster("east");
+ assertTrue(dao.insertMappingDO(mapping));
+ verify(statement).setString(1, "payments");
+ verify(statement).setString(2, "tenant");
+ verify(statement).setString(3, "east");
+ verify(connection).setAutoCommit(true);
+ verify(statement).close();
+ verify(connection).close();
+ }
+
+ @Test
+ void queryMapsRowsAndDeletesUseGroupPredicate() throws Exception {
+ when(result.next()).thenReturn(true, false);
+ when(result.getString("namespace")).thenReturn("tenant");
+ when(result.getString("cluster")).thenReturn("east");
+ when(result.getString("vGroup")).thenReturn("payments");
+ List<MappingDO> mappings = dao.queryMappingDO();
+ assertEquals(1, mappings.size());
+ assertEquals("payments", mappings.get(0).getVGroup());
+ assertEquals("tenant", mappings.get(0).getNamespace());
+ assertEquals("east", mappings.get(0).getCluster());
+ verify(result).close();
+ assertTrue(dao.clearMappingDOByVGroup("payments"));
+ assertTrue(dao.deleteMappingDOByVGroup("payments"));
+ verify(statement, times(2)).setString(1, "payments");
+ when(statement.executeUpdate()).thenReturn(0);
+ assertFalse(dao.clearMappingDOByVGroup("missing"));
+ assertFalse(dao.deleteMappingDOByVGroup("missing"));
+ }
+
+ @Test
+ void sqlFailuresPropagateAndStillCloseConnection() throws Exception {
+ when(connection.prepareStatement(anyString())).thenThrow(new SQLException("offline"));
+ assertThrows(SeataRuntimeException.class, () -> dao.insertMappingDO(new MappingDO()));
+ assertThrows(SeataRuntimeException.class, () -> dao.clearMappingDOByVGroup("g"));
+ assertThrows(RuntimeException.class, () -> dao.deleteMappingDOByVGroup("g"));
+ assertThrows(SeataRuntimeException.class, dao::queryMappingDO);
+ verify(connection, times(4)).close();
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/file/session/FileSessionReplayUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/file/session/FileSessionReplayUnitTest.java
new file mode 100644
index 0000000000..b64af2a38d
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/file/session/FileSessionReplayUnitTest.java
@@ -0,0 +1,134 @@
+/*
+ * 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.server.storage.file.session;
+
+import org.apache.seata.core.model.*;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.session.*;
+import org.apache.seata.server.storage.file.*;
+import org.apache.seata.server.storage.file.store.FileTransactionStoreManager;
+import org.apache.seata.server.store.SessionStorable;
+import org.apache.seata.server.store.TransactionStoreManager.LogOperation;
+import org.junit.jupiter.api.Test;
+
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class FileSessionReplayUnitTest extends BaseSpringBootTest {
+ private GlobalSession global(long id, GlobalStatus status) {
+ GlobalSession g = new GlobalSession("app", "group", "test", 1000);
+ g.setXid("host:8091:" + id);
+ g.setTransactionId(id);
+ g.setStatus(status);
+ return g;
+ }
+
+ private BranchSession branch(long tid, long bid, BranchStatus status) {
+ BranchSession b = new BranchSession();
+ b.setXid("host:8091:" + tid);
+ b.setTransactionId(tid);
+ b.setBranchId(bid);
+ b.setStatus(status);
+ return b;
+ }
+
+ private TransactionWriteStore event(SessionStorable s, LogOperation op) {
+ return new TransactionWriteStore(s, op);
+ }
+
+ private FileSessionManager replay(List<TransactionWriteStore> history, List<TransactionWriteStore> current) {
+ FileTransactionStoreManager store = mock(FileTransactionStoreManager.class);
+ when(store.hasRemaining(true)).thenReturn(true, false);
+ when(store.hasRemaining(false)).thenReturn(true, false);
+ when(store.readWriteStore(anyInt(), eq(true))).thenReturn(history);
+ when(store.readWriteStore(anyInt(), eq(false))).thenReturn(current);
+ FileSessionManager manager = new FileSessionManager("replay");
+ manager.setTransactionStoreManager(store);
+ manager.reload();
+ return manager;
+ }
+
+ @Test
+ void outOfOrderBranchesAreReconciledAfterBothFiles() {
+ BranchSession before = branch(1, 7, BranchStatus.Registered);
+ FileSessionManager manager = replay(
+ Arrays.asList(
+ event(before, LogOperation.BRANCH_ADD),
+ event(branch(2, 8, BranchStatus.Registered), LogOperation.BRANCH_ADD)),
+ Arrays.asList(event(global(1, GlobalStatus.Begin), LogOperation.GLOBAL_ADD)));
+ assertEquals(
+ Collections.singleton("host:8091:1"), manager.getSessionMap().keySet());
+ assertEquals(
+ 1, manager.findGlobalSession("host:8091:1").getBranchSessions().size());
+ assertEquals(
+ BranchStatus.Registered,
+ manager.findGlobalSession("host:8091:1").getBranch(7).getStatus());
+ }
+
+ @Test
+ void terminalAndRemovedGlobalsCannotBeResurrected() {
+ FileSessionManager manager = replay(
+ Arrays.asList(
+ event(global(1, GlobalStatus.Begin), LogOperation.GLOBAL_ADD),
+ event(branch(1, 7, BranchStatus.Registered), LogOperation.BRANCH_ADD)),
+ Arrays.asList(
+ event(global(1, GlobalStatus.Committed), LogOperation.GLOBAL_UPDATE),
+ event(global(1, GlobalStatus.Begin), LogOperation.GLOBAL_ADD),
+ event(branch(1, 8, BranchStatus.Registered), LogOperation.BRANCH_ADD),
+ event(branch(1, 7, BranchStatus.Registered), LogOperation.BRANCH_REMOVE),
+ event(global(1, GlobalStatus.Begin), LogOperation.GLOBAL_REMOVE),
+ event(global(2, GlobalStatus.Begin), LogOperation.GLOBAL_REMOVE),
+ event(global(2, GlobalStatus.Begin), LogOperation.GLOBAL_REMOVE),
+ event(global(3, GlobalStatus.Rollbacked), LogOperation.GLOBAL_ADD)));
+ assertTrue(manager.getSessionMap().isEmpty());
+ }
+
+ @Test
+ void replayAppliesStatusUpdatesAndBranchRemovals() {
+ FileSessionManager manager = replay(
+ Arrays.asList(
+ event(global(1, GlobalStatus.Begin), LogOperation.GLOBAL_ADD),
+ event(branch(1, 7, BranchStatus.Registered), LogOperation.BRANCH_ADD)),
+ Arrays.asList(
+ event(global(1, GlobalStatus.Committing), LogOperation.GLOBAL_UPDATE),
+ event(branch(1, 7, BranchStatus.PhaseOne_Done), LogOperation.BRANCH_UPDATE),
+ event(branch(1, 7, BranchStatus.Registered), LogOperation.BRANCH_REMOVE),
+ event(branch(1, 99, BranchStatus.Registered), LogOperation.BRANCH_REMOVE),
+ event(branch(2, 8, BranchStatus.Registered), LogOperation.BRANCH_REMOVE)));
+ GlobalSession restored = manager.findGlobalSession("host:8091:1");
+ assertEquals(GlobalStatus.Committing, restored.getStatus());
+ assertTrue(restored.getBranchSessions().isEmpty());
+ Map<String, GlobalSession> replacement = new HashMap<>();
+ manager.setSessionMap(replacement);
+ assertSame(replacement, manager.getSessionMap());
+ }
+
+ @Test
+ void bufferedBranchAttachesToLaterGlobalAndRemovedBufferIsDiscarded() {
+ FileSessionManager manager = replay(
+ Arrays.asList(
+ event(branch(1, 7, BranchStatus.Registered), LogOperation.BRANCH_ADD),
+ event(branch(2, 8, BranchStatus.Registered), LogOperation.BRANCH_ADD)),
+ Arrays.asList(
+ event(global(1, GlobalStatus.Begin), LogOperation.GLOBAL_ADD),
+ event(global(2, GlobalStatus.Finished), LogOperation.GLOBAL_ADD)));
+ assertNotNull(manager.findGlobalSession("host:8091:1").getBranch(7));
+ assertNull(manager.findGlobalSession("host:8091:2"));
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/file/store/FileMappingFailureUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/file/store/FileMappingFailureUnitTest.java
new file mode 100644
index 0000000000..ac350b1dda
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/file/store/FileMappingFailureUnitTest.java
@@ -0,0 +1,49 @@
+/*
+ * 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.server.storage.file.store;
+
+import org.apache.seata.core.store.MappingDO;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class FileMappingFailureUnitTest extends BaseSpringBootTest {
+ @TempDir
+ Path directory;
+
+ @Test
+ void invalidStorageAndCorruptMappingAreReported() throws Exception {
+ Path file = Files.write(directory.resolve("file"), new byte[] {1});
+ FileVGroupMappingStoreManager manager = new FileVGroupMappingStoreManager(file.toString());
+ MappingDO mapping = new MappingDO();
+ mapping.setVGroup("orders");
+ mapping.setUnit("unit");
+ assertFalse(manager.addVGroup(mapping));
+ assertFalse(manager.removeVGroup("orders"));
+ RuntimeException failure = assertThrows(RuntimeException.class, manager::loadVGroups);
+ assertEquals("mapping relationship load failed", failure.getMessage());
+ FileVGroupMappingStoreManager corrupted = new FileVGroupMappingStoreManager(directory.toString());
+ Files.write(
+ directory.resolve(FileVGroupMappingStoreManager.ROOT_MAPPING_MANAGER_NAME),
+ "invalid-json".getBytes(java.nio.charset.StandardCharsets.UTF_8));
+ assertThrows(RuntimeException.class, corrupted::loadVGroups);
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/file/store/FileStoreFailureUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/file/store/FileStoreFailureUnitTest.java
new file mode 100644
index 0000000000..a7f32b13cd
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/file/store/FileStoreFailureUnitTest.java
@@ -0,0 +1,139 @@
+/*
+ * 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.server.storage.file.store;
+
+import org.apache.seata.common.exception.StoreException;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.session.*;
+import org.apache.seata.server.store.TransactionStoreManager.LogOperation;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.io.*;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.*;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class FileStoreFailureUnitTest extends BaseSpringBootTest {
+ @TempDir
+ Path directory;
+
+ @Test
+ void rotationPreservesSessionsAndCreatesNewCurrentFile() throws Exception {
+ SessionManager sessions = mock(SessionManager.class);
+ GlobalSession global = GlobalSession.createGlobalSession("app", "group", "order", 1000);
+ when(sessions.findGlobalSessions(any())).thenReturn(Collections.singletonList(global));
+ Path file = directory.resolve("nested/session");
+ FileTransactionStoreManager store = new FileTransactionStoreManager(file.toString(), sessions);
+ try {
+ assertTrue(store.writeSession(LogOperation.GLOBAL_ADD, global));
+ assertTrue((Boolean) ReflectionTestUtils.invokeMethod(store, "saveHistory"));
+ assertTrue(Files.exists(Paths.get(file + ".1")));
+ assertEquals(
+ global.getXid(),
+ ((GlobalSession) store.readWriteStore(10, true).get(0).getSessionRequest()).getXid());
+ assertFalse(store.hasRemaining(true));
+ assertFalse(store.hasRemaining(false));
+ assertThrows(StoreException.class, () -> store.readSession("xid"));
+ assertThrows(StoreException.class, () -> store.readSession(new SessionCondition()));
+ assertFalse(store.writeSession(LogOperation.GLOBAL_ADD, null));
+ } finally {
+ store.shutdown();
+ }
+ }
+
+ @Test
+ void incompleteFramesStopRecoveryWithoutInventingSessions() throws Exception {
+ Path file = directory.resolve("session");
+ FileTransactionStoreManager store =
+ new FileTransactionStoreManager(file.toString(), mock(SessionManager.class));
+ try {
+ Files.write(file, new byte[] {0, 0});
+ assertTrue(store.readWriteStore(10, false).isEmpty());
+ ReflectionTestUtils.setField(store, "recoverCurrOffset", 0L);
+ Files.write(file, ByteBuffer.allocate(5).putInt(20).put((byte) 1).array());
+ assertTrue(store.readWriteStore(10, false).isEmpty());
+ ReflectionTestUtils.setField(store, "recoverCurrOffset", 0L);
+ Files.write(file, ByteBuffer.allocate(5).putInt(1).put((byte) 127).array());
+ assertTrue(store.readWriteStore(10, false).isEmpty());
+ assertNull(store.readWriteStore(10, true));
+ } finally {
+ store.shutdown();
+ }
+ }
+
+ @Test
+ void failedWritesAreRetriedAndReturnFalse() throws Exception {
+ FileTransactionStoreManager store =
+ new FileTransactionStoreManager(directory.resolve("session").toString(), mock(SessionManager.class));
+ FileChannel original = (FileChannel) ReflectionTestUtils.getField(store, "currFileChannel");
+ FileChannel failing = mock(FileChannel.class);
+ when(failing.write(any(ByteBuffer.class))).thenThrow(new IOException("disk full"));
+ try {
+ ReflectionTestUtils.setField(store, "currFileChannel", failing);
+ assertFalse(store.writeSession(
+ LogOperation.GLOBAL_ADD, GlobalSession.createGlobalSession("app", "group", "order", 1000)));
+ verify(failing, times(5)).write(any(ByteBuffer.class));
+ } finally {
+ ReflectionTestUtils.setField(store, "currFileChannel", original);
+ store.shutdown();
+ }
+ }
+
+ @Test
+ void workerDrainsFlushAndCloseRequestsOnStop() throws Exception {
+ FileTransactionStoreManager store = mock(FileTransactionStoreManager.class, CALLS_REAL_METHODS);
+ ReflectionTestUtils.setField(store, "stopping", true);
+ FileChannel channel = mock(FileChannel.class);
+ RandomAccessFile file = mock(RandomAccessFile.class);
+ AtomicLong writes =
+ (AtomicLong) ReflectionTestUtils.getField(FileTransactionStoreManager.class, "FILE_TRX_NUM");
+ AtomicLong flushes =
+ (AtomicLong) ReflectionTestUtils.getField(FileTransactionStoreManager.class, "FILE_FLUSH_NUM");
+ long oldWrites = writes.get(), oldFlushes = flushes.get();
+ try {
+ writes.set(10);
+ flushes.set(0);
+ FileTransactionStoreManager.WriteDataFileRunnable worker = store.new WriteDataFileRunnable();
+ FileTransactionStoreManager.SyncFlushRequest sync = store.new SyncFlushRequest(10, channel);
+ worker.putRequest(sync);
+ worker.putRequest(store.new AsyncFlushRequest(10, channel));
+ FileTransactionStoreManager.CloseFileRequest close =
+ new FileTransactionStoreManager.CloseFileRequest(channel, file);
+ worker.putRequest(close);
+ worker.run();
+ assertEquals(10, flushes.get());
+ verify(channel, atLeastOnce()).force(false);
+ verify(file).close();
+ sync.waitForFlush(1);
+ close.waitForClose(1);
+ doThrow(new IOException("force failed")).when(channel).force(false);
+ writes.set(11);
+ worker.putRequest(store.new SyncFlushRequest(11, channel));
+ assertDoesNotThrow(worker::run);
+ } finally {
+ writes.set(oldWrites);
+ flushes.set(oldFlushes);
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/redis/JedisPoolUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/redis/JedisPoolUnitTest.java
new file mode 100644
index 0000000000..3496f88637
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/redis/JedisPoolUnitTest.java
@@ -0,0 +1,142 @@
+/*
+ * 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.server.storage.redis;
+
+import org.apache.seata.common.exception.RedisException;
+import org.apache.seata.config.*;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.junit.jupiter.api.*;
+import org.mockito.MockedConstruction;
+import org.springframework.test.util.ReflectionTestUtils;
+import redis.clients.jedis.*;
+import redis.clients.jedis.util.Pool;
+
+import java.util.*;
+
+import static org.apache.seata.common.ConfigurationKeys.*;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class JedisPoolUnitTest extends BaseSpringBootTest {
+ private Object previous;
+ private final Map<String, String> properties = new HashMap<>();
+
+ @BeforeEach
+ void open() {
+ previous = ReflectionTestUtils.getField(JedisPooledFactory.class, "jedisPool");
+ ReflectionTestUtils.setField(JedisPooledFactory.class, "jedisPool", null);
+ }
+
+ @AfterEach
+ void close() {
+ ReflectionTestUtils.setField(JedisPooledFactory.class, "jedisPool", previous);
+ properties.forEach((key, value) -> {
+ if (value == null) {
+ System.clearProperty(key);
+ } else {
+ System.setProperty(key, value);
+ }
+ });
+ properties.clear();
+ ConfigurationCache.clear();
+ }
+
+ private void config(String key, String value) {
+ if (!properties.containsKey(key)) {
+ properties.put(key, System.getProperty(key));
+ }
+ System.setProperty(key, value);
+ ConfigurationCache.clear();
+ }
+
+ @Test
+ void suppliedPoolIsCachedAndConnectionsComeFromIt() {
+ Pool<Jedis> pool = mock(Pool.class);
+ Jedis jedis = mock(Jedis.class);
+ when(pool.getResource()).thenReturn(jedis);
+ assertSame(pool, JedisPooledFactory.getJedisPoolInstance(pool));
+ assertSame(pool, JedisPooledFactory.getJedisPoolInstance(mock(Pool.class)));
+ assertSame(jedis, JedisPooledFactory.getJedisInstance());
+ }
+
+ @Test
+ void singleModePassesHostPortPasswordAndPoolBounds() {
+ config(STORE_REDIS_MODE, REDIS_SINGLE_MODE);
+ config(STORE_REDIS_SINGLE_HOST, "redis.example");
+ config(STORE_REDIS_SINGLE_PORT, "1234");
+ config(STORE_REDIS_PASSWORD, "secret");
+ config(STORE_PUBLIC_KEY, "");
+ try (MockedConstruction<JedisPool> pools = mockConstruction(JedisPool.class, (mock, context) -> {
+ assertEquals("redis.example", context.arguments().get(1));
+ assertEquals(1234, context.arguments().get(2));
+ assertEquals("secret", context.arguments().get(4));
+ assertTrue(((JedisPoolConfig) context.arguments().get(0)).getMaxTotal() > 0);
+ })) {
+ assertSame(
+ JedisPooledFactory.getJedisPoolInstance(),
+ pools.constructed().get(0));
+ }
+ }
+
+ @Test
+ void singleModeUsesLegacyAddressAndNoPassword() {
+ config(STORE_REDIS_MODE, REDIS_SINGLE_MODE);
+ config(STORE_REDIS_SINGLE_HOST, "");
+ config(STORE_REDIS_SINGLE_PORT, "0");
+ config(STORE_REDIS_HOST, "legacy");
+ config(STORE_REDIS_PORT, "6380");
+ config(STORE_REDIS_PASSWORD, "");
+ try (MockedConstruction<JedisPool> pools = mockConstruction(JedisPool.class, (mock, context) -> {
+ assertEquals("legacy", context.arguments().get(1));
+ assertEquals(6380, context.arguments().get(2));
+ assertNull(context.arguments().get(4));
+ })) {
+ assertSame(
+ JedisPooledFactory.getJedisPoolInstance(),
+ pools.constructed().get(0));
+ }
+ }
+
+ @Test
+ void sentinelModeParsesHostsAndRejectsMissingMaster() {
+ config(STORE_REDIS_MODE, REDIS_SENTINEL_MODE);
+ config(STORE_REDIS_SENTINEL_MASTERNAME, "");
+ config(STORE_REDIS_PASSWORD, "");
+ assertThrows(RedisException.class, JedisPooledFactory::getJedisPoolInstance);
+ config(STORE_REDIS_SENTINEL_MASTERNAME, "primary");
+ config(STORE_REDIS_SENTINEL_HOST, "s1:26379,s2:26379");
+ config(STORE_REDIS_SENTINEL_PASSWORD, "");
+ try (MockedConstruction<JedisSentinelPool> pools =
+ mockConstruction(JedisSentinelPool.class, (mock, context) -> {
+ assertEquals("primary", context.arguments().get(0));
+ assertEquals(
+ new HashSet<>(Arrays.asList("s1:26379", "s2:26379")),
+ context.arguments().get(1));
+ })) {
+ assertSame(
+ JedisPooledFactory.getJedisPoolInstance(),
+ pools.constructed().get(0));
+ }
+ }
+
+ @Test
+ void invalidModeIsRejectedBeforeOpeningPool() {
+ config(STORE_REDIS_MODE, "invalid");
+ config(STORE_REDIS_PASSWORD, "");
+ assertThrows(RedisException.class, JedisPooledFactory::getJedisPoolInstance);
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/redis/RedisFactoriesUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/redis/RedisFactoriesUnitTest.java
new file mode 100644
index 0000000000..662daf69c7
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/redis/RedisFactoriesUnitTest.java
@@ -0,0 +1,103 @@
+/*
+ * 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.server.storage.redis;
+
+import org.apache.seata.common.ConfigurationKeys;
+import org.apache.seata.config.ConfigurationCache;
+import org.apache.seata.core.lock.Locker;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.lock.AbstractLockManager;
+import org.apache.seata.server.session.*;
+import org.apache.seata.server.storage.db.lock.DataBaseLockManager;
+import org.apache.seata.server.storage.redis.lock.*;
+import org.apache.seata.server.storage.redis.store.*;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.Arrays;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class RedisFactoriesUnitTest extends BaseSpringBootTest {
+ @Test
+ void modeSelectsAndCachesPipelineOrLuaImplementations() {
+ Object oldLocker = ReflectionTestUtils.getField(RedisLockerFactory.class, "locker"),
+ oldStore = ReflectionTestUtils.getField(RedisTransactionStoreManagerFactory.class, "instance");
+ String oldMode = System.getProperty(ConfigurationKeys.STORE_REDIS_TYPE);
+ try (MockedConstruction<RedisLocker> lockers = mockConstruction(RedisLocker.class);
+ MockedConstruction<RedisLuaLocker> luaLockers = mockConstruction(RedisLuaLocker.class);
+ MockedConstruction<RedisTransactionStoreManager> stores =
+ mockConstruction(RedisTransactionStoreManager.class);
+ MockedConstruction<RedisLuaTransactionStoreManager> luaStores =
+ mockConstruction(RedisLuaTransactionStoreManager.class)) {
+ for (String mode : Arrays.asList("pipeline", "lua")) {
+ ReflectionTestUtils.setField(RedisLockerFactory.class, "locker", null);
+ ReflectionTestUtils.setField(RedisTransactionStoreManagerFactory.class, "instance", null);
+ System.setProperty(ConfigurationKeys.STORE_REDIS_TYPE, mode);
+ ConfigurationCache.clear();
+ Locker locker = RedisLockerFactory.getLocker();
+ RedisTransactionStoreManager store = RedisTransactionStoreManagerFactory.getInstance();
+ assertSame(locker, RedisLockerFactory.getLocker());
+ assertSame(store, RedisTransactionStoreManagerFactory.getInstance());
+ if ("pipeline".equals(mode)) {
+ assertSame(lockers.constructed().get(0), locker);
+ assertSame(stores.constructed().get(0), store);
+ } else {
+ assertSame(luaLockers.constructed().get(0), locker);
+ assertSame(luaStores.constructed().get(0), store);
+ }
+ RedisLockManager manager = new RedisLockManager();
+ manager.init();
+ assertSame(locker, manager.getLocker(new BranchSession()));
+ }
+ } finally {
+ ReflectionTestUtils.setField(RedisLockerFactory.class, "locker", oldLocker);
+ ReflectionTestUtils.setField(RedisTransactionStoreManagerFactory.class, "instance", oldStore);
+ if (oldMode == null) {
+ System.clearProperty(ConfigurationKeys.STORE_REDIS_TYPE);
+ } else {
+ System.setProperty(ConfigurationKeys.STORE_REDIS_TYPE, oldMode);
+ }
+ ConfigurationCache.clear();
+ }
+ }
+
+ @Test
+ void lockManagersReleaseOnlyRequestedOwnerAndContainBackendFailure() throws Exception {
+ for (AbstractLockManager manager : Arrays.asList(new RedisLockManager(), new DataBaseLockManager())) {
+ Locker locker = mock(Locker.class);
+ ReflectionTestUtils.setField(manager, "locker", locker);
+ BranchSession branch = new BranchSession();
+ branch.setXid("xid");
+ branch.setBranchId(7L);
+ GlobalSession global = new GlobalSession();
+ global.setXid("xid");
+ when(locker.releaseLock("xid", 7L)).thenReturn(true);
+ when(locker.releaseLock("xid")).thenReturn(true);
+ assertTrue(manager.releaseLock(branch));
+ assertTrue(manager.releaseGlobalSessionLock(global));
+ verify(locker).releaseLock("xid", 7L);
+ verify(locker).releaseLock("xid");
+ doThrow(new IllegalStateException("offline")).when(locker).releaseLock("xid", 7L);
+ doThrow(new IllegalStateException("offline")).when(locker).releaseLock("xid");
+ assertFalse(manager.releaseLock(branch));
+ assertFalse(manager.releaseGlobalSessionLock(global));
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/redis/lock/RedisLockerUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/redis/lock/RedisLockerUnitTest.java
new file mode 100644
index 0000000000..44a814ad2b
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/redis/lock/RedisLockerUnitTest.java
@@ -0,0 +1,156 @@
+/*
+ * 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.server.storage.redis.lock;
+
+import org.apache.seata.common.exception.StoreException;
+import org.apache.seata.core.lock.RowLock;
+import org.apache.seata.core.model.LockStatus;
+import org.apache.seata.core.store.DistributedLockDO;
+import org.apache.seata.server.storage.redis.JedisPooledFactory;
+import org.junit.jupiter.api.*;
+import org.mockito.MockedStatic;
+import redis.clients.jedis.*;
+import redis.clients.jedis.params.SetParams;
+
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class RedisLockerUnitTest extends org.apache.seata.server.BaseSpringBootTest {
+ private Jedis jedis;
+ private Pipeline pipeline;
+ private MockedStatic<JedisPooledFactory> factory;
+ private final RedisLocker locker = new RedisLocker();
+
+ @BeforeEach
+ void open() {
+ jedis = mock(Jedis.class);
+ pipeline = mock(Pipeline.class);
+ factory = mockStatic(JedisPooledFactory.class);
+ factory.when(JedisPooledFactory::getJedisInstance).thenReturn(jedis);
+ when(jedis.pipelined()).thenReturn(pipeline);
+ }
+
+ @AfterEach
+ void close() {
+ factory.close();
+ }
+
+ private RowLock row(String pk) {
+ RowLock r = new RowLock();
+ r.setXid("host:8091:42");
+ r.setTransactionId(42L);
+ r.setBranchId(7L);
+ r.setResourceId("db");
+ r.setTableName("orders");
+ r.setPk(pk);
+ r.setRowKey("db^^^orders^^^" + pk);
+ return r;
+ }
+
+ @Test
+ void emptyLocksAndReentrantOwnership() {
+ assertTrue(locker.acquireLock(Collections.emptyList()));
+ assertTrue(locker.isLockable(Collections.emptyList()));
+ assertTrue(locker.releaseLock("host:8091:42", null));
+ factory.verifyNoInteractions();
+ when(pipeline.syncAndReturnAll()).thenReturn(Collections.singletonList("host:8091:42"));
+ assertTrue(locker.acquireLock(Collections.singletonList(row("1"))));
+ assertTrue(locker.isLockable(Collections.singletonList(row("1"))));
+ verify(pipeline, never()).hsetnx(anyString(), anyString(), anyString());
+ }
+
+ @Test
+ void conflictingOwnershipAndRollbackFailFast() {
+ when(pipeline.syncAndReturnAll()).thenReturn(Collections.singletonList("other:8091:1"));
+ assertFalse(locker.acquireLock(Collections.singletonList(row("1"))));
+ assertFalse(locker.isLockable(Collections.singletonList(row("1"))));
+ when(pipeline.syncAndReturnAll())
+ .thenReturn(Arrays.asList("other:8091:1", String.valueOf(LockStatus.Rollbacking.getCode())));
+ assertThrows(StoreException.class, () -> locker.acquireLock(Collections.singletonList(row("1")), false, false));
+ verify(jedis, times(3)).close();
+ }
+
+ @Test
+ void deduplicatesAndIndexesSuccessfulAcquisition() {
+ when(pipeline.syncAndReturnAll())
+ .thenReturn(Collections.singletonList(null), Arrays.asList(1, 1, 1, 1, 1, 1, 1));
+ RowLock r = row("1");
+ assertTrue(locker.acquireLock(Arrays.asList(r, r)));
+ verify(pipeline, times(1)).hsetnx(locker.buildLockKey(r.getRowKey()), "xid", r.getXid());
+ verify(jedis).hset(locker.buildXidLockKey(r.getXid()), "7", locker.buildLockKey(r.getRowKey()));
+ }
+
+ @Test
+ void acquisitionRaceRemovesOnlyNewlyAcquiredRows() {
+ when(pipeline.syncAndReturnAll())
+ .thenReturn(Arrays.asList(null, null), Arrays.asList(1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0));
+ assertFalse(locker.acquireLock(Arrays.asList(row("1"), row("2"))));
+ verify(jedis).del(any(String[].class));
+ verify(jedis, never()).hset(anyString(), anyString(), anyString());
+ }
+
+ @Test
+ void releaseAndStatusUpdateTraverseAllBranchRows() {
+ String xid = "host:8091:42", key = locker.buildXidLockKey(xid);
+ Map<String, String> branches = new LinkedHashMap<>();
+ branches.put("7", "row1;row2");
+ branches.put("8", "row3");
+ branches.put("9", "");
+ when(jedis.hgetAll(key)).thenReturn(branches);
+ locker.updateLockStatus(xid, LockStatus.Rollbacking);
+ for (String row : Arrays.asList("row1", "row2", "row3")) {
+ verify(pipeline).hset(row, "status", String.valueOf(LockStatus.Rollbacking.getCode()));
+ }
+ assertTrue(locker.releaseLock(xid));
+ verify(pipeline).del(key);
+ verify(pipeline).del(new String[] {"row1", "row2"});
+ verify(pipeline).del("row3");
+ when(jedis.hget(key, "7")).thenReturn("row1");
+ assertTrue(locker.releaseLock(xid, 7L));
+ verify(pipeline).hdel(key, "7");
+ when(jedis.hgetAll(key)).thenReturn(Collections.emptyMap());
+ assertTrue(locker.releaseLock(xid));
+ locker.updateLockStatus(xid, LockStatus.Locked);
+ }
+
+ @Test
+ void distributedLockHonorsOwnershipAndConnectionFailures() {
+ DistributedLockDO lock = new DistributedLockDO();
+ lock.setLockKey("lease");
+ lock.setLockValue("owner");
+ lock.setExpireTime(1000L);
+ RedisDistributedLocker distributed = new RedisDistributedLocker();
+ when(jedis.set(eq("lease"), eq("owner"), any(SetParams.class))).thenReturn("OK", null);
+ assertTrue(distributed.acquireLock(lock));
+ assertFalse(distributed.acquireLock(lock));
+ when(jedis.get("lease")).thenReturn("other");
+ assertTrue(distributed.releaseLock(lock));
+ verify(jedis).unwatch();
+ verify(jedis, never()).multi();
+ Transaction transaction = mock(Transaction.class);
+ when(jedis.multi()).thenReturn(transaction);
+ when(jedis.get("lease")).thenReturn("owner");
+ assertTrue(distributed.releaseLock(lock));
+ verify(transaction).del("lease");
+ verify(transaction).exec();
+ factory.when(JedisPooledFactory::getJedisInstance).thenThrow(new IllegalStateException("offline"));
+ assertFalse(distributed.acquireLock(lock));
+ assertFalse(distributed.releaseLock(lock));
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/redis/lock/RedisLuaLockerUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/redis/lock/RedisLuaLockerUnitTest.java
new file mode 100644
index 0000000000..46b205e738
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/redis/lock/RedisLuaLockerUnitTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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.server.storage.redis.lock;
+
+import org.apache.seata.common.exception.StoreException;
+import org.apache.seata.core.lock.RowLock;
+import org.apache.seata.core.model.LockStatus;
+import org.apache.seata.server.storage.redis.JedisPooledFactory;
+import org.junit.jupiter.api.*;
+import org.mockito.MockedStatic;
+import org.springframework.test.util.ReflectionTestUtils;
+import redis.clients.jedis.*;
+
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class RedisLuaLockerUnitTest extends org.apache.seata.server.BaseSpringBootTest {
+ private Jedis jedis;
+ private MockedStatic<JedisPooledFactory> factory;
+ private RedisLuaLocker locker;
+ private Map<String, String> scripts;
+ private Map<String, String> previous;
+
+ @BeforeEach
+ void open() {
+ scripts = (Map<String, String>) ReflectionTestUtils.getField(RedisLuaLocker.class, "LOCK_SHA_MAP");
+ previous = new HashMap<>(scripts);
+ scripts.clear();
+ jedis = mock(Jedis.class);
+ when(jedis.scriptLoad(anyString())).thenReturn("sha");
+ factory = mockStatic(JedisPooledFactory.class);
+ factory.when(JedisPooledFactory::getJedisInstance).thenReturn(jedis);
+ locker = new RedisLuaLocker();
+ }
+
+ @AfterEach
+ void close() {
+ scripts.clear();
+ scripts.putAll(previous);
+ factory.close();
+ }
+
+ private List<RowLock> rows() {
+ RowLock r = new RowLock();
+ r.setXid("host:8091:42");
+ r.setTransactionId(42L);
+ r.setBranchId(7L);
+ r.setResourceId("db");
+ r.setTableName("orders");
+ r.setPk("1");
+ r.setRowKey("db^^^orders^^^1");
+ return Arrays.asList(r, r);
+ }
+
+ @Test
+ void acquireDeduplicatesRowsAndChecksReturnedOwner() {
+ when(jedis.evalsha(anyString(), anyList(), anyList()))
+ .thenReturn("{\"success\":true,\"data\":\"host:8091:42\"}");
+ assertTrue(locker.acquireLock(rows()));
+ verify(jedis)
+ .evalsha(
+ eq("sha"),
+ argThat(k -> k.size() == 3 && k.get(2).equals("7")),
+ argThat(a -> a.get(0).equals("1") && a.get(2).equals("host:8091:42")));
+ when(jedis.evalsha(anyString(), anyList(), anyList()))
+ .thenReturn("{\"success\":false,\"status\":\"AnotherHoldIng\",\"data\":\"other\"}");
+ assertFalse(locker.acquireLock(rows()));
+ when(jedis.evalsha(anyString(), anyList(), anyList()))
+ .thenReturn("{\"success\":false,\"status\":\"AnotherRollbackIng\"}");
+ assertThrows(StoreException.class, () -> locker.acquireLock(rows()));
+ }
+
+ @Test
+ void lockabilityReleaseAndStatusUseCorrectScriptArguments() {
+ assertTrue(locker.acquireLock(Collections.emptyList()));
+ assertTrue(locker.isLockable(Collections.emptyList()));
+ when(jedis.evalsha(anyString(), anyList(), anyList())).thenReturn("true", "false");
+ assertTrue(locker.isLockable(rows()));
+ assertFalse(locker.isLockable(rows()));
+ String xid = "host:8091:42", key = locker.buildXidLockKey(xid);
+ assertTrue(locker.releaseLock(xid));
+ assertTrue(locker.releaseLock(xid, 7L));
+ assertTrue(locker.releaseLock(xid, null));
+ verify(jedis).evalsha("sha", Arrays.asList(key, "7"), Collections.emptyList());
+ locker.updateLockStatus(xid, LockStatus.Rollbacking);
+ verify(jedis)
+ .evalsha(
+ "sha",
+ Arrays.asList(key, "status"),
+ Collections.singletonList(String.valueOf(LockStatus.Rollbacking.getCode())));
+ }
+
+ @Test
+ void unavailableScriptsFallBackToPipeline() {
+ scripts.clear();
+ when(jedis.scriptLoad(anyString())).thenThrow(new UnsupportedOperationException("no scripting"));
+ locker = new RedisLuaLocker();
+ assertTrue(scripts.isEmpty());
+ Pipeline pipeline = mock(Pipeline.class);
+ when(jedis.pipelined()).thenReturn(pipeline);
+ when(pipeline.syncAndReturnAll()).thenReturn(Collections.singletonList("host:8091:42"));
+ assertTrue(locker.acquireLock(rows()));
+ assertTrue(locker.isLockable(rows()));
+ assertTrue(locker.releaseLock("host:8091:42"));
+ assertTrue(locker.releaseLock("host:8091:42", 7L));
+ locker.updateLockStatus("host:8091:42", LockStatus.Locked);
+ verify(jedis, never()).evalsha(anyString(), anyList(), anyList());
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/redis/session/RedisSessionManagerUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/redis/session/RedisSessionManagerUnitTest.java
new file mode 100644
index 0000000000..4128f377bb
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/redis/session/RedisSessionManagerUnitTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.server.storage.redis.session;
+
+import org.apache.seata.common.exception.StoreException;
+import org.apache.seata.core.model.*;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.session.*;
+import org.apache.seata.server.storage.redis.store.*;
+import org.apache.seata.server.store.TransactionStoreManager.LogOperation;
+import org.junit.jupiter.api.*;
+import org.mockito.MockedStatic;
+
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class RedisSessionManagerUnitTest extends BaseSpringBootTest {
+ @Test
+ void writesPropagateOperationAndRejectFailedPersistence() throws Exception {
+ RedisTransactionStoreManager store = mock(RedisTransactionStoreManager.class);
+ try (MockedStatic<RedisTransactionStoreManagerFactory> factory =
+ mockStatic(RedisTransactionStoreManagerFactory.class)) {
+ factory.when(RedisTransactionStoreManagerFactory::getInstance).thenReturn(store);
+ RedisSessionManager manager = new RedisSessionManager();
+ manager.init();
+ GlobalSession global = new GlobalSession();
+ BranchSession branch = new BranchSession();
+ when(store.writeSession(any(), any())).thenReturn(true);
+ manager.addGlobalSession(global);
+ manager.updateGlobalSessionStatus(global, GlobalStatus.Committing);
+ manager.removeGlobalSession(global);
+ manager.addBranchSession(global, branch);
+ manager.updateBranchSessionStatus(branch, BranchStatus.Registered);
+ manager.removeBranchSession(global, branch);
+ assertEquals(GlobalStatus.Committing, global.getStatus());
+ for (LogOperation op :
+ new LogOperation[] {LogOperation.GLOBAL_ADD, LogOperation.GLOBAL_UPDATE, LogOperation.GLOBAL_REMOVE
+ }) {
+ verify(store).writeSession(op, global);
+ }
+ for (LogOperation op :
+ new LogOperation[] {LogOperation.BRANCH_ADD, LogOperation.BRANCH_UPDATE, LogOperation.BRANCH_REMOVE
+ }) {
+ verify(store).writeSession(op, branch);
+ }
+ when(store.writeSession(any(), any())).thenReturn(false);
+ assertThrows(StoreException.class, () -> manager.addGlobalSession(global));
+ assertThrows(
+ StoreException.class, () -> manager.updateGlobalSessionStatus(global, GlobalStatus.Committing));
+ assertThrows(StoreException.class, () -> manager.removeGlobalSession(global));
+ assertThrows(StoreException.class, () -> manager.addBranchSession(global, branch));
+ assertThrows(
+ StoreException.class, () -> manager.updateBranchSessionStatus(branch, BranchStatus.Registered));
+ assertThrows(StoreException.class, () -> manager.removeBranchSession(global, branch));
+ when(store.readSession("xid", true)).thenReturn(global);
+ assertSame(global, manager.findGlobalSession("xid"));
+ List<GlobalSession> sessions = Collections.singletonList(global);
+ when(store.readSession(any(SessionCondition.class))).thenReturn(sessions);
+ assertSame(sessions, manager.allSessions());
+ assertSame(sessions, manager.findGlobalSessions(new SessionCondition()));
+ assertEquals("executed", manager.lockAndExecute(global, () -> "executed"));
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/redis/store/RedisLuaStoreUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/redis/store/RedisLuaStoreUnitTest.java
new file mode 100644
index 0000000000..695168f51e
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/redis/store/RedisLuaStoreUnitTest.java
@@ -0,0 +1,148 @@
+/*
+ * 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.server.storage.redis.store;
+
+import org.apache.seata.common.exception.RedisException;
+import org.apache.seata.core.model.GlobalStatus;
+import org.apache.seata.core.store.BranchTransactionDO;
+import org.apache.seata.core.store.GlobalTransactionDO;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.storage.redis.JedisPooledFactory;
+import org.junit.jupiter.api.*;
+import org.mockito.MockedStatic;
+import org.springframework.test.util.ReflectionTestUtils;
+import redis.clients.jedis.*;
+import redis.clients.jedis.exceptions.JedisNoScriptException;
+
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class RedisLuaStoreUnitTest extends BaseSpringBootTest {
+ private Jedis jedis;
+ private MockedStatic<JedisPooledFactory> factory;
+ private RedisLuaTransactionStoreManager store;
+ private Map<String, String> scripts;
+ private Map<String, String> previous;
+
+ @BeforeEach
+ void open() {
+ scripts = (Map<String, String>)
+ ReflectionTestUtils.getField(RedisLuaTransactionStoreManager.class, "LOCK_SHA_MAP");
+ previous = new HashMap<>(scripts);
+ scripts.clear();
+ jedis = mock(Jedis.class);
+ when(jedis.scriptLoad(anyString())).thenReturn("sha");
+ factory = mockStatic(JedisPooledFactory.class);
+ factory.when(JedisPooledFactory::getJedisInstance).thenReturn(jedis);
+ store = new RedisLuaTransactionStoreManager();
+ }
+
+ @AfterEach
+ void close() {
+ scripts.clear();
+ scripts.putAll(previous);
+ factory.close();
+ }
+
+ private GlobalTransactionDO global() {
+ GlobalTransactionDO g = new GlobalTransactionDO();
+ g.setXid("host:8091:42");
+ g.setTransactionId(42L);
+ g.setStatus(GlobalStatus.Committing.getCode());
+ g.setBeginTime(100L);
+ g.setTimeout(1000);
+ return g;
+ }
+
+ private BranchTransactionDO branch() {
+ BranchTransactionDO b = new BranchTransactionDO();
+ b.setBranchId(7L);
+ b.setTransactionId(42L);
+ b.setXid("host:8091:42");
+ b.setStatus(1);
+ b.setApplicationData("payload");
+ return b;
+ }
+
+ @Test
+ void writesCarryTransactionKeysAndTimeoutIndex() {
+ assertTrue(store.insertGlobalTransactionDO(global()));
+ assertTrue(store.insertBranchTransactionDO(branch()));
+ verify(jedis)
+ .evalsha(
+ eq("sha"),
+ argThat(k -> k.contains("SEATA_GLOBAL_42") && k.contains("SEATA_BEGIN_TRANSACTIONS")),
+ argThat(a -> a.contains("1100") && a.get(0).equals("global")));
+ verify(jedis).evalsha(eq("sha"), argThat(k -> k.contains("SEATA_BRANCH_7")), argThat(a -> a.get(0)
+ .equals("branch")));
+ assertTrue(store.deleteGlobalTransactionDO(global()));
+ assertTrue(store.deleteBranchTransactionDO(branch()));
+ verify(jedis)
+ .evalsha(
+ "sha",
+ Arrays.asList("SEATA_BRANCH_7", "SEATA_BRANCHES_host:8091:42", "xid"),
+ Collections.singletonList("branch"));
+ }
+
+ @Test
+ void updatesInterpretScriptResultsAndReloadEvictedScript() {
+ when(jedis.evalsha(anyString(), anyList(), anyList()))
+ .thenThrow(new JedisNoScriptException("evicted"))
+ .thenReturn("{\"success\":true}");
+ assertTrue(store.updateBranchTransactionDO(branch()));
+ assertTrue(store.updateGlobalTransactionDO(global()));
+ verify(jedis, times(3)).evalsha(anyString(), anyList(), anyList());
+ when(jedis.evalsha(anyString(), anyList(), anyList()))
+ .thenReturn("{\"success\":false,\"status\":\"NotExisted\"}");
+ assertThrows(RedisException.class, () -> store.updateBranchTransactionDO(branch()));
+ assertThrows(RedisException.class, () -> store.updateGlobalTransactionDO(global()));
+ when(jedis.evalsha(anyString(), anyList(), anyList()))
+ .thenReturn("{\"success\":false,\"status\":\"ChangeStatusFail\",\"data\":\"9\"}");
+ assertThrows(RedisException.class, () -> store.updateGlobalTransactionDO(global()));
+ }
+
+ @Test
+ void failedScriptLoadFallsBackToPipeline() {
+ when(jedis.scriptLoad(anyString())).thenThrow(new UnsupportedOperationException("no scripting"));
+ store = new RedisLuaTransactionStoreManager();
+ assertTrue(scripts.isEmpty());
+ Pipeline pipeline = mock(Pipeline.class);
+ when(jedis.pipelined()).thenReturn(pipeline);
+ assertTrue(store.insertGlobalTransactionDO(global()));
+ assertTrue(store.insertBranchTransactionDO(branch()));
+ assertTrue(store.deleteGlobalTransactionDO(global()));
+ assertTrue(store.deleteBranchTransactionDO(branch()));
+ when(jedis.hget("SEATA_BRANCH_7", "status")).thenReturn("1");
+ assertTrue(store.updateBranchTransactionDO(branch()));
+ when(jedis.hmget(anyString(), any(String[].class))).thenReturn(Arrays.asList("2", "100"));
+ assertTrue(store.updateGlobalTransactionDO(global()));
+ verify(pipeline, times(2)).sync();
+ }
+
+ @Test
+ void everyScriptWriteWrapsConnectionFailure() {
+ factory.when(JedisPooledFactory::getJedisInstance).thenThrow(new IllegalStateException("offline"));
+ assertThrows(RedisException.class, () -> store.insertGlobalTransactionDO(global()));
+ assertThrows(RedisException.class, () -> store.updateGlobalTransactionDO(global()));
+ assertThrows(RedisException.class, () -> store.deleteGlobalTransactionDO(global()));
+ assertThrows(RedisException.class, () -> store.insertBranchTransactionDO(branch()));
+ assertThrows(RedisException.class, () -> store.updateBranchTransactionDO(branch()));
+ assertThrows(RedisException.class, () -> store.deleteBranchTransactionDO(branch()));
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/redis/store/RedisTransactionStoreUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/redis/store/RedisTransactionStoreUnitTest.java
new file mode 100644
index 0000000000..c62bc39628
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/redis/store/RedisTransactionStoreUnitTest.java
@@ -0,0 +1,273 @@
+/*
+ * 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.server.storage.redis.store;
+
+import org.apache.seata.common.exception.RedisException;
+import org.apache.seata.common.util.BeanUtils;
+import org.apache.seata.core.model.GlobalStatus;
+import org.apache.seata.core.store.BranchTransactionDO;
+import org.apache.seata.core.store.GlobalTransactionDO;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.console.entity.param.GlobalSessionParam;
+import org.apache.seata.server.session.GlobalSession;
+import org.apache.seata.server.session.SessionCondition;
+import org.apache.seata.server.storage.redis.JedisPooledFactory;
+import org.junit.jupiter.api.*;
+import org.mockito.MockedStatic;
+import redis.clients.jedis.*;
+
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class RedisTransactionStoreUnitTest extends BaseSpringBootTest {
+ private Jedis jedis;
+ private Pipeline pipeline;
+ private Transaction transaction;
+ private MockedStatic<JedisPooledFactory> factory;
+ private RedisTransactionStoreManager store;
+
+ @BeforeEach
+ void open() {
+ jedis = mock(Jedis.class);
+ pipeline = mock(Pipeline.class);
+ transaction = mock(Transaction.class);
+ factory = mockStatic(JedisPooledFactory.class);
+ factory.when(JedisPooledFactory::getJedisInstance).thenReturn(jedis);
+ when(jedis.pipelined()).thenReturn(pipeline);
+ when(jedis.multi()).thenReturn(transaction);
+ store = new RedisTransactionStoreManager();
+ }
+
+ @AfterEach
+ void close() {
+ factory.close();
+ }
+
+ private GlobalTransactionDO global() {
+ GlobalTransactionDO g = new GlobalTransactionDO();
+ g.setXid("127.0.0.1:8091:42");
+ g.setTransactionId(42L);
+ g.setStatus(GlobalStatus.Begin.getCode());
+ g.setBeginTime(100L);
+ g.setTimeout(2000);
+ g.setApplicationId("app");
+ g.setTransactionServiceGroup("group");
+ g.setTransactionName("order");
+ return g;
+ }
+
+ private BranchTransactionDO branch() {
+ BranchTransactionDO b = new BranchTransactionDO();
+ b.setXid(global().getXid());
+ b.setTransactionId(42L);
+ b.setBranchId(7L);
+ b.setBranchType("AT");
+ b.setStatus(1);
+ b.setResourceId("db");
+ b.setClientId("client");
+ return b;
+ }
+
+ @Test
+ void insertIndexesAndTimestamps() {
+ GlobalTransactionDO g = global();
+ BranchTransactionDO b = branch();
+ assertTrue(store.insertGlobalTransactionDO(g));
+ assertTrue(store.insertBranchTransactionDO(b));
+ verify(pipeline).zadd("SEATA_BEGIN_TRANSACTIONS", 2100D, "SEATA_GLOBAL_42");
+ verify(pipeline).rpush("SEATA_STATUS_1", g.getXid());
+ verify(pipeline).rpush("SEATA_BRANCHES_" + g.getXid(), "SEATA_BRANCH_7");
+ verify(pipeline).hmset(eq("SEATA_BRANCH_7"), argThat(m -> "7".equals(m.get("branchId"))));
+ assertEquals(g.getGmtCreate(), g.getGmtModified());
+ assertNotNull(b.getGmtCreate());
+ verify(pipeline, times(2)).sync();
+ verify(jedis, times(2)).close();
+ }
+
+ @Test
+ void deleteIsIdempotentAndRemovesIndexes() {
+ assertTrue(store.deleteGlobalTransactionDO(global()));
+ assertTrue(store.deleteBranchTransactionDO(branch()));
+ verify(jedis, never()).pipelined();
+ when(jedis.hget(anyString(), eq("xid"))).thenReturn(global().getXid());
+ assertTrue(store.deleteGlobalTransactionDO(global()));
+ assertTrue(store.deleteBranchTransactionDO(branch()));
+ verify(pipeline).zrem("SEATA_BEGIN_TRANSACTIONS", "SEATA_GLOBAL_42");
+ verify(pipeline).del("SEATA_GLOBAL_42");
+ verify(pipeline).del("SEATA_BRANCH_7");
+ verify(pipeline).lrem("SEATA_BRANCHES_" + global().getXid(), 0, "SEATA_BRANCH_7");
+ }
+
+ @Test
+ void branchUpdateRejectsMissingAndPreservesBlankApplicationData() {
+ BranchTransactionDO b = branch();
+ assertThrows(RedisException.class, () -> store.updateBranchTransactionDO(b));
+ when(jedis.hget("SEATA_BRANCH_7", "status")).thenReturn("1");
+ assertTrue(store.updateBranchTransactionDO(b));
+ verify(jedis)
+ .hmset(
+ eq("SEATA_BRANCH_7"),
+ argThat(m -> !m.containsKey("applicationData") && "1".equals(m.get("status"))));
+ b.setApplicationData("payload");
+ assertTrue(store.updateBranchTransactionDO(b));
+ verify(jedis).hmset(eq("SEATA_BRANCH_7"), argThat(m -> "payload".equals(m.get("applicationData"))));
+ }
+
+ @Test
+ void globalUpdateHandlesMissingUnchangedInvalidAndConcurrentChanges() {
+ GlobalTransactionDO g = global();
+ when(jedis.hmget(anyString(), any(String[].class))).thenReturn(Arrays.asList(null, null));
+ assertThrows(RedisException.class, () -> store.updateGlobalTransactionDO(g));
+ when(jedis.hmget(anyString(), any(String[].class))).thenReturn(Arrays.asList("1", "100"));
+ assertTrue(store.updateGlobalTransactionDO(g));
+ verify(jedis, times(2)).unwatch();
+ verify(jedis, never()).multi();
+ g.setStatus(GlobalStatus.Committing.getCode());
+ when(transaction.exec()).thenReturn(Collections.emptyList());
+ assertTrue(store.updateGlobalTransactionDO(g));
+ when(transaction.exec()).thenReturn(Arrays.asList("OK", 1L, 1L, 1L));
+ assertTrue(store.updateGlobalTransactionDO(g));
+ verify(transaction, times(2)).zrem("SEATA_BEGIN_TRANSACTIONS", "SEATA_GLOBAL_42");
+ when(jedis.hmget(anyString(), any(String[].class)))
+ .thenReturn(Arrays.asList(String.valueOf(GlobalStatus.Rollbacking.getCode()), "100"));
+ assertThrows(RedisException.class, () -> store.updateGlobalTransactionDO(g));
+ }
+
+ @Test
+ void failedStatusIndexUpdateCompensatesSuccessfulOperations() {
+ GlobalTransactionDO g = global();
+ g.setStatus(GlobalStatus.Committing.getCode());
+ when(jedis.hmget(anyString(), any(String[].class))).thenReturn(Arrays.asList("1", "100"));
+ when(jedis.hget("SEATA_GLOBAL_42", "xid")).thenReturn(g.getXid());
+ when(transaction.exec()).thenReturn(Arrays.asList("OK", 0L, 1L));
+ assertFalse(store.updateGlobalTransactionDO(g));
+ verify(transaction).hmset(eq("SEATA_GLOBAL_42"), eq(new HashMap<String, String>() {
+ {
+ put("status", "1");
+ put("gmtModified", "100");
+ }
+ }));
+ verify(jedis).lrem("SEATA_STATUS_2", 0, g.getXid());
+ when(transaction.exec()).thenReturn(Arrays.asList("ERR", 1L, 0L));
+ assertFalse(store.updateGlobalTransactionDO(g));
+ verify(jedis).rpush("SEATA_STATUS_1", g.getXid());
+ }
+
+ @Test
+ void connectionFailuresAreWrappedForEveryWrite() {
+ factory.when(JedisPooledFactory::getJedisInstance).thenThrow(new IllegalStateException("offline"));
+ assertThrows(RedisException.class, () -> store.insertGlobalTransactionDO(global()));
+ assertThrows(RedisException.class, () -> store.deleteGlobalTransactionDO(global()));
+ assertThrows(RedisException.class, () -> store.updateGlobalTransactionDO(global()));
+ assertThrows(RedisException.class, () -> store.insertBranchTransactionDO(branch()));
+ assertThrows(RedisException.class, () -> store.deleteBranchTransactionDO(branch()));
+ assertThrows(RedisException.class, () -> store.updateBranchTransactionDO(branch()));
+ }
+
+ @Test
+ void readsReconstructSessionsAndHandleMissingRecords() {
+ String xid = global().getXid();
+ assertNull(store.readSession(xid));
+ assertTrue(store.readSession(new SessionCondition(xid)).isEmpty());
+ SessionCondition condition = new SessionCondition();
+ condition.setTransactionId(42L);
+ assertTrue(store.readSession(condition).isEmpty());
+ assertNull(store.readSession(new SessionCondition()));
+ when(jedis.hgetAll("SEATA_GLOBAL_42")).thenReturn(BeanUtils.objectToMap(global()));
+ when(jedis.lrange("SEATA_BRANCHES_" + xid, 0, 20)).thenReturn(Collections.singletonList("SEATA_BRANCH_7"));
+ when(pipeline.syncAndReturnAll()).thenReturn(Arrays.asList(null, BeanUtils.objectToMap(branch())));
+ GlobalSession session = store.readSession(xid);
+ assertEquals(xid, session.getXid());
+ assertEquals(1, session.getBranchSessions().size());
+ assertEquals(7L, session.getBranchSessions().get(0).getBranchId());
+ assertEquals(1, store.findBranchSessionByXid(xid).size());
+ assertEquals(1, store.readSession(condition).size());
+ assertEquals(1, store.readSession(new SessionCondition(xid)).size());
+ assertTrue(store.readSession(xid, false).getBranchSessions().isEmpty());
+ }
+
+ @Test
+ void statusPaginationCountsAndLimitsResults() {
+ RedisTransactionStoreManager reader = spy(store);
+ GlobalSession session = new GlobalSession();
+ doReturn(session).when(reader).readSession(anyString(), anyBoolean());
+ when(pipeline.syncAndReturnAll()).thenReturn(Arrays.asList(2L, 1L));
+ assertEquals(
+ 3L, reader.countByGlobalSessions(new GlobalStatus[] {GlobalStatus.Begin, GlobalStatus.Committing}));
+ when(jedis.lrange(anyString(), anyLong(), anyLong())).thenReturn(Collections.singletonList(global().getXid()));
+ assertEquals(1, reader.findGlobalSessionByPage(1, 1, false).size());
+ GlobalSessionParam param = new GlobalSessionParam();
+ param.setPageNum(1);
+ param.setPageSize(2);
+ assertTrue(reader.readSessionStatusByPage(param).isEmpty());
+ param.setStatus(1);
+ assertEquals(1, reader.readSessionStatusByPage(param).size());
+ reader.setLogQueryLimit(1);
+ assertEquals(
+ 1,
+ reader.readSession(new GlobalStatus[] {GlobalStatus.Begin, GlobalStatus.Committing}, false)
+ .size());
+ reader.setLogQueryLimit(0);
+ assertTrue(reader.readSession(new GlobalStatus[] {GlobalStatus.Begin, GlobalStatus.Committing}, false)
+ .isEmpty());
+ when(pipeline.syncAndReturnAll()).thenReturn(Collections.singletonList(1L));
+ assertTrue(reader.readSortByTimeoutBeginSessions(false).isEmpty());
+ when(pipeline.syncAndReturnAll()).thenReturn(Collections.emptyList());
+ assertEquals(0L, reader.countByGlobalSessions(new GlobalStatus[0]));
+ }
+
+ @Test
+ void timeoutQueryLoadsBranchesAndDispatchesConditions() {
+ when(pipeline.syncAndReturnAll())
+ .thenReturn(Collections.singletonList(1L), Collections.singletonList(BeanUtils.objectToMap(global())));
+ when(jedis.zrangeByScore(eq("SEATA_BEGIN_TRANSACTIONS"), eq(0D), anyDouble(), eq(0), eq(1)))
+ .thenReturn(Collections.singletonList("SEATA_GLOBAL_42"));
+ assertEquals(
+ global().getXid(),
+ store.readSortByTimeoutBeginSessions(true).get(0).getXid());
+ RedisTransactionStoreManager reader = spy(store);
+ doReturn(Collections.emptyList()).when(reader).readSortByTimeoutBeginSessions(anyBoolean());
+ doReturn(Collections.emptyList()).when(reader).readSession(any(GlobalStatus[].class), anyBoolean());
+ reader.readSession(new SessionCondition(GlobalStatus.Begin));
+ reader.readSession(new SessionCondition(GlobalStatus.Committing));
+ verify(reader).readSortByTimeoutBeginSessions(true);
+ verify(reader).readSession(new GlobalStatus[] {GlobalStatus.Committing}, true);
+ }
+
+ @Test
+ void sessionWritesDispatchToGlobalAndBranchStores() {
+ org.apache.seata.server.session.GlobalSession global =
+ new org.apache.seata.server.session.GlobalSession("app", "group", "order", 1000);
+ global.setXid("127.0.0.1:8091:42");
+ global.setTransactionId(42L);
+ global.setStatus(GlobalStatus.Begin);
+ org.apache.seata.server.session.BranchSession branch = new org.apache.seata.server.session.BranchSession();
+ branch.setXid(global.getXid());
+ branch.setTransactionId(42L);
+ branch.setBranchId(7L);
+ branch.setBranchType(org.apache.seata.core.model.BranchType.AT);
+ branch.setStatus(org.apache.seata.core.model.BranchStatus.Registered);
+ assertTrue(store.writeSession(
+ org.apache.seata.server.store.TransactionStoreManager.LogOperation.GLOBAL_ADD, global));
+ assertTrue(store.writeSession(
+ org.apache.seata.server.store.TransactionStoreManager.LogOperation.BRANCH_ADD, branch));
+ verify(pipeline).hmset(eq("SEATA_GLOBAL_42"), anyMap());
+ verify(pipeline).hmset(eq("SEATA_BRANCH_7"), anyMap());
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/storage/redis/store/RedisVGroupMappingUnitTest.java b/server/src/test/java/org/apache/seata/server/storage/redis/store/RedisVGroupMappingUnitTest.java
new file mode 100644
index 0000000000..ae3621ad75
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/storage/redis/store/RedisVGroupMappingUnitTest.java
@@ -0,0 +1,68 @@
+/*
+ * 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.server.storage.redis.store;
+
+import org.apache.seata.common.exception.RedisException;
+import org.apache.seata.common.metadata.Instance;
+import org.apache.seata.core.store.MappingDO;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.storage.redis.JedisPooledFactory;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import redis.clients.jedis.Jedis;
+
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class RedisVGroupMappingUnitTest extends BaseSpringBootTest {
+ @Test
+ void mappingOperationsAreScopedToNamespaceAndCluster() {
+ Jedis jedis = mock(Jedis.class);
+ Instance instance = mock(Instance.class);
+ when(instance.getNamespace()).thenReturn("tenant");
+ when(instance.getClusterName()).thenReturn("east");
+ try (MockedStatic<JedisPooledFactory> pool = mockStatic(JedisPooledFactory.class);
+ MockedStatic<Instance> instances = mockStatic(Instance.class)) {
+ pool.when(JedisPooledFactory::getJedisInstance).thenReturn(jedis);
+ instances.when(Instance::getInstance).thenReturn(instance);
+ RedisVGroupMappingStoreManager manager = new RedisVGroupMappingStoreManager();
+ MappingDO mapping = new MappingDO();
+ mapping.setVGroup("payments");
+ mapping.setNamespace("tenant");
+ mapping.setCluster("east");
+ assertTrue(manager.addVGroup(mapping));
+ String key = "SEATA_NAMINGSERVER_NAMESPACE_tenant";
+ verify(jedis).hset(key, "payments", "east");
+ when(jedis.hget(key, "payments")).thenReturn("west", "east");
+ assertFalse(manager.removeVGroup("payments"));
+ assertTrue(manager.removeVGroup("payments"));
+ verify(jedis).hdel(key, "payments");
+ Map<String, String> mappings = new HashMap<>();
+ mappings.put("payments", "east");
+ mappings.put("orders", "west");
+ when(jedis.hgetAll(key)).thenReturn(mappings);
+ assertEquals(
+ Collections.singleton("payments"), manager.loadVGroups().keySet());
+ pool.when(JedisPooledFactory::getJedisInstance).thenThrow(new IllegalStateException("offline"));
+ assertThrows(RedisException.class, () -> manager.addVGroup(mapping));
+ assertThrows(RedisException.class, () -> manager.removeVGroup("payments"));
+ assertThrows(RedisException.class, manager::loadVGroups);
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/transaction/at/ATLockUnitTest.java b/server/src/test/java/org/apache/seata/server/transaction/at/ATLockUnitTest.java
new file mode 100644
index 0000000000..0d643306af
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/transaction/at/ATLockUnitTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.server.transaction.at;
+
+import org.apache.seata.common.exception.StoreException;
+import org.apache.seata.core.exception.*;
+import org.apache.seata.core.rpc.RemotingServer;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.session.*;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class ATLockUnitTest extends BaseSpringBootTest {
+ @Test
+ void applicationDataControlsLockFlagsAndMalformedJsonFallsBack() throws Exception {
+ ATCore core = new ATCore(mock(RemotingServer.class));
+ GlobalSession global = mock(GlobalSession.class);
+ BranchSession branch = mock(BranchSession.class);
+ when(branch.lock(anyBoolean(), anyBoolean())).thenReturn(true);
+ when(branch.getApplicationData()).thenReturn("{\"autoCommit\":false,\"skipCheckLock\":true}");
+ core.branchSessionLock(global, branch);
+ verify(branch).lock(false, true);
+ when(branch.getApplicationData()).thenReturn("invalid-json");
+ core.branchSessionLock(global, branch);
+ verify(branch).lock(true, false);
+ when(branch.lock(true, false)).thenReturn(false);
+ assertEquals(
+ TransactionExceptionCode.LockKeyConflict,
+ assertThrows(BranchTransactionException.class, () -> core.branchSessionLock(global, branch))
+ .getCode());
+ doThrow(new StoreException(new BranchTransactionException(TransactionExceptionCode.LockKeyConflictFailFast)))
+ .when(branch)
+ .lock(anyBoolean(), anyBoolean());
+ assertEquals(
+ TransactionExceptionCode.LockKeyConflictFailFast,
+ assertThrows(BranchTransactionException.class, () -> core.branchSessionLock(global, branch))
+ .getCode());
+ doThrow(new StoreException("offline")).when(branch).lock(anyBoolean(), anyBoolean());
+ assertThrows(StoreException.class, () -> core.branchSessionLock(global, branch));
+ core.branchSessionUnlock(branch);
+ verify(branch).unlock();
+ }
+
+ @Test
+ void transportFailuresAreTranslatedToBranchErrors() throws Exception {
+ RemotingServer remoting = mock(RemotingServer.class);
+ ATCore core = new ATCore(remoting);
+ BranchSession branch = new BranchSession();
+ branch.setResourceId("db");
+ branch.setClientId("client");
+ when(remoting.sendSyncRequest(eq("db"), eq("client"), any(), anyBoolean()))
+ .thenThrow(new IOException("offline"));
+ assertEquals(
+ TransactionExceptionCode.FailedToSendBranchCommitRequest,
+ assertThrows(BranchTransactionException.class, () -> core.branchDelete(new GlobalSession(), branch))
+ .getCode());
+ assertEquals(
+ TransactionExceptionCode.FailedToSendBranchRollbackRequest,
+ assertThrows(BranchTransactionException.class, () -> core.branchRollback(new GlobalSession(), branch))
+ .getCode());
+ }
+}
diff --git a/server/src/test/java/org/apache/seata/server/transaction/saga/SagaCoreUnitTest.java b/server/src/test/java/org/apache/seata/server/transaction/saga/SagaCoreUnitTest.java
new file mode 100644
index 0000000000..814cdaa564
--- /dev/null
+++ b/server/src/test/java/org/apache/seata/server/transaction/saga/SagaCoreUnitTest.java
@@ -0,0 +1,204 @@
+/*
+ * 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.server.transaction.saga;
+
+import io.netty.channel.Channel;
+import org.apache.seata.common.exception.ShouldNeverHappenException;
+import org.apache.seata.core.exception.TransactionException;
+import org.apache.seata.core.model.*;
+import org.apache.seata.core.protocol.transaction.*;
+import org.apache.seata.core.rpc.RemotingServer;
+import org.apache.seata.core.rpc.netty.ChannelManager;
+import org.apache.seata.server.BaseSpringBootTest;
+import org.apache.seata.server.session.*;
+import org.junit.jupiter.api.*;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.mockito.MockedStatic;
+
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class SagaCoreUnitTest extends BaseSpringBootTest {
+ private RemotingServer remoting;
+ private SagaCore core;
+ private GlobalSession session;
+ private SessionManager manager;
+ private MockedStatic<SessionHelper> helper;
+ private MockedStatic<SessionHolder> holder;
+
+ @BeforeEach
+ void open() {
+ remoting = mock(RemotingServer.class);
+ core = spy(new SagaCore(remoting));
+ session = mock(GlobalSession.class);
+ when(session.getXid()).thenReturn("host:8091:42");
+ when(session.getApplicationId()).thenReturn("app");
+ when(session.getTransactionServiceGroup()).thenReturn("group");
+ when(session.getStatus()).thenReturn(GlobalStatus.Begin);
+ manager = mock(SessionManager.class);
+ holder = mockStatic(SessionHolder.class);
+ holder.when(SessionHolder::getRootSessionManager).thenReturn(manager);
+ helper = mockStatic(SessionHelper.class);
+ helper.when(() -> SessionHelper.newBranch(any(), anyString(), anyLong(), anyString(), anyString()))
+ .thenReturn(new BranchSession());
+ }
+
+ @AfterEach
+ void close() {
+ helper.close();
+ holder.close();
+ }
+
+ @Test
+ void sendUsesSagaResourceChannelAndReturnsRetryableWhenUnavailable() throws Exception {
+ BranchCommitRequest commit = new BranchCommitRequest();
+ BranchRollbackRequest rollback = new BranchRollbackRequest();
+ try (MockedStatic<ChannelManager> channels = mockStatic(ChannelManager.class)) {
+ channels.when(ChannelManager::getRmChannels).thenReturn(Collections.emptyMap());
+ assertEquals(BranchStatus.PhaseTwo_CommitFailed_Retryable, core.branchCommitSend(commit, session, null));
+ assertEquals(
+ BranchStatus.PhaseTwo_RollbackFailed_Retryable, core.branchRollbackSend(rollback, session, null));
+ Channel channel = mock(Channel.class);
+ channels.when(ChannelManager::getRmChannels).thenReturn(Collections.singletonMap("other", channel));
+ assertEquals(BranchStatus.PhaseTwo_CommitFailed_Retryable, core.branchCommitSend(commit, session, null));
+ assertEquals(
+ BranchStatus.PhaseTwo_RollbackFailed_Retryable, core.branchRollbackSend(rollback, session, null));
+ channels.when(ChannelManager::getRmChannels).thenReturn(Collections.singletonMap("app#group", channel));
+ BranchCommitResponse cr = new BranchCommitResponse();
+ cr.setBranchStatus(BranchStatus.PhaseTwo_Committed);
+ BranchRollbackResponse rr = new BranchRollbackResponse();
+ rr.setBranchStatus(BranchStatus.PhaseTwo_Rollbacked);
+ when(remoting.sendSyncRequest(channel, commit)).thenReturn(cr);
+ when(remoting.sendSyncRequest(channel, rollback)).thenReturn(rr);
+ assertEquals(BranchStatus.PhaseTwo_Committed, core.branchCommitSend(commit, session, null));
+ assertEquals(BranchStatus.PhaseTwo_Rollbacked, core.branchRollbackSend(rollback, session, null));
+ assertEquals(BranchType.SAGA, core.getHandleBranchType());
+ core.globalSessionStatusCheck(session);
+ assertThrows(ShouldNeverHappenException.class, () -> core.branchDelete(session, new BranchSession()));
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = BranchStatus.class,
+ names = {
+ "PhaseTwo_Committed",
+ "PhaseTwo_Rollbacked",
+ "PhaseTwo_RollbackFailed_Retryable",
+ "PhaseOne_Failed",
+ "PhaseTwo_CommitFailed_Unretryable",
+ "PhaseTwo_CommitFailed_Retryable"
+ })
+ void commitRoutesOutcomeToCleanupOrRecovery(BranchStatus status) throws Exception {
+ doReturn(status).when(core).branchCommit(eq(session), any(BranchSession.class));
+ assertEquals(status == BranchStatus.PhaseTwo_Committed, core.doGlobalCommit(session, false));
+ switch (status) {
+ case PhaseTwo_Committed:
+ helper.verify(() -> SessionHelper.removeAllBranch(session, true));
+ break;
+ case PhaseTwo_Rollbacked:
+ helper.verify(() -> SessionHelper.endRollbacked(session, false));
+ break;
+ case PhaseTwo_RollbackFailed_Retryable:
+ verify(manager).removeGlobalSession(session);
+ verify(session).queueToRetryRollback();
+ break;
+ case PhaseOne_Failed:
+ verify(session).changeGlobalStatus(GlobalStatus.Finished);
+ verify(session).end();
+ break;
+ case PhaseTwo_CommitFailed_Unretryable:
+ helper.verify(() -> SessionHelper.endCommitFailed(session, false));
+ when(session.canBeCommittedAsync()).thenReturn(true);
+ assertTrue(core.doGlobalCommit(session, true));
+ break;
+ default:
+ verify(session).queueToRetryCommit();
+ assertFalse(core.doGlobalCommit(session, true));
+ verify(session, times(1)).queueToRetryCommit();
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = BranchStatus.class,
+ names = {
+ "PhaseTwo_Rollbacked",
+ "PhaseTwo_RollbackFailed_Unretryable",
+ "PhaseTwo_CommitFailed_Retryable",
+ "PhaseTwo_RollbackFailed_Retryable"
+ })
+ void rollbackRoutesOutcomeToCleanupOrRecovery(BranchStatus status) throws Exception {
+ doReturn(status).when(core).branchRollback(eq(session), any(BranchSession.class));
+ assertEquals(status == BranchStatus.PhaseTwo_Rollbacked, core.doGlobalRollback(session, false));
+ switch (status) {
+ case PhaseTwo_Rollbacked:
+ helper.verify(() -> SessionHelper.removeAllBranch(session, true));
+ break;
+ case PhaseTwo_RollbackFailed_Unretryable:
+ helper.verify(() -> SessionHelper.endRollbackFailed(session, false));
+ break;
+ case PhaseTwo_CommitFailed_Retryable:
+ verify(manager).removeGlobalSession(session);
+ verify(session).queueToRetryCommit();
+ break;
+ default:
+ verify(session).queueToRetryRollback();
+ assertFalse(core.doGlobalRollback(session, true));
+ verify(session, times(1)).queueToRetryRollback();
+ }
+ }
+
+ @Test
+ void transportFailuresQueueRecoveryOnlyOnFirstAttempt() throws Exception {
+ doThrow(new TransactionException("offline")).when(core).branchCommit(eq(session), any(BranchSession.class));
+ assertThrows(TransactionException.class, () -> core.doGlobalCommit(session, false));
+ assertThrows(TransactionException.class, () -> core.doGlobalCommit(session, true));
+ verify(session).queueToRetryRollback();
+ clearInvocations(session);
+ doThrow(new TransactionException("offline")).when(core).branchRollback(eq(session), any(BranchSession.class));
+ assertThrows(TransactionException.class, () -> core.doGlobalRollback(session, false));
+ assertThrows(TransactionException.class, () -> core.doGlobalRollback(session, true));
+ verify(session).queueToRetryRollback();
+ }
+
+ @ParameterizedTest
+ @EnumSource(GlobalStatus.class)
+ void reportRoutesFinalAndRetryStatuses(GlobalStatus status) throws Exception {
+ core.doGlobalReport(session, session.getXid(), status);
+ if (status == GlobalStatus.Committed) {
+ helper.verify(() -> SessionHelper.endCommitted(session, false));
+ } else if (status == GlobalStatus.Rollbacked || status == GlobalStatus.Finished) {
+ helper.verify(() -> SessionHelper.endRollbacked(session, false));
+ } else {
+ verify(session).changeGlobalStatus(status);
+ if (status == GlobalStatus.RollbackRetrying
+ || status == GlobalStatus.TimeoutRollbackRetrying
+ || status == GlobalStatus.UnKnown) {
+ verify(session).queueToRetryRollback();
+ } else if (status == GlobalStatus.CommitRetrying) {
+ verify(session).queueToRetryCommit();
+ } else {
+ verify(session, never()).queueToRetryCommit();
+ verify(session, never()).queueToRetryRollback();
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org