update seata 1.5.2 => 1.6.0 适配升级

2026-05-13 07:02:08 +08:00 · 2022-12-19 11:59:29 +08:00
parent 0bb385f6cf
commit efc4bf5a0a
76 changed files with 908 additions and 778 deletions
--- a/ruoyi-visual/ruoyi-seata-server/src/main/java/io/seata/server/session/BranchSession.java
+++ b/ruoyi-visual/ruoyi-seata-server/src/main/java/io/seata/server/session/BranchSession.java
@@ -15,11 +15,6 @@
 */
 package io.seata.server.session;

-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
 import io.seata.common.util.CompressUtil;
 import io.seata.core.exception.TransactionException;
 import io.seata.core.model.BranchStatus;
@@ -32,6 +27,11 @@ import io.seata.server.store.StoreConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;

 import static io.seata.core.model.LockStatus.Locked;

--- a/ruoyi-visual/ruoyi-seata-server/src/main/java/io/seata/server/session/GlobalSession.java
+++ b/ruoyi-visual/ruoyi-seata-server/src/main/java/io/seata/server/session/GlobalSession.java
@@ -15,17 +15,6 @@
 */
 package io.seata.server.session;

-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-
 import io.seata.common.Constants;
 import io.seata.common.DefaultValues;
 import io.seata.common.XID;
@@ -46,9 +35,13 @@ import io.seata.server.store.StoreConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

-import static io.seata.core.model.GlobalStatus.AsyncCommitting;
-import static io.seata.core.model.GlobalStatus.CommitRetrying;
-import static io.seata.core.model.GlobalStatus.Committing;
+import java.nio.ByteBuffer;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import static io.seata.core.model.GlobalStatus.*;

 /**
 * The type Global session.
@@ -756,7 +749,7 @@ public class GlobalSession implements SessionLifecycle, SessionStorable {
    public void queueToRetryRollback() throws TransactionException {
        this.addSessionLifecycleListener(SessionHolder.getRetryRollbackingSessionManager());
        GlobalStatus currentStatus = this.getStatus();
-        if (SessionHelper.isTimeoutGlobalStatus(currentStatus)) {
+        if (SessionStatusValidator.isTimeoutGlobalStatus(currentStatus)) {
            this.setStatus(GlobalStatus.TimeoutRollbackRetrying);
        } else {
            this.setStatus(GlobalStatus.RollbackRetrying);
--- a/ruoyi-visual/ruoyi-seata-server/src/main/java/io/seata/server/session/SessionCondition.java
+++ b/ruoyi-visual/ruoyi-seata-server/src/main/java/io/seata/server/session/SessionCondition.java
@@ -89,6 +89,7 @@ public class SessionCondition {
     */
    public void setStatus(GlobalStatus status) {
        this.status = status;
+        this.statuses = new GlobalStatus[] {status};
    }

    /**
--- a/ruoyi-visual/ruoyi-seata-server/src/main/java/io/seata/server/session/SessionHelper.java
+++ b/ruoyi-visual/ruoyi-seata-server/src/main/java/io/seata/server/session/SessionHelper.java
@@ -15,12 +15,7 @@
 */
 package io.seata.server.session;

-import java.util.Collection;
-import java.util.List;
-import java.util.Objects;
-
 import io.seata.common.util.CollectionUtils;
-import io.seata.common.util.StringUtils;
 import io.seata.config.Configuration;
 import io.seata.config.ConfigurationFactory;
 import io.seata.core.constants.ConfigurationKeys;
@@ -28,15 +23,22 @@ import io.seata.core.context.RootContext;
 import io.seata.core.exception.TransactionException;
 import io.seata.core.model.BranchType;
 import io.seata.core.model.GlobalStatus;
-import io.seata.core.store.StoreMode;
 import io.seata.metrics.IdConstants;
 import io.seata.server.UUIDGenerator;
 import io.seata.server.coordinator.DefaultCoordinator;
 import io.seata.server.metrics.MetricsPublisher;
+import io.seata.server.store.StoreConfig;
+import io.seata.server.store.StoreConfig.SessionMode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;

+import java.util.Collection;
+import java.util.List;
+import java.util.Objects;
+
+import static io.seata.common.DefaultValues.DEFAULT_ENABLE_BRANCH_ASYNC_REMOVE;
+
 /**
 * The type Session helper.
 *
@@ -51,16 +53,14 @@ public class SessionHelper {
    private static final Configuration CONFIG = ConfigurationFactory.getInstance();

    private static final Boolean ENABLE_BRANCH_ASYNC_REMOVE = CONFIG.getBoolean(
-            ConfigurationKeys.ENABLE_BRANCH_ASYNC_REMOVE, false);
+            ConfigurationKeys.ENABLE_BRANCH_ASYNC_REMOVE, DEFAULT_ENABLE_BRANCH_ASYNC_REMOVE);

    /**
     * The instance of DefaultCoordinator
     */
    private static final DefaultCoordinator COORDINATOR = DefaultCoordinator.getInstance();

-    private static final boolean DELAY_HANDLE_SESSION =
-        !StringUtils.equalsIgnoreCase(ConfigurationFactory.getInstance().getConfig(ConfigurationKeys.STORE_SESSION_MODE,
-            ConfigurationFactory.getInstance().getConfig(ConfigurationKeys.STORE_MODE)), StoreMode.FILE.getName());
+    private static final boolean DELAY_HANDLE_SESSION = StoreConfig.getSessionMode() != SessionMode.FILE;

    private SessionHelper() {
    }
@@ -146,7 +146,24 @@ public class SessionHelper {
     * @throws TransactionException the transaction exception
     */
    public static void endCommitFailed(GlobalSession globalSession, boolean retryGlobal) throws TransactionException {
-        globalSession.changeGlobalStatus(GlobalStatus.CommitFailed);
+        endCommitFailed(globalSession, retryGlobal, false);
+    }
+
+    /**
+     * End commit failed.
+     *
+     * @param globalSession the global session
+     * @param retryGlobal the retry global
+     * @param isRetryTimeout is retry timeout
+     * @throws TransactionException the transaction exception
+     */
+    public static void endCommitFailed(GlobalSession globalSession, boolean retryGlobal, boolean isRetryTimeout)
+        throws TransactionException {
+        if (isRetryTimeout) {
+            globalSession.changeGlobalStatus(GlobalStatus.CommitRetryTimeout);
+        } else {
+            globalSession.changeGlobalStatus(GlobalStatus.CommitFailed);
+        }
        LOGGER.error("The Global session {} has changed the status to {}, need to be handled it manually.",
            globalSession.getXid(), globalSession.getStatus());

@@ -164,22 +181,27 @@ public class SessionHelper {
    public static void endRollbacked(GlobalSession globalSession, boolean retryGlobal) throws TransactionException {
        if (retryGlobal || !DELAY_HANDLE_SESSION) {
            long beginTime = System.currentTimeMillis();
+            boolean timeoutDone = false;
            GlobalStatus currentStatus = globalSession.getStatus();
+            if (currentStatus == GlobalStatus.TimeoutRollbacking) {
+                MetricsPublisher.postSessionDoneEvent(globalSession, GlobalStatus.TimeoutRollbacked, false, false);
+                timeoutDone = true;
+            }
            boolean retryBranch =
-                currentStatus == GlobalStatus.TimeoutRollbackRetrying || currentStatus == GlobalStatus.RollbackRetrying;
-            if (isTimeoutGlobalStatus(currentStatus)) {
+                    currentStatus == GlobalStatus.TimeoutRollbackRetrying || currentStatus == GlobalStatus.RollbackRetrying;
+            if (SessionStatusValidator.isTimeoutGlobalStatus(currentStatus)) {
                globalSession.changeGlobalStatus(GlobalStatus.TimeoutRollbacked);
            } else {
                globalSession.changeGlobalStatus(GlobalStatus.Rollbacked);
            }
            globalSession.end();
-            if (!DELAY_HANDLE_SESSION) {
+            if (!DELAY_HANDLE_SESSION && !timeoutDone) {
                MetricsPublisher.postSessionDoneEvent(globalSession, false, false);
            }
            MetricsPublisher.postSessionDoneEvent(globalSession, IdConstants.STATUS_VALUE_AFTER_ROLLBACKED_KEY, true,
-                beginTime, retryBranch);
+                    beginTime, retryBranch);
        } else {
-            MetricsPublisher.postSessionDoneEvent(globalSession, false, false);
+            MetricsPublisher.postSessionDoneEvent(globalSession, GlobalStatus.Rollbacked, false, false);
        }
    }

@@ -191,8 +213,22 @@ public class SessionHelper {
     * @throws TransactionException the transaction exception
     */
    public static void endRollbackFailed(GlobalSession globalSession, boolean retryGlobal) throws TransactionException {
+        endRollbackFailed(globalSession, retryGlobal, false);
+    }
+
+    /**
+     * End rollback failed.
+     *
+     * @param globalSession the global session
+     * @param retryGlobal   the retry global
+     * @param isRetryTimeout   is retry timeout
+     * @throws TransactionException the transaction exception
+     */
+    public static void endRollbackFailed(GlobalSession globalSession, boolean retryGlobal, boolean isRetryTimeout) throws TransactionException {
        GlobalStatus currentStatus = globalSession.getStatus();
-        if (isTimeoutGlobalStatus(currentStatus)) {
+        if (isRetryTimeout) {
+            globalSession.changeGlobalStatus(GlobalStatus.RollbackRetryTimeout);
+        } else if (SessionStatusValidator.isTimeoutGlobalStatus(currentStatus)) {
            globalSession.changeGlobalStatus(GlobalStatus.TimeoutRollbackFailed);
        } else {
            globalSession.changeGlobalStatus(GlobalStatus.RollbackFailed);
@@ -202,13 +238,6 @@ public class SessionHelper {
        MetricsPublisher.postSessionDoneEvent(globalSession, retryGlobal, false);
    }

-    public static boolean isTimeoutGlobalStatus(GlobalStatus status) {
-        return status == GlobalStatus.TimeoutRollbacked
-                || status == GlobalStatus.TimeoutRollbackFailed
-                || status == GlobalStatus.TimeoutRollbacking
-                || status == GlobalStatus.TimeoutRollbackRetrying;
-    }
-
    /**
     * Foreach global sessions.
     *
--- a/ruoyi-visual/ruoyi-seata-server/src/main/java/io/seata/server/session/SessionHolder.java
+++ b/ruoyi-visual/ruoyi-seata-server/src/main/java/io/seata/server/session/SessionHolder.java
@@ -15,16 +15,7 @@
 */
 package io.seata.server.session;

-import java.io.IOException;
-import java.util.Collection;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-
 import io.seata.common.ConfigurationKeys;
-import io.seata.core.model.LockStatus;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import io.seata.common.XID;
 import io.seata.common.exception.ShouldNeverHappenException;
 import io.seata.common.exception.StoreException;
@@ -35,12 +26,21 @@ import io.seata.config.Configuration;
 import io.seata.config.ConfigurationFactory;
 import io.seata.core.exception.TransactionException;
 import io.seata.core.model.GlobalStatus;
+import io.seata.core.model.LockStatus;
 import io.seata.core.store.DistributedLockDO;
 import io.seata.core.store.DistributedLocker;
 import io.seata.server.lock.distributed.DistributedLockerFactory;
-import io.seata.core.store.StoreMode;
+import io.seata.server.store.StoreConfig;
+import io.seata.server.store.StoreConfig.SessionMode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;

-import static io.seata.common.DefaultValues.SERVER_DEFAULT_STORE_MODE;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import static io.seata.common.DefaultValues.DEFAULT_DISTRIBUTED_LOCK_EXPIRE_TIME;

 /**
 * The type Session holder.
@@ -80,7 +80,7 @@ public class SessionHolder {
    /**
     * The redis distributed lock expire time
     */
-    private static long DISTRIBUTED_LOCK_EXPIRE_TIME = CONFIG.getLong(ConfigurationKeys.DISTRIBUTED_LOCK_EXPIRE_TIME, 10000);
+    private static long DISTRIBUTED_LOCK_EXPIRE_TIME = CONFIG.getLong(ConfigurationKeys.DISTRIBUTED_LOCK_EXPIRE_TIME, DEFAULT_DISTRIBUTED_LOCK_EXPIRE_TIME);

    private static SessionManager ROOT_SESSION_MANAGER;
    private static SessionManager ASYNC_COMMITTING_SESSION_MANAGER;
@@ -89,56 +89,57 @@ public class SessionHolder {

    private static DistributedLocker DISTRIBUTED_LOCKER;

+    public static void init() {
+        init(null);
+    }
    /**
     * Init.
     *
-     * @param mode the store mode: file, db, redis
+     * @param sessionMode the store mode: file, db, redis
     * @throws IOException the io exception
     */
-    public static void init(String mode) {
-        if (StringUtils.isBlank(mode)) {
-            mode = CONFIG.getConfig(ConfigurationKeys.STORE_SESSION_MODE,
-                    CONFIG.getConfig(ConfigurationKeys.STORE_MODE, SERVER_DEFAULT_STORE_MODE));
+    public static void init(SessionMode sessionMode) {
+        if (null == sessionMode) {
+            sessionMode = StoreConfig.getSessionMode();
        }
-        StoreMode storeMode = StoreMode.get(mode);
-        if (StoreMode.DB.equals(storeMode)) {
-            ROOT_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, StoreMode.DB.getName());
-            ASYNC_COMMITTING_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, StoreMode.DB.getName(),
+        if (SessionMode.DB.equals(sessionMode)) {
+            ROOT_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, SessionMode.DB.getName());
+            ASYNC_COMMITTING_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, SessionMode.DB.getName(),
                new Object[]{ASYNC_COMMITTING_SESSION_MANAGER_NAME});
-            RETRY_COMMITTING_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, StoreMode.DB.getName(),
+            RETRY_COMMITTING_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, SessionMode.DB.getName(),
                new Object[]{RETRY_COMMITTING_SESSION_MANAGER_NAME});
-            RETRY_ROLLBACKING_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, StoreMode.DB.getName(),
+            RETRY_ROLLBACKING_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, SessionMode.DB.getName(),
                new Object[]{RETRY_ROLLBACKING_SESSION_MANAGER_NAME});

-            DISTRIBUTED_LOCKER = DistributedLockerFactory.getDistributedLocker(StoreMode.DB.getName());
-        } else if (StoreMode.FILE.equals(storeMode)) {
+            DISTRIBUTED_LOCKER = DistributedLockerFactory.getDistributedLocker(SessionMode.DB.getName());
+        } else if (SessionMode.FILE.equals(sessionMode)) {
            String sessionStorePath = CONFIG.getConfig(ConfigurationKeys.STORE_FILE_DIR,
                    DEFAULT_SESSION_STORE_FILE_DIR);
            if (StringUtils.isBlank(sessionStorePath)) {
                throw new StoreException("the {store.file.dir} is empty.");
            }
-            ROOT_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, StoreMode.FILE.getName(),
+            ROOT_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, SessionMode.FILE.getName(),
                new Object[]{ROOT_SESSION_MANAGER_NAME, sessionStorePath});
            ASYNC_COMMITTING_SESSION_MANAGER = ROOT_SESSION_MANAGER;
            RETRY_COMMITTING_SESSION_MANAGER = ROOT_SESSION_MANAGER;
            RETRY_ROLLBACKING_SESSION_MANAGER = ROOT_SESSION_MANAGER;

-            DISTRIBUTED_LOCKER = DistributedLockerFactory.getDistributedLocker(StoreMode.FILE.getName());
-        } else if (StoreMode.REDIS.equals(storeMode)) {
-            ROOT_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, StoreMode.REDIS.getName());
+            DISTRIBUTED_LOCKER = DistributedLockerFactory.getDistributedLocker(SessionMode.FILE.getName());
+        } else if (SessionMode.REDIS.equals(sessionMode)) {
+            ROOT_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, SessionMode.REDIS.getName());
            ASYNC_COMMITTING_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class,
-                StoreMode.REDIS.getName(), new Object[]{ASYNC_COMMITTING_SESSION_MANAGER_NAME});
+                SessionMode.REDIS.getName(), new Object[]{ASYNC_COMMITTING_SESSION_MANAGER_NAME});
            RETRY_COMMITTING_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class,
-                StoreMode.REDIS.getName(), new Object[]{RETRY_COMMITTING_SESSION_MANAGER_NAME});
+                SessionMode.REDIS.getName(), new Object[]{RETRY_COMMITTING_SESSION_MANAGER_NAME});
            RETRY_ROLLBACKING_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class,
-                StoreMode.REDIS.getName(), new Object[]{RETRY_ROLLBACKING_SESSION_MANAGER_NAME});
+                SessionMode.REDIS.getName(), new Object[]{RETRY_ROLLBACKING_SESSION_MANAGER_NAME});

-            DISTRIBUTED_LOCKER = DistributedLockerFactory.getDistributedLocker(StoreMode.REDIS.getName());
+            DISTRIBUTED_LOCKER = DistributedLockerFactory.getDistributedLocker(SessionMode.REDIS.getName());
        } else {
            // unknown store
-            throw new IllegalArgumentException("unknown store mode:" + mode);
+            throw new IllegalArgumentException("unknown store mode:" + sessionMode.getName());
        }
-        reload(storeMode);
+        reload(sessionMode);
    }

    //region reload
@@ -146,15 +147,15 @@ public class SessionHolder {
    /**
     * Reload.
     *
-     * @param storeMode the mode of store
+     * @param sessionMode the mode of store
     */
-    protected static void reload(StoreMode storeMode) {
+    protected static void reload(SessionMode sessionMode) {

        if (ROOT_SESSION_MANAGER instanceof Reloadable) {
            ((Reloadable) ROOT_SESSION_MANAGER).reload();
        }

-        if (storeMode == StoreMode.FILE) {
+        if (SessionMode.FILE.equals(sessionMode)) {
            Collection<GlobalSession> allSessions = ROOT_SESSION_MANAGER.allSessions();
            if (CollectionUtils.isNotEmpty(allSessions)) {
                for (GlobalSession globalSession : allSessions) {
--- a/ruoyi-visual/ruoyi-seata-server/src/main/java/io/seata/server/session/SessionManager.java
+++ b/ruoyi-visual/ruoyi-seata-server/src/main/java/io/seata/server/session/SessionManager.java
@@ -15,13 +15,14 @@
 */
 package io.seata.server.session;

-import java.util.Collection;
-import java.util.List;
 import io.seata.core.exception.TransactionException;
 import io.seata.core.model.BranchStatus;
 import io.seata.core.model.GlobalStatus;
 import io.seata.core.rpc.Disposable;

+import java.util.Collection;
+import java.util.List;
+
 /**
 * The interface Session manager.
 *
@@ -119,6 +120,7 @@ public interface SessionManager extends SessionLifecycleListener, Disposable {
     * @param globalSession the global session
     * @param lockCallable the lock Callable
     * @return the value
+     * @throws TransactionException the transaction exception
     */
    <T> T lockAndExecute(GlobalSession globalSession, GlobalSession.LockCallable<T> lockCallable)
            throws TransactionException;
--- a/ruoyi-visual/ruoyi-seata-server/src/main/java/io/seata/server/session/SessionStatusValidator.java
+++ b/ruoyi-visual/ruoyi-seata-server/src/main/java/io/seata/server/session/SessionStatusValidator.java
@@ -0,0 +1,87 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+package io.seata.server.session;
+
+import io.seata.core.model.GlobalStatus;
+
+/**
+ * The type change status validator.
+ *
+ * @author Bughue
+ */
+public class SessionStatusValidator {
+
+    /**
+     * is timeout global status
+     *
+     * @param status the global session
+     */
+    public static boolean isTimeoutGlobalStatus(GlobalStatus status) {
+        return status == GlobalStatus.TimeoutRollbacked
+                || status == GlobalStatus.TimeoutRollbackFailed
+                || status == GlobalStatus.TimeoutRollbacking
+                || status == GlobalStatus.TimeoutRollbackRetrying;
+    }
+
+    /**
+     * is rollback global status
+     *
+     * @param status the global session
+     */
+    public static boolean isRollbackGlobalStatus(GlobalStatus status) {
+        return status == GlobalStatus.Rollbacking
+                || status == GlobalStatus.RollbackRetrying
+                || status == GlobalStatus.Rollbacked
+                || status == GlobalStatus.RollbackFailed
+                || status == GlobalStatus.RollbackRetryTimeout;
+    }
+
+    /**
+     * is commit global status
+     *
+     * @param status the global session
+     */
+    public static boolean isCommitGlobalStatus(GlobalStatus status) {
+        return status == GlobalStatus.Committing
+                || status == GlobalStatus.AsyncCommitting
+                || status == GlobalStatus.CommitRetrying
+                || status == GlobalStatus.Committed
+                || status == GlobalStatus.CommitFailed
+                || status == GlobalStatus.CommitRetryTimeout;
+    }
+
+    /**
+     * check the relation of before status and after status
+     *
+     * @param before the global session
+     * @param after the global session
+     */
+    public static boolean validateUpdateStatus(GlobalStatus before, GlobalStatus after) {
+        if (isTimeoutGlobalStatus(before) && isCommitGlobalStatus(after)) {
+            return false;
+        }
+        if (isCommitGlobalStatus(before) && isTimeoutGlobalStatus(after)) {
+            return false;
+        }
+        if (isRollbackGlobalStatus(before) && isCommitGlobalStatus(after)) {
+            return false;
+        }
+        if (isCommitGlobalStatus(before) && isRollbackGlobalStatus(after)) {
+            return false;
+        }
+        return true;
+    }
+}