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 2024/11/27 06:13:10 UTC
(incubator-seata) branch 2.x updated: optimize: raft mode maintains the reload logic consistent with the file (#7027)
This is an automated email from the ASF dual-hosted git repository.
jimin 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 c34aa4e1fa optimize: raft mode maintains the reload logic consistent with the file (#7027)
c34aa4e1fa is described below
commit c34aa4e1fa5155682750eee8af6611d4c04ee49d
Author: funkye <ji...@apache.org>
AuthorDate: Wed Nov 27 14:13:05 2024 +0800
optimize: raft mode maintains the reload logic consistent with the file (#7027)
---
changes/en-us/2.x.md | 1 +
changes/zh-cn/2.x.md | 1 +
.../apache/seata/server/session/SessionHolder.java | 18 ++++--
.../storage/file/session/FileSessionManager.java | 2 +-
.../storage/raft/session/RaftSessionManager.java | 27 +++++++-
.../raft/execute/BranchSessionExecuteTest.java | 75 ++++++++++------------
.../seata/server/raft/execute/LockExecuteTest.java | 37 ++++++++---
7 files changed, 103 insertions(+), 58 deletions(-)
diff --git a/changes/en-us/2.x.md b/changes/en-us/2.x.md
index 6e97f62977..c44e9f0d63 100644
--- a/changes/en-us/2.x.md
+++ b/changes/en-us/2.x.md
@@ -52,6 +52,7 @@ Add changes here for all PR submitted to the 2.x branch.
- [[#6995](https://github.com/apache/incubator-seata/pull/6995)] upgrade outdate npmjs dependencies
- [[#6996](https://github.com/apache/incubator-seata/pull/6996)] optimize lock release logic in AT transaction mode
- [[#7023](https://github.com/apache/incubator-seata/pull/7023)] optimize fail fast, when all server not available
+- [[#7027](https://github.com/apache/incubator-seata/pull/7027)] raft mode maintains the reload logic consistent with the file
### refactor:
diff --git a/changes/zh-cn/2.x.md b/changes/zh-cn/2.x.md
index b6b2262c57..5b6dc03fe0 100644
--- a/changes/zh-cn/2.x.md
+++ b/changes/zh-cn/2.x.md
@@ -55,6 +55,7 @@
- [[#6995](https://github.com/apache/incubator-seata/pull/6995)] 升级过时的 npmjs 依赖
- [[#6996](https://github.com/apache/incubator-seata/pull/6996)] 优化 AT 事务模式锁释放逻辑
- [[#7023](https://github.com/apache/incubator-seata/pull/7023)] 优化快速失败
+- [[#7027](https://github.com/apache/incubator-seata/pull/7027)] raft模式下reload行为于file保持一致
### refactor:
diff --git a/server/src/main/java/org/apache/seata/server/session/SessionHolder.java b/server/src/main/java/org/apache/seata/server/session/SessionHolder.java
index aa32ea001e..cd6f6273c7 100644
--- a/server/src/main/java/org/apache/seata/server/session/SessionHolder.java
+++ b/server/src/main/java/org/apache/seata/server/session/SessionHolder.java
@@ -170,6 +170,7 @@ public class SessionHolder {
public static void reload(Collection<GlobalSession> allSessions, SessionMode storeMode, boolean acquireLock) {
if ((SessionMode.FILE == storeMode || SessionMode.RAFT == storeMode)
&& CollectionUtils.isNotEmpty(allSessions)) {
+ long currentTimeMillis = System.currentTimeMillis();
for (GlobalSession globalSession : allSessions) {
GlobalStatus globalStatus = globalSession.getStatus();
switch (globalStatus) {
@@ -227,12 +228,17 @@ public class SessionHolder {
break;
case Begin:
if (Objects.equals(storeMode, SessionMode.RAFT)) {
- try {
- globalSession.changeGlobalStatus(GlobalStatus.RollbackRetrying);
- LOGGER.info("change global status: {}, xid: {}", globalSession.getStatus(),
- globalSession.getXid());
- } catch (TransactionException e) {
- LOGGER.error("change global status fail: {}", e.getMessage(), e);
+ // Avoid rolling back the global session created after becoming the leader.
+ if (globalSession.getBeginTime() < currentTimeMillis) {
+ try {
+ globalSession.changeGlobalStatus(GlobalStatus.RollbackRetrying);
+ LOGGER.info("change global status: {}, xid: {}", globalSession.getStatus(),
+ globalSession.getXid());
+ } catch (TransactionException e) {
+ LOGGER.error("change global status fail: {}", e.getMessage(), e);
+ }
+ } else {
+ globalSession.setActive(true);
}
} else {
globalSession.setActive(true);
diff --git a/server/src/main/java/org/apache/seata/server/storage/file/session/FileSessionManager.java b/server/src/main/java/org/apache/seata/server/storage/file/session/FileSessionManager.java
index d7a097fed2..c58f046434 100644
--- a/server/src/main/java/org/apache/seata/server/storage/file/session/FileSessionManager.java
+++ b/server/src/main/java/org/apache/seata/server/storage/file/session/FileSessionManager.java
@@ -66,7 +66,7 @@ public class FileSessionManager extends AbstractSessionManager implements Reload
/**
* The Session map.
*/
- private Map<String, GlobalSession> sessionMap = new ConcurrentHashMap<>(64);
+ protected Map<String, GlobalSession> sessionMap = new ConcurrentHashMap<>(64);
/**
diff --git a/server/src/main/java/org/apache/seata/server/storage/raft/session/RaftSessionManager.java b/server/src/main/java/org/apache/seata/server/storage/raft/session/RaftSessionManager.java
index b59c8d34ad..32d56c8e67 100644
--- a/server/src/main/java/org/apache/seata/server/storage/raft/session/RaftSessionManager.java
+++ b/server/src/main/java/org/apache/seata/server/storage/raft/session/RaftSessionManager.java
@@ -17,10 +17,12 @@
package org.apache.seata.server.storage.raft.session;
import java.io.IOException;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
import com.alipay.sofa.jraft.Closure;
import org.apache.seata.common.loader.LoadLevel;
import org.apache.seata.common.loader.Scope;
+import org.apache.seata.common.util.CollectionUtils;
import org.apache.seata.core.exception.TransactionException;
import org.apache.seata.core.exception.TransactionExceptionCode;
import org.apache.seata.core.model.BranchStatus;
@@ -80,7 +82,7 @@ public class RaftSessionManager extends FileSessionManager {
"seata raft state machine exception: " + status.getErrorMsg()));
} finally {
try {
- super.removeGlobalSession(globalSession);
+ removeGlobalSession(globalSession);
} catch (TransactionException e) {
completableFuture.completeExceptionally(e);
}
@@ -93,6 +95,23 @@ public class RaftSessionManager extends FileSessionManager {
RaftTaskUtil.createTask(closure, raftSyncMsg, completableFuture);
}
+ @Override
+ public void removeGlobalSession(GlobalSession session) throws TransactionException {
+ GlobalSession globalSession = sessionMap.remove(session.getXid());
+ if (globalSession != null) {
+ List<BranchSession> branchSessionList = globalSession.getBranchSessions();
+ // For the follower, the following code will not be executed because when the follower receives the remove global session
+ // the branch session on the leader side has already been completely cleared.
+ if (CollectionUtils.isNotEmpty(branchSessionList)) {
+ for (BranchSession branchSession : branchSessionList) {
+ branchSession.unlock();
+ onRemoveBranch(globalSession, branchSession);
+ }
+ end(globalSession);
+ }
+ }
+ }
+
@Override
public void onStatusChange(GlobalSession globalSession, GlobalStatus globalStatus) throws TransactionException {
CompletableFuture<Boolean> completableFuture = new CompletableFuture<>();
@@ -187,11 +206,15 @@ public class RaftSessionManager extends FileSessionManager {
@Override
public void onSuccessEnd(GlobalSession globalSession) throws TransactionException {
+ end(globalSession);
+ }
+
+ public void end(GlobalSession globalSession) throws TransactionException {
CompletableFuture<Boolean> completableFuture = new CompletableFuture<>();
Closure closure = status -> {
if (status.isOk()) {
try {
- super.removeGlobalSession(globalSession);
+ removeGlobalSession(globalSession);
completableFuture.complete(true);
} catch (TransactionException e) {
completableFuture.completeExceptionally(e);
diff --git a/server/src/test/java/org/apache/seata/server/raft/execute/BranchSessionExecuteTest.java b/server/src/test/java/org/apache/seata/server/raft/execute/BranchSessionExecuteTest.java
index ade3a25ef9..7c499f410c 100644
--- a/server/src/test/java/org/apache/seata/server/raft/execute/BranchSessionExecuteTest.java
+++ b/server/src/test/java/org/apache/seata/server/raft/execute/BranchSessionExecuteTest.java
@@ -16,7 +16,11 @@
*/
package org.apache.seata.server.raft.execute;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.seata.common.XID;
import org.apache.seata.common.util.NetUtil;
+import org.apache.seata.common.util.UUIDGenerator;
import org.apache.seata.config.ConfigurationCache;
import org.apache.seata.core.exception.TransactionException;
import org.apache.seata.core.model.BranchStatus;
@@ -33,10 +37,8 @@ import org.apache.seata.server.session.SessionHolder;
import org.apache.seata.server.storage.SessionConverter;
import org.apache.seata.server.store.StoreConfig;
import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.ApplicationContext;
@@ -51,8 +53,6 @@ class BranchSessionExecuteTest {
private static GlobalSession GLOBAL_SESSION;
- private static final String XID = "123:123";
-
private static final long BRANCH_ID = 0L;
@BeforeAll
@@ -61,10 +61,14 @@ class BranchSessionExecuteTest {
SessionHolder.init(StoreConfig.SessionMode.RAFT);
LockerManagerFactory.destroy();
LockerManagerFactory.init(StoreConfig.LockMode.RAFT);
+ GLOBAL_SESSION = mockGlobalSession();
+ SessionHolder.getRootSessionManager().addGlobalSession(GLOBAL_SESSION);
}
@AfterAll
- public static void destroy() throws TransactionException {
+ public static void destroy() throws Throwable {
+ testRemove();
+ SessionHolder.getRootSessionManager().removeGlobalSession(GLOBAL_SESSION);
// Clear configuration
ConfigurationCache.clear();
System.clearProperty("server.raft.serverAddr");
@@ -75,75 +79,66 @@ class BranchSessionExecuteTest {
LockerManagerFactory.destroy();
}
- @BeforeEach
- public void addGlobalSession() throws TransactionException {
- GLOBAL_SESSION = mockGlobalSession();
- SessionHolder.getRootSessionManager().addGlobalSession(GLOBAL_SESSION);
- }
- @AfterEach
- public void removeTestSession() throws TransactionException {
- SessionHolder.getRootSessionManager().removeGlobalSession(GLOBAL_SESSION);
- }
@Test
public void testAdd() throws Throwable {
- BranchSession expected = mockBranchSession();
+ BranchSession expected = mockBranchSession(GLOBAL_SESSION.getXid(), GLOBAL_SESSION.getTransactionId());
AddBranchSessionExecute execute = new AddBranchSessionExecute();
boolean success = execute.execute(convertToBranchSessionMsg(expected));
Assertions.assertTrue(success);
- BranchSession branchSession = GLOBAL_SESSION.getBranch(BRANCH_ID);
+ BranchSession branchSession = GLOBAL_SESSION.getBranch(expected.getBranchId());
assertBranchSessionValid(expected, branchSession);
}
- @Test
- public void testRemove() throws Throwable {
- GLOBAL_SESSION.add(mockBranchSession());
-
- BranchSession branchSession = GLOBAL_SESSION.getBranch(BRANCH_ID);
- Assertions.assertNotNull(branchSession);
+ public static void testRemove() throws Throwable {
+ List<BranchSession> list = new ArrayList<>(GLOBAL_SESSION.getBranchSessions());
+ for (BranchSession branchSession : list) {
+ Assertions.assertNotNull(branchSession);
- RemoveBranchSessionExecute execute = new RemoveBranchSessionExecute();
- boolean success = execute.execute(convertToBranchSessionMsg(branchSession));
- Assertions.assertTrue(success);
+ RemoveBranchSessionExecute execute = new RemoveBranchSessionExecute();
+ boolean success = execute.execute(convertToBranchSessionMsg(branchSession));
+ Assertions.assertTrue(success);
- branchSession = GLOBAL_SESSION.getBranch(BRANCH_ID);
- Assertions.assertNull(branchSession);
+ branchSession = GLOBAL_SESSION.getBranch(branchSession.getBranchId());
+ Assertions.assertNull(branchSession);
+ }
}
@Test
public void testUpdate() throws Throwable {
- GLOBAL_SESSION.add(mockBranchSession());
+ BranchSession branchSession = mockBranchSession(GLOBAL_SESSION.getXid(), GLOBAL_SESSION.getTransactionId());
+ GLOBAL_SESSION.add(branchSession);
- BranchSession branchSession = GLOBAL_SESSION.getBranch(BRANCH_ID);
+ branchSession = GLOBAL_SESSION.getBranch(branchSession.getBranchId());
Assertions.assertNotNull(branchSession);
- BranchSession expected = mockBranchSession();
- expected.setStatus(BranchStatus.PhaseTwo_Committed);
+ branchSession.setStatus(BranchStatus.PhaseTwo_Committed);
UpdateBranchSessionExecute execute = new UpdateBranchSessionExecute();
- boolean success = execute.execute(convertToBranchSessionMsg(expected));
+ boolean success = execute.execute(convertToBranchSessionMsg(branchSession));
Assertions.assertTrue(success);
- branchSession = GLOBAL_SESSION.getBranch(BRANCH_ID);
- assertBranchSessionValid(expected, branchSession);
+ branchSession = GLOBAL_SESSION.getBranch(branchSession.getBranchId());
+ assertBranchSessionValid(branchSession, branchSession);
}
private static GlobalSession mockGlobalSession() {
+ long txId = UUIDGenerator.generateUUID();
GlobalSession session = new GlobalSession("test", "test", "test", 5000);
- session.setXid(XID);
+ session.setXid(XID.generateXID(txId));
session.setApplicationData("hello, world");
- session.setTransactionId(123);
+ session.setTransactionId(txId);
session.setBeginTime(System.currentTimeMillis());
return session;
}
- private static BranchSession mockBranchSession() {
+ private static BranchSession mockBranchSession(String xid,long transactionId) {
BranchSession session = new BranchSession();
- session.setXid(XID);
- session.setTransactionId(123);
- session.setBranchId(BRANCH_ID);
+ session.setXid(xid);
+ session.setTransactionId(transactionId);
+ session.setBranchId(UUIDGenerator.generateUUID());
session.setClientId("client");
session.setResourceGroupId(DEFAULT_TX_GROUP);
session.setResourceId("resource");
diff --git a/server/src/test/java/org/apache/seata/server/raft/execute/LockExecuteTest.java b/server/src/test/java/org/apache/seata/server/raft/execute/LockExecuteTest.java
index 36cbbaf0d8..50aece2589 100644
--- a/server/src/test/java/org/apache/seata/server/raft/execute/LockExecuteTest.java
+++ b/server/src/test/java/org/apache/seata/server/raft/execute/LockExecuteTest.java
@@ -16,10 +16,14 @@
*/
package org.apache.seata.server.raft.execute;
+import java.util.ArrayList;
+import java.util.List;
import org.apache.seata.common.util.NetUtil;
+import org.apache.seata.common.util.UUIDGenerator;
import org.apache.seata.config.ConfigurationCache;
import org.apache.seata.core.exception.TransactionException;
import org.apache.seata.core.model.BranchType;
+import org.apache.seata.server.cluster.raft.execute.branch.RemoveBranchSessionExecute;
import org.apache.seata.server.cluster.raft.execute.lock.BranchReleaseLockExecute;
import org.apache.seata.server.cluster.raft.execute.lock.GlobalReleaseLockExecute;
import org.apache.seata.server.cluster.raft.sync.msg.RaftBranchSessionSyncMsg;
@@ -83,14 +87,29 @@ class LockExecuteTest {
}
@AfterEach
- public void removeTestSession() throws TransactionException {
+ public void removeTestSession() throws Throwable {
+ testRemove();
SessionHolder.getRootSessionManager().removeGlobalSession(GLOBAL_SESSION);
}
+ public void testRemove() throws Throwable {
+ List<BranchSession> list = new ArrayList<>(GLOBAL_SESSION.getBranchSessions());
+ for (BranchSession branchSession : list) {
+ Assertions.assertNotNull(branchSession);
+
+ RemoveBranchSessionExecute execute = new RemoveBranchSessionExecute();
+ boolean success = execute.execute(convertToBranchSessionMsg(branchSession));
+ Assertions.assertTrue(success);
+
+ branchSession = GLOBAL_SESSION.getBranch(branchSession.getBranchId());
+ Assertions.assertNull(branchSession);
+ }
+ }
+
@Test
public void testGlobalRelease() throws Throwable {
- BranchSession branchSession1 = mockBranchSession("test:0");
- BranchSession branchSession2 = mockBranchSession("test:1");
+ BranchSession branchSession1 = mockBranchSession(GLOBAL_SESSION.getXid(),GLOBAL_SESSION.getTransactionId(),"t1:53");
+ BranchSession branchSession2 = mockBranchSession(GLOBAL_SESSION.getXid(),GLOBAL_SESSION.getTransactionId(),"t1:54");
GLOBAL_SESSION.add(branchSession1);
GLOBAL_SESSION.add(branchSession2);
@@ -109,7 +128,7 @@ class LockExecuteTest {
@Test
public void testBranchRelease() throws Throwable {
- BranchSession branchSession = mockBranchSession("test:0");
+ BranchSession branchSession = mockBranchSession(GLOBAL_SESSION.getXid(),GLOBAL_SESSION.getTransactionId(),"t1:55");
GLOBAL_SESSION.add(branchSession);
LockManager lockerManager = LockerManagerFactory.getLockManager();
@@ -131,14 +150,14 @@ class LockExecuteTest {
return session;
}
- private static BranchSession mockBranchSession(String lockKey) {
+ private static BranchSession mockBranchSession(String xid,long transactionId,String lockKey) {
BranchSession session = new BranchSession();
- session.setXid(XID);
- session.setTransactionId(123);
- session.setBranchId(BRANCH_ID);
+ session.setXid(xid);
+ session.setTransactionId(transactionId);
+ session.setBranchId(UUIDGenerator.generateUUID());
session.setClientId("client");
session.setResourceGroupId(DEFAULT_TX_GROUP);
- session.setResourceId("db");
+ session.setResourceId("resource");
session.setLockKey(lockKey);
session.setBranchType(BranchType.AT);
session.setApplicationData("hello, world");
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org