Support RocksDB storage engine in File mode

common/src/main/java/org/apache/seata/common/ConfigurationKeys.java

@@ -204,6 +204,11 @@ public interface ConfigurationKeys {
      */
     String SERVICE_SESSION_RELOAD_READ_SIZE = STORE_FILE_PREFIX + "sessionReloadReadSize";
 
+    /**
+     * The constant STORE_FILE_ENGINE.
+     */
+    String STORE_FILE_ENGINE = STORE_FILE_PREFIX + "engine";
+
     /**
      * The constant CLIENT_REPORT_SUCCESS_ENABLE.
      */

dependencies/pom.xml

@@ -74,6 +74,7 @@
         <ant.version>1.10.12</ant.version>
         <lz4.version>1.10.1</lz4.version>
         <jraft.version>1.3.14</jraft.version>
+        <rocksdb.version>8.8.1</rocksdb.version>
         <snakeyaml.version>2.0</snakeyaml.version>
         <netty.version>4.1.101.Final</netty.version>
         <sofa.hessian.version>4.0.3</sofa.hessian.version>
@@ -790,6 +791,11 @@
                 <artifactId>jraft-core</artifactId>
                 <version>${jraft.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.rocksdb</groupId>
+                <artifactId>rocksdbjni</artifactId>
+                <version>${rocksdb.version}</version>
+            </dependency>
             <dependency>
                 <groupId>com.github.luben</groupId>
                 <artifactId>zstd-jni</artifactId>

server/pom.xml

@@ -282,6 +282,10 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>org.rocksdb</groupId>
+            <artifactId>rocksdbjni</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.alipay.sofa</groupId>
             <artifactId>bolt</artifactId>

server/src/main/java/org/apache/seata/server/storage/file/FileStoreEngine.java

@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+/**
+ * The local file mode storage engine.
+ */
+public enum FileStoreEngine {
+
+    /**
+     * Append-only data file engine.
+     */
+    FILE("file"),
+
+    /**
+     * RocksDB embedded key-value engine.
+     */
+    ROCKSDB("rocksdb");
+
+    private final String name;
+
+    FileStoreEngine(String name) {
+        this.name = name;
+    }
+
+    public static FileStoreEngine get(String name) {
+        for (FileStoreEngine engine : FileStoreEngine.values()) {
+            if (engine.getName().equalsIgnoreCase(name)) {
+                return engine;
+            }
+        }
+        throw new IllegalArgumentException("unknown file store engine:" + name);
+    }
+
+    public String getName() {
+        return name;
+    }
+}

server/src/main/java/org/apache/seata/server/storage/file/session/FileSessionManager.java

@@ -30,11 +30,14 @@
 import org.apache.seata.server.session.GlobalSession;
 import org.apache.seata.server.session.Reloadable;
 import org.apache.seata.server.session.SessionCondition;
+import org.apache.seata.server.storage.file.FileStoreEngine;
 import org.apache.seata.server.storage.file.ReloadableStore;
 import org.apache.seata.server.storage.file.TransactionWriteStore;
 import org.apache.seata.server.storage.file.store.FileTransactionStoreManager;
+import org.apache.seata.server.storage.file.store.RocksDBTransactionStoreManager;
 import org.apache.seata.server.store.AbstractTransactionStoreManager;
 import org.apache.seata.server.store.SessionStorable;
+import org.apache.seata.server.store.StoreConfig;
 import org.apache.seata.server.store.TransactionStoreManager;
 
 import java.io.File;
@@ -62,18 +65,36 @@ public class FileSessionManager extends AbstractSessionManager implements Reload
     private static final int READ_SIZE = ConfigurationFactory.getInstance()
             .getInt(ConfigurationKeys.SERVICE_SESSION_RELOAD_READ_SIZE, DEFAULT_SERVICE_SESSION_RELOAD_READ_SIZE);
 
+    private static final GlobalStatus[] ACTIVE_STATUSES = {
+        GlobalStatus.UnKnown,
+        GlobalStatus.Begin,
+        GlobalStatus.Committing,
+        GlobalStatus.CommitRetrying,
+        GlobalStatus.Rollbacking,
+        GlobalStatus.RollbackRetrying,
+        GlobalStatus.TimeoutRollbacking,
+        GlobalStatus.TimeoutRollbackRetrying,
+        GlobalStatus.AsyncCommitting,
+        GlobalStatus.StopRollbackOrRollbackRetry,
+        GlobalStatus.StopCommitOrCommitRetry,
+        GlobalStatus.Deleting
+    };
+
     /**
      * The Session map.
      */
     protected Map<String, GlobalSession> sessionMap = new ConcurrentHashMap<>(64);
 
+    private final FileStoreEngine fileStoreEngine;
+
     /**
      * Instantiates a new File based session manager.
      *
      * @param name the name
      */
     public FileSessionManager(String name) {
         super(name);
+        fileStoreEngine = FileStoreEngine.FILE;
         transactionStoreManager = new AbstractTransactionStoreManager() {
             @Override
             public boolean writeSession(LogOperation logOperation, SessionStorable session) {
@@ -91,9 +112,15 @@ public boolean writeSession(LogOperation logOperation, SessionStorable session)
      */
     public FileSessionManager(String name, String sessionStoreFilePath) throws IOException {
         super(name);
+        fileStoreEngine = StoreConfig.getFileStoreEngine();
         if (StringUtils.isNotBlank(sessionStoreFilePath)) {
-            transactionStoreManager =
-                    new FileTransactionStoreManager(sessionStoreFilePath + File.separator + name, this);
+            if (fileStoreEngine == FileStoreEngine.ROCKSDB) {
+                transactionStoreManager =
+                        new RocksDBTransactionStoreManager(sessionStoreFilePath + File.separator + name + ".rocksdb");
+            } else {
+                transactionStoreManager =
+                        new FileTransactionStoreManager(sessionStoreFilePath + File.separator + name, this);
+            }
         } else {
             transactionStoreManager = new AbstractTransactionStoreManager() {
                 @Override
@@ -106,11 +133,18 @@ public boolean writeSession(LogOperation logOperation, SessionStorable session)
 
     @Override
     public void reload() {
+        if (isRocksDBStore()) {
+            return;
+        }
         restoreSessions();
     }
 
     @Override
     public void addGlobalSession(GlobalSession session) throws TransactionException {
+        if (isRocksDBStore()) {
+            super.addGlobalSession(session);
+            return;
+        }
         CollectionUtils.computeIfAbsent(sessionMap, session.getXid(), k -> {
             try {
                 super.addGlobalSession(session);
@@ -123,29 +157,45 @@ public void addGlobalSession(GlobalSession session) throws TransactionException
 
     @Override
     public GlobalSession findGlobalSession(String xid) {
+        if (isRocksDBStore()) {
+            return transactionStoreManager.readSession(xid);
+        }
         return sessionMap.get(xid);
     }
 
     @Override
     public GlobalSession findGlobalSession(String xid, boolean withBranchSessions) {
+        if (isRocksDBStore()) {
+            return transactionStoreManager.readSession(xid, withBranchSessions);
+        }
         // withBranchSessions without process in memory
         return sessionMap.get(xid);
     }
 
     @Override
     public void removeGlobalSession(GlobalSession session) throws TransactionException {
+        if (isRocksDBStore()) {
+            super.removeGlobalSession(session);
+            return;
+        }
         if (sessionMap.remove(session.getXid()) != null) {
             super.removeGlobalSession(session);
         }
     }
 
     @Override
     public Collection<GlobalSession> allSessions() {
+        if (isRocksDBStore()) {
+            return findGlobalSessions(new SessionCondition(ACTIVE_STATUSES));
+        }
         return sessionMap.values();
     }
 
     @Override
     public List<GlobalSession> findGlobalSessions(SessionCondition condition) {
+        if (isRocksDBStore()) {
+            return transactionStoreManager.readSession(condition);
+        }
         List<GlobalStatus> globalStatuses = null;
         if (null != condition.getStatuses() && condition.getStatuses().length > 0) {
             globalStatuses = Arrays.asList(condition.getStatuses());
@@ -385,6 +435,10 @@ public void setSessionMap(Map<String, GlobalSession> sessionMap) {
         this.sessionMap = sessionMap;
     }
 
+    private boolean isRocksDBStore() {
+        return fileStoreEngine == FileStoreEngine.ROCKSDB;
+    }
+
     @Override
     public void destroy() {
         transactionStoreManager.shutdown();