You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ib...@apache.org on 2022/07/20 13:50:57 UTC
[ignite-3] branch main updated: IGNITE-17393 Some code rolled back from jraft sources and redone in a less invasive way. (#950)
This is an automated email from the ASF dual-hosted git repository.
ibessonov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 5aa243a0b IGNITE-17393 Some code rolled back from jraft sources and redone in a less invasive way. (#950)
5aa243a0b is described below
commit 5aa243a0b0ed75aae040f713c569609e49de8758
Author: ibessonov <be...@gmail.com>
AuthorDate: Wed Jul 20 16:50:53 2022 +0300
IGNITE-17393 Some code rolled back from jraft sources and redone in a less invasive way. (#950)
---
.../marshal/MarshallingValidations.java | 4 +-
.../apache/ignite/raft/jraft/core/ItNodeTest.java | 43 ++++-------
.../internal/raft/server/RaftGroupOptions.java | 71 ++++++++++++++---
.../internal/raft/server/impl/JraftServerImpl.java | 30 +++++---
.../raft}/storage/LogStorageFactory.java | 9 ++-
.../raft/storage/RaftMetaStorageFactory.java} | 21 +++---
.../raft/storage/SnapshotStorageFactory.java} | 23 +++---
.../storage/impl/DefaultLogStorageFactory.java | 8 +-
.../storage/impl/IgniteJraftServiceFactory.java | 88 ++++++++++++++++++++++
.../storage/impl/RocksDbSharedLogStorage.java | 4 +-
.../storage/impl/VolatileLogStorageFactory.java} | 25 ++++--
.../ignite/raft/jraft/JRaftServiceFactory.java | 14 +---
.../jraft/core/DefaultJRaftServiceFactory.java | 28 ++++---
.../apache/ignite/raft/jraft/core/NodeImpl.java | 3 +-
.../jraft/core/VolatileJRaftServiceFactory.java | 54 -------------
.../ignite/raft/jraft/option/BootstrapOptions.java | 11 +++
.../ignite/raft/jraft/option/NodeOptions.java | 23 ++++--
.../impl/RocksDbSharedLogStorageAdvancedTest.java | 18 ++---
.../storage/impl/RocksDbSharedLogStorageTest.java | 9 ++-
.../impl/SharedVsNonSharedLogStorageBenchmark.java | 11 +--
.../apache/ignite/raft/jraft/core/TestCluster.java | 1 +
.../raft/jraft/core/TestJRaftServiceFactory.java | 7 +-
.../core/VolatileJRaftServiceFactoryTest.java | 40 ----------
.../raft/jraft/storage/SnapshotExecutorTest.java | 6 +-
.../storage/impl/VolatileRaftMetaStorageTest.java | 1 -
.../internal/table/distributed/TableManager.java | 10 ++-
26 files changed, 319 insertions(+), 243 deletions(-)
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/marshal/MarshallingValidations.java b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/marshal/MarshallingValidations.java
index ff043bdd8..09b7e5c03 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/marshal/MarshallingValidations.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/marshal/MarshallingValidations.java
@@ -20,7 +20,7 @@ package org.apache.ignite.internal.network.serialization.marshal;
import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.util.Map;
-import java.util.WeakHashMap;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import org.apache.ignite.internal.network.serialization.Classes;
import org.jetbrains.annotations.Nullable;
@@ -29,7 +29,7 @@ import org.jetbrains.annotations.Nullable;
* Validations that are run before marshalling objects.
*/
class MarshallingValidations {
- private final Map<Class<?>, Marshallability> marshallabilityMap = new WeakHashMap<>();
+ private final Map<Class<?>, Marshallability> marshallabilityMap = new ConcurrentHashMap<>();
private final Function<Class<?>, Marshallability> marshallabilityFunction = this::marshallability;
diff --git a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItNodeTest.java b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItNodeTest.java
index d51c2965c..c9a2492ec 100644
--- a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItNodeTest.java
+++ b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItNodeTest.java
@@ -71,6 +71,8 @@ import java.util.stream.Stream;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.raft.server.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.storage.impl.DefaultLogStorageFactory;
+import org.apache.ignite.internal.raft.storage.impl.IgniteJraftServiceFactory;
import org.apache.ignite.internal.testframework.WorkDirectory;
import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
import org.apache.ignite.network.ClusterService;
@@ -107,9 +109,7 @@ import org.apache.ignite.raft.jraft.rpc.TestIgniteRpcServer;
import org.apache.ignite.raft.jraft.rpc.impl.IgniteRpcClient;
import org.apache.ignite.raft.jraft.rpc.impl.IgniteRpcServer;
import org.apache.ignite.raft.jraft.rpc.impl.core.DefaultRaftClientService;
-import org.apache.ignite.raft.jraft.storage.LogStorage;
import org.apache.ignite.raft.jraft.storage.SnapshotThrottle;
-import org.apache.ignite.raft.jraft.storage.impl.DefaultLogStorageFactory;
import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotReader;
import org.apache.ignite.raft.jraft.storage.snapshot.ThroughputSnapshotThrottle;
import org.apache.ignite.raft.jraft.test.TestUtils;
@@ -2920,16 +2920,13 @@ public class ItNodeTest {
fsm.getLogs().add(ByteBuffer.wrap(new byte[] {(byte) ch}));
BootstrapOptions opts = new BootstrapOptions();
- DefaultLogStorageFactory logStorageProvider = new DefaultLogStorageFactory(path) {
- @Override public LogStorage getLogStorage(String groupId, RaftOptions raftOptions) {
- return super.getLogStorage("test", raftOptions);
- }
- };
+ DefaultLogStorageFactory logStorageProvider = new DefaultLogStorageFactory(path);
logStorageProvider.start();
- opts.setServiceFactory(new DefaultJRaftServiceFactory(logStorageProvider));
+ opts.setServiceFactory(new IgniteJraftServiceFactory(logStorageProvider));
opts.setLastLogIndex(fsm.getLogs().size());
opts.setRaftMetaUri(dataPath + File.separator + "meta");
opts.setSnapshotUri(dataPath + File.separator + "snapshot");
+ opts.setLogUri("test");
opts.setGroupConf(JRaftUtils.getConfiguration("127.0.0.1:5006"));
opts.setFsm(fsm);
@@ -2941,13 +2938,10 @@ public class ItNodeTest {
nodeOpts.setRaftMetaUri(dataPath + File.separator + "meta");
nodeOpts.setSnapshotUri(dataPath + File.separator + "snapshot");
- DefaultLogStorageFactory log2 = new DefaultLogStorageFactory(path) {
- @Override public LogStorage getLogStorage(String groupId, RaftOptions raftOptions) {
- return super.getLogStorage("test", raftOptions);
- }
- };
+ nodeOpts.setLogUri("test");
+ DefaultLogStorageFactory log2 = new DefaultLogStorageFactory(path);
log2.start();
- nodeOpts.setServiceFactory(new DefaultJRaftServiceFactory(log2));
+ nodeOpts.setServiceFactory(new IgniteJraftServiceFactory(log2));
nodeOpts.setFsm(fsm);
RaftGroupService service = createService("test", new PeerId(addr, 0), nodeOpts);
@@ -2974,16 +2968,13 @@ public class ItNodeTest {
Files.createDirectories(path);
BootstrapOptions opts = new BootstrapOptions();
- DefaultLogStorageFactory logStorageProvider = new DefaultLogStorageFactory(path) {
- @Override public LogStorage getLogStorage(String groupId, RaftOptions raftOptions) {
- return super.getLogStorage("test", raftOptions);
- }
- };
+ DefaultLogStorageFactory logStorageProvider = new DefaultLogStorageFactory(path);
logStorageProvider.start();
- opts.setServiceFactory(new DefaultJRaftServiceFactory(logStorageProvider));
+ opts.setServiceFactory(new IgniteJraftServiceFactory(logStorageProvider));
opts.setLastLogIndex(0);
opts.setRaftMetaUri(dataPath + File.separator + "meta");
opts.setSnapshotUri(dataPath + File.separator + "snapshot");
+ opts.setLogUri("test");
opts.setGroupConf(JRaftUtils.getConfiguration("127.0.0.1:5006"));
opts.setFsm(fsm);
NodeOptions nodeOpts = new NodeOptions();
@@ -2994,14 +2985,11 @@ public class ItNodeTest {
nodeOpts.setRaftMetaUri(dataPath + File.separator + "meta");
nodeOpts.setSnapshotUri(dataPath + File.separator + "snapshot");
+ nodeOpts.setLogUri("test");
nodeOpts.setFsm(fsm);
- DefaultLogStorageFactory log2 = new DefaultLogStorageFactory(path) {
- @Override public LogStorage getLogStorage(String groupId, RaftOptions raftOptions) {
- return super.getLogStorage("test", raftOptions);
- }
- };
+ DefaultLogStorageFactory log2 = new DefaultLogStorageFactory(path);
log2.start();
- nodeOpts.setServiceFactory(new DefaultJRaftServiceFactory(log2));
+ nodeOpts.setServiceFactory(new IgniteJraftServiceFactory(log2));
RaftGroupService service = createService("test", new PeerId(addr, 0), nodeOpts);
@@ -3779,7 +3767,8 @@ public class ItNodeTest {
DefaultLogStorageFactory log = new DefaultLogStorageFactory(Path.of(dataPath, "node" + nodeIdx, "log"));
log.start();
- options.setServiceFactory(new DefaultJRaftServiceFactory(log));
+ options.setServiceFactory(new IgniteJraftServiceFactory(log));
+ options.setLogUri("test");
return options;
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/RaftGroupOptions.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/RaftGroupOptions.java
index 4880e3188..0bbc6b735 100644
--- a/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/RaftGroupOptions.java
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/RaftGroupOptions.java
@@ -17,6 +17,10 @@
package org.apache.ignite.internal.raft.server;
+import org.apache.ignite.internal.raft.storage.LogStorageFactory;
+import org.apache.ignite.internal.raft.storage.RaftMetaStorageFactory;
+import org.apache.ignite.internal.raft.storage.SnapshotStorageFactory;
+
/**
* Options specific to a Raft group that is being started.
*/
@@ -24,6 +28,15 @@ public class RaftGroupOptions {
/** Whether volatile stores should be used for the corresponding Raft Group. Classic Raft uses persistent ones. */
private final boolean volatileStores;
+ /** Log storage factory. */
+ private LogStorageFactory logStorageFactory;
+
+ /** Snapshot storage factory. */
+ private SnapshotStorageFactory snapshotStorageFactory;
+
+ /** Raft meta storage factory. */
+ private RaftMetaStorageFactory raftMetaStorageFactory;
+
/**
* Returns default options as defined by classic Raft (so stores are persistent).
*
@@ -51,16 +64,6 @@ public class RaftGroupOptions {
return new RaftGroupOptions(true);
}
- /**
- * Creates options derived from table configuration.
- *
- * @param isVolatile Whether the table is configured as volatile (in-memory) or not.
- * @return Options derived from table configuration.
- */
- public static RaftGroupOptions forTable(boolean isVolatile) {
- return isVolatile ? forVolatileStores() : forPersistentStores();
- }
-
private RaftGroupOptions(boolean volatileStores) {
this.volatileStores = volatileStores;
}
@@ -75,4 +78,52 @@ public class RaftGroupOptions {
public boolean volatileStores() {
return volatileStores;
}
+
+ /**
+ * Returns a log storage factory that's used to create log storage for a raft group.
+ */
+ public LogStorageFactory getLogStorageFactory() {
+ return logStorageFactory;
+ }
+
+ /**
+ * Adds log storage factory to options.
+ */
+ public RaftGroupOptions setLogStorageFactory(LogStorageFactory logStorageFactory) {
+ this.logStorageFactory = logStorageFactory;
+
+ return this;
+ }
+
+ /**
+ * Returns a snapshot storage factory that's used to create snapshot storage for a raft group.
+ */
+ public SnapshotStorageFactory snapshotStorageFactory() {
+ return snapshotStorageFactory;
+ }
+
+ /**
+ * Adds snapshot storage factory to options.
+ */
+ public RaftGroupOptions snapshotStorageFactory(SnapshotStorageFactory snapshotStorageFactory) {
+ this.snapshotStorageFactory = snapshotStorageFactory;
+
+ return this;
+ }
+
+ /**
+ * Returns a raft meta storage factory that's used to create raft meta storage for a raft group.
+ */
+ public RaftMetaStorageFactory raftMetaStorageFactory() {
+ return raftMetaStorageFactory;
+ }
+
+ /**
+ * Adds raft meta storage factory to options.
+ */
+ public RaftGroupOptions raftMetaStorageFactory(RaftMetaStorageFactory raftMetaStorageFactory) {
+ this.raftMetaStorageFactory = raftMetaStorageFactory;
+
+ return this;
+ }
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java
index 438f96729..6081e65a5 100644
--- a/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java
@@ -35,6 +35,9 @@ import java.util.stream.Collectors;
import org.apache.ignite.internal.raft.server.RaftGroupEventsListener;
import org.apache.ignite.internal.raft.server.RaftGroupOptions;
import org.apache.ignite.internal.raft.server.RaftServer;
+import org.apache.ignite.internal.raft.storage.LogStorageFactory;
+import org.apache.ignite.internal.raft.storage.impl.DefaultLogStorageFactory;
+import org.apache.ignite.internal.raft.storage.impl.IgniteJraftServiceFactory;
import org.apache.ignite.internal.thread.NamedThreadFactory;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.lang.IgniteStringFormatter;
@@ -53,20 +56,16 @@ import org.apache.ignite.raft.jraft.NodeManager;
import org.apache.ignite.raft.jraft.RaftGroupService;
import org.apache.ignite.raft.jraft.Status;
import org.apache.ignite.raft.jraft.conf.Configuration;
-import org.apache.ignite.raft.jraft.core.DefaultJRaftServiceFactory;
import org.apache.ignite.raft.jraft.core.FSMCallerImpl;
import org.apache.ignite.raft.jraft.core.NodeImpl;
import org.apache.ignite.raft.jraft.core.ReadOnlyServiceImpl;
import org.apache.ignite.raft.jraft.core.StateMachineAdapter;
-import org.apache.ignite.raft.jraft.core.VolatileJRaftServiceFactory;
import org.apache.ignite.raft.jraft.disruptor.StripedDisruptor;
import org.apache.ignite.raft.jraft.entity.PeerId;
import org.apache.ignite.raft.jraft.error.RaftError;
import org.apache.ignite.raft.jraft.option.NodeOptions;
import org.apache.ignite.raft.jraft.rpc.impl.IgniteRpcClient;
import org.apache.ignite.raft.jraft.rpc.impl.IgniteRpcServer;
-import org.apache.ignite.raft.jraft.storage.LogStorageFactory;
-import org.apache.ignite.raft.jraft.storage.impl.DefaultLogStorageFactory;
import org.apache.ignite.raft.jraft.storage.impl.LogManagerImpl;
import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotReader;
import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotWriter;
@@ -132,7 +131,6 @@ public class JraftServerImpl implements RaftServer {
this.opts.setRpcConnectTimeoutMs(this.opts.getElectionTimeoutMs() / 3);
this.opts.setRpcDefaultTimeout(this.opts.getElectionTimeoutMs() / 2);
this.opts.setSharedPools(true);
- this.opts.setServiceFactory(new DefaultJRaftServiceFactory(logStorageFactory));
if (opts.getServerName() == null) {
this.opts.setServerName(service.localConfiguration().getName());
@@ -351,19 +349,31 @@ public class JraftServerImpl implements RaftServer {
throw new IgniteInternalException(e);
}
- if (!groupOptions.volatileStores()) {
- nodeOptions.setRaftMetaUri(serverDataPath.resolve("meta").toString());
- }
+ nodeOptions.setLogUri(groupId);
+
+ nodeOptions.setRaftMetaUri(serverDataPath.resolve("meta").toString());
+
nodeOptions.setSnapshotUri(serverDataPath.resolve("snapshot").toString());
nodeOptions.setFsm(new DelegatingStateMachine(lsnr));
nodeOptions.setRaftGrpEvtsLsnr(evLsnr);
- if (groupOptions.volatileStores()) {
- nodeOptions.setServiceFactory(new VolatileJRaftServiceFactory());
+ LogStorageFactory logStorageFactory = groupOptions.getLogStorageFactory() == null
+ ? this.logStorageFactory : groupOptions.getLogStorageFactory();
+
+ IgniteJraftServiceFactory serviceFactory = new IgniteJraftServiceFactory(logStorageFactory);
+
+ if (groupOptions.snapshotStorageFactory() != null) {
+ serviceFactory.setSnapshotStorageFactory(groupOptions.snapshotStorageFactory());
+ }
+
+ if (groupOptions.raftMetaStorageFactory() != null) {
+ serviceFactory.setRaftMetaStorageFactory(groupOptions.raftMetaStorageFactory());
}
+ nodeOptions.setServiceFactory(serviceFactory);
+
if (initialConf != null) {
List<PeerId> mapped = initialConf.stream().map(PeerId::fromPeer).collect(Collectors.toList());
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/LogStorageFactory.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/LogStorageFactory.java
similarity index 79%
copy from modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/LogStorageFactory.java
copy to modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/LogStorageFactory.java
index c373c66ca..405b28f7d 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/LogStorageFactory.java
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/LogStorageFactory.java
@@ -1,6 +1,6 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
+ * 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
@@ -15,9 +15,10 @@
* limitations under the License.
*/
-package org.apache.ignite.raft.jraft.storage;
+package org.apache.ignite.internal.raft.storage;
import org.apache.ignite.raft.jraft.option.RaftOptions;
+import org.apache.ignite.raft.jraft.storage.LogStorage;
/** Log storage factory interface. */
public interface LogStorageFactory extends AutoCloseable {
@@ -29,9 +30,9 @@ public interface LogStorageFactory extends AutoCloseable {
/**
* Creates a log storage.
*
- * @param groupId Raft group id.
+ * @param uri Log storage URI.
* @param raftOptions Raft options.
* @return Log storage.
*/
- LogStorage getLogStorage(String groupId, RaftOptions raftOptions);
+ LogStorage createLogStorage(String uri, RaftOptions raftOptions);
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/LogStorageFactory.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/RaftMetaStorageFactory.java
similarity index 66%
copy from modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/LogStorageFactory.java
copy to modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/RaftMetaStorageFactory.java
index c373c66ca..9b23b559a 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/LogStorageFactory.java
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/RaftMetaStorageFactory.java
@@ -1,6 +1,6 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
+ * 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
@@ -15,23 +15,20 @@
* limitations under the License.
*/
-package org.apache.ignite.raft.jraft.storage;
+package org.apache.ignite.internal.raft.storage;
import org.apache.ignite.raft.jraft.option.RaftOptions;
+import org.apache.ignite.raft.jraft.storage.RaftMetaStorage;
-/** Log storage factory interface. */
-public interface LogStorageFactory extends AutoCloseable {
+/** Raft meta storage factory interface. */
+@FunctionalInterface
+public interface RaftMetaStorageFactory {
/**
- * Starts the log storage factory.
- */
- void start();
-
- /**
- * Creates a log storage.
+ * Creates a raft meta storage.
*
* @param groupId Raft group id.
* @param raftOptions Raft options.
- * @return Log storage.
+ * @return Raft meta storage.
*/
- LogStorage getLogStorage(String groupId, RaftOptions raftOptions);
+ RaftMetaStorage createRaftMetaStorage(String groupId, RaftOptions raftOptions);
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/LogStorageFactory.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/SnapshotStorageFactory.java
similarity index 64%
rename from modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/LogStorageFactory.java
rename to modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/SnapshotStorageFactory.java
index c373c66ca..c2e69298a 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/LogStorageFactory.java
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/SnapshotStorageFactory.java
@@ -1,6 +1,6 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
+ * 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
@@ -15,23 +15,20 @@
* limitations under the License.
*/
-package org.apache.ignite.raft.jraft.storage;
+package org.apache.ignite.internal.raft.storage;
import org.apache.ignite.raft.jraft.option.RaftOptions;
+import org.apache.ignite.raft.jraft.storage.SnapshotStorage;
-/** Log storage factory interface. */
-public interface LogStorageFactory extends AutoCloseable {
+/** Snapshot storage factory interface. */
+@FunctionalInterface
+public interface SnapshotStorageFactory {
/**
- * Starts the log storage factory.
- */
- void start();
-
- /**
- * Creates a log storage.
+ * Creates a snapshot storage.
*
- * @param groupId Raft group id.
+ * @param uri Snapshot URI.
* @param raftOptions Raft options.
- * @return Log storage.
+ * @return Snapshot storage.
*/
- LogStorage getLogStorage(String groupId, RaftOptions raftOptions);
+ SnapshotStorage createSnapshotStorage(String uri, RaftOptions raftOptions);
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/impl/DefaultLogStorageFactory.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/impl/DefaultLogStorageFactory.java
similarity index 95%
rename from modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/impl/DefaultLogStorageFactory.java
rename to modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/impl/DefaultLogStorageFactory.java
index 229a729df..4fcb044fa 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/impl/DefaultLogStorageFactory.java
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/impl/DefaultLogStorageFactory.java
@@ -1,6 +1,6 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
+ * 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
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.ignite.raft.jraft.storage.impl;
+package org.apache.ignite.internal.raft.storage.impl;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.rocksdb.RocksDB.DEFAULT_COLUMN_FAMILY;
@@ -29,11 +29,11 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.raft.storage.LogStorageFactory;
import org.apache.ignite.internal.thread.NamedThreadFactory;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.raft.jraft.option.RaftOptions;
import org.apache.ignite.raft.jraft.storage.LogStorage;
-import org.apache.ignite.raft.jraft.storage.LogStorageFactory;
import org.apache.ignite.raft.jraft.util.ExecutorServiceHelper;
import org.apache.ignite.raft.jraft.util.Platform;
import org.rocksdb.ColumnFamilyDescriptor;
@@ -135,7 +135,7 @@ public class DefaultLogStorageFactory implements LogStorageFactory {
/** {@inheritDoc} */
@Override
- public LogStorage getLogStorage(String groupId, RaftOptions raftOptions) {
+ public LogStorage createLogStorage(String groupId, RaftOptions raftOptions) {
return new RocksDbSharedLogStorage(db, confHandle, dataHandle, groupId, raftOptions, executorService);
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/impl/IgniteJraftServiceFactory.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/impl/IgniteJraftServiceFactory.java
new file mode 100644
index 000000000..c27d2125a
--- /dev/null
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/impl/IgniteJraftServiceFactory.java
@@ -0,0 +1,88 @@
+/*
+ * 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.ignite.internal.raft.storage.impl;
+
+import org.apache.ignite.internal.raft.storage.LogStorageFactory;
+import org.apache.ignite.internal.raft.storage.RaftMetaStorageFactory;
+import org.apache.ignite.internal.raft.storage.SnapshotStorageFactory;
+import org.apache.ignite.raft.jraft.core.DefaultJRaftServiceFactory;
+import org.apache.ignite.raft.jraft.option.RaftOptions;
+import org.apache.ignite.raft.jraft.storage.LogStorage;
+import org.apache.ignite.raft.jraft.storage.RaftMetaStorage;
+import org.apache.ignite.raft.jraft.storage.SnapshotStorage;
+import org.apache.ignite.raft.jraft.storage.impl.LocalRaftMetaStorage;
+import org.apache.ignite.raft.jraft.storage.snapshot.local.LocalSnapshotStorage;
+import org.apache.ignite.raft.jraft.util.Requires;
+import org.apache.ignite.raft.jraft.util.StringUtils;
+
+/**
+ * The default factory for JRaft services for Ignite.
+ */
+public class IgniteJraftServiceFactory extends DefaultJRaftServiceFactory {
+ /** Log storage factory .*/
+ private final LogStorageFactory logStorageFactory;
+
+ /** Snapshot storage factory. */
+ private volatile SnapshotStorageFactory snapshotStorageFactory = LocalSnapshotStorage::new;
+
+ /** Raft meta storage factory. */
+ private volatile RaftMetaStorageFactory raftMetaStorageFactory = LocalRaftMetaStorage::new;
+
+ public IgniteJraftServiceFactory(LogStorageFactory factory) {
+ logStorageFactory = factory;
+ }
+
+ /**
+ * Updates a value of {@link SnapshotStorageFactory} in this service factory.
+ */
+ public IgniteJraftServiceFactory setSnapshotStorageFactory(SnapshotStorageFactory snapshotStorageFactory) {
+ this.snapshotStorageFactory = snapshotStorageFactory;
+
+ return this;
+ }
+
+ /**
+ * Updates a value of {@link RaftMetaStorageFactory} in this service factory.
+ */
+ public IgniteJraftServiceFactory setRaftMetaStorageFactory(RaftMetaStorageFactory raftMetaStorageFactory) {
+ this.raftMetaStorageFactory = raftMetaStorageFactory;
+
+ return this;
+ }
+
+ @Override
+ public LogStorage createLogStorage(final String groupId, final RaftOptions raftOptions) {
+ Requires.requireTrue(StringUtils.isNotBlank(groupId), "Blank group id.");
+
+ return logStorageFactory.createLogStorage(groupId, raftOptions);
+ }
+
+ @Override
+ public SnapshotStorage createSnapshotStorage(final String uri, final RaftOptions raftOptions) {
+ Requires.requireTrue(!StringUtils.isBlank(uri), "Blank snapshot storage uri.");
+
+ return snapshotStorageFactory.createSnapshotStorage(uri, raftOptions);
+ }
+
+ @Override
+ public RaftMetaStorage createRaftMetaStorage(final String uri, final RaftOptions raftOptions) {
+ Requires.requireTrue(!StringUtils.isBlank(uri), "Blank raft meta storage uri.");
+
+ return raftMetaStorageFactory.createRaftMetaStorage(uri, raftOptions);
+ }
+}
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/impl/RocksDbSharedLogStorage.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/impl/RocksDbSharedLogStorage.java
similarity index 99%
rename from modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/impl/RocksDbSharedLogStorage.java
rename to modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/impl/RocksDbSharedLogStorage.java
index fa81057b8..df1f829dc 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/impl/RocksDbSharedLogStorage.java
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/impl/RocksDbSharedLogStorage.java
@@ -1,6 +1,6 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
+ * 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
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.ignite.raft.jraft.storage.impl;
+package org.apache.ignite.internal.raft.storage.impl;
import static java.util.Arrays.copyOfRange;
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/TestJRaftServiceFactory.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/impl/VolatileLogStorageFactory.java
similarity index 59%
copy from modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/TestJRaftServiceFactory.java
copy to modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/impl/VolatileLogStorageFactory.java
index 4766cd419..efaf709f9 100644
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/TestJRaftServiceFactory.java
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/impl/VolatileLogStorageFactory.java
@@ -1,12 +1,12 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
+ * 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
+ * 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,
@@ -14,20 +14,31 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.ignite.raft.jraft.core;
+package org.apache.ignite.internal.raft.storage.impl;
+
+import org.apache.ignite.internal.raft.storage.LogStorageFactory;
import org.apache.ignite.raft.jraft.option.RaftOptions;
import org.apache.ignite.raft.jraft.storage.LogStorage;
import org.apache.ignite.raft.jraft.storage.impl.LocalLogStorage;
-public class TestJRaftServiceFactory extends DefaultJRaftServiceFactory {
- public TestJRaftServiceFactory() {
- super(null);
+/**
+ * Log storage factory based on {@link LocalLogStorage}.
+ */
+public class VolatileLogStorageFactory implements LogStorageFactory {
+ /** {@inheritDoc} */
+ @Override
+ public void start() {
}
+ /** {@inheritDoc} */
@Override
- public LogStorage createLogStorage(final String groupId, final RaftOptions raftOptions) {
+ public LogStorage createLogStorage(String uri, RaftOptions raftOptions) {
return new LocalLogStorage(raftOptions);
}
+ /** {@inheritDoc} */
+ @Override
+ public void close() throws Exception {
+ }
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/JRaftServiceFactory.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/JRaftServiceFactory.java
index 5ec6b4efc..a6aac3874 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/JRaftServiceFactory.java
+++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/JRaftServiceFactory.java
@@ -17,13 +17,11 @@
package org.apache.ignite.raft.jraft;
import org.apache.ignite.raft.jraft.entity.codec.LogEntryCodecFactory;
-import org.apache.ignite.raft.jraft.entity.codec.v1.LogEntryV1CodecFactory;
import org.apache.ignite.raft.jraft.option.NodeOptions;
import org.apache.ignite.raft.jraft.option.RaftOptions;
import org.apache.ignite.raft.jraft.storage.LogStorage;
import org.apache.ignite.raft.jraft.storage.RaftMetaStorage;
import org.apache.ignite.raft.jraft.storage.SnapshotStorage;
-import org.apache.ignite.raft.jraft.util.timer.DefaultRaftTimerFactory;
import org.apache.ignite.raft.jraft.util.timer.RaftTimerFactory;
/**
@@ -33,11 +31,11 @@ public interface JRaftServiceFactory {
/**
* Creates a raft log storage.
*
- * @param groupId The name of the raft group.
+ * @param uri The log storage uri from {@link NodeOptions#getLogUri()}
* @param raftOptions the raft options.
* @return storage to store raft log entires.
*/
- LogStorage createLogStorage(final String groupId, final RaftOptions raftOptions);
+ LogStorage createLogStorage(final String uri, final RaftOptions raftOptions);
/**
* Creates a raft snapshot storage
@@ -62,16 +60,12 @@ public interface JRaftServiceFactory {
*
* @return a codec factory to create encoder/decoder for raft log entry.
*/
- default LogEntryCodecFactory createLogEntryCodecFactory() {
- return LogEntryV1CodecFactory.getInstance();
- }
+ LogEntryCodecFactory createLogEntryCodecFactory();
/**
* Creates raft timer factory.
*
* @return The factory.
*/
- default RaftTimerFactory createRaftTimerFactory() {
- return new DefaultRaftTimerFactory();
- }
+ RaftTimerFactory createRaftTimerFactory();
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/DefaultJRaftServiceFactory.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/DefaultJRaftServiceFactory.java
index c6c71bcf5..92a3a84d7 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/DefaultJRaftServiceFactory.java
+++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/DefaultJRaftServiceFactory.java
@@ -17,31 +17,27 @@
package org.apache.ignite.raft.jraft.core;
import org.apache.ignite.raft.jraft.JRaftServiceFactory;
+import org.apache.ignite.raft.jraft.entity.codec.LogEntryCodecFactory;
+import org.apache.ignite.raft.jraft.entity.codec.v1.LogEntryV1CodecFactory;
import org.apache.ignite.raft.jraft.option.RaftOptions;
import org.apache.ignite.raft.jraft.storage.LogStorage;
-import org.apache.ignite.raft.jraft.storage.LogStorageFactory;
import org.apache.ignite.raft.jraft.storage.RaftMetaStorage;
import org.apache.ignite.raft.jraft.storage.SnapshotStorage;
import org.apache.ignite.raft.jraft.storage.impl.LocalRaftMetaStorage;
+import org.apache.ignite.raft.jraft.storage.impl.RocksDBLogStorage;
import org.apache.ignite.raft.jraft.storage.snapshot.local.LocalSnapshotStorage;
import org.apache.ignite.raft.jraft.util.Requires;
import org.apache.ignite.raft.jraft.util.StringUtils;
+import org.apache.ignite.raft.jraft.util.timer.DefaultRaftTimerFactory;
+import org.apache.ignite.raft.jraft.util.timer.RaftTimerFactory;
/**
* The default factory for JRaft services.
*/
public class DefaultJRaftServiceFactory implements JRaftServiceFactory {
-
- private final LogStorageFactory logStorageFactory;
-
- public DefaultJRaftServiceFactory(LogStorageFactory factory) {
- logStorageFactory = factory;
- }
-
- @Override public LogStorage createLogStorage(final String groupId, final RaftOptions raftOptions) {
- Requires.requireTrue(StringUtils.isNotBlank(groupId), "Blank group id.");
-
- return logStorageFactory.getLogStorage(groupId, raftOptions);
+ @Override public LogStorage createLogStorage(final String uri, final RaftOptions raftOptions) {
+ Requires.requireTrue(StringUtils.isNotBlank(uri), "Blank log storage uri.");
+ return new RocksDBLogStorage(uri, raftOptions);
}
@Override public SnapshotStorage createSnapshotStorage(final String uri, final RaftOptions raftOptions) {
@@ -53,4 +49,12 @@ public class DefaultJRaftServiceFactory implements JRaftServiceFactory {
Requires.requireTrue(!StringUtils.isBlank(uri), "Blank raft meta storage uri.");
return new LocalRaftMetaStorage(uri, raftOptions);
}
+
+ @Override public LogEntryCodecFactory createLogEntryCodecFactory() {
+ return LogEntryV1CodecFactory.getInstance();
+ }
+
+ @Override public RaftTimerFactory createRaftTimerFactory() {
+ return new DefaultRaftTimerFactory();
+ }
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/NodeImpl.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/NodeImpl.java
index 8d71ff43b..c7565cb03 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/NodeImpl.java
+++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/NodeImpl.java
@@ -564,7 +564,7 @@ public class NodeImpl implements Node, RaftServerService {
private boolean initLogStorage() {
Requires.requireNonNull(this.fsmCaller, "Null fsm caller");
- this.logStorage = this.serviceFactory.createLogStorage(groupId, this.raftOptions);
+ this.logStorage = this.serviceFactory.createLogStorage(this.options.getLogUri(), this.raftOptions);
this.logManager = new LogManagerImpl();
final LogManagerOptions opts = new LogManagerOptions();
opts.setGroupId(groupId);
@@ -845,6 +845,7 @@ public class NodeImpl implements Node, RaftServerService {
this.raftOptions = this.options.getRaftOptions();
this.metrics = new NodeMetrics(opts.isEnableMetrics());
this.options.setFsm(opts.getFsm());
+ this.options.setLogUri(opts.getLogUri());
this.options.setRaftMetaUri(opts.getRaftMetaUri());
this.options.setSnapshotUri(opts.getSnapshotUri());
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/VolatileJRaftServiceFactory.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/VolatileJRaftServiceFactory.java
deleted file mode 100644
index 6eb22616e..000000000
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/VolatileJRaftServiceFactory.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.ignite.raft.jraft.core;
-
-import org.apache.ignite.raft.jraft.JRaftServiceFactory;
-import org.apache.ignite.raft.jraft.option.RaftOptions;
-import org.apache.ignite.raft.jraft.storage.LogStorage;
-import org.apache.ignite.raft.jraft.storage.RaftMetaStorage;
-import org.apache.ignite.raft.jraft.storage.SnapshotStorage;
-import org.apache.ignite.raft.jraft.storage.impl.LocalLogStorage;
-import org.apache.ignite.raft.jraft.storage.impl.VolatileRaftMetaStorage;
-import org.apache.ignite.raft.jraft.storage.snapshot.local.LocalSnapshotStorage;
-import org.apache.ignite.raft.jraft.util.Requires;
-import org.apache.ignite.raft.jraft.util.StringUtils;
-
-/**
- * The factory for JRaft services producing volatile stores. Useful for Raft groups hosting partitions of in-memory tables.
- */
-public class VolatileJRaftServiceFactory implements JRaftServiceFactory {
- @Override
- public LogStorage createLogStorage(final String groupId, final RaftOptions raftOptions) {
- Requires.requireTrue(StringUtils.isNotBlank(groupId), "Blank group id.");
-
- return new LocalLogStorage(raftOptions);
- }
-
- @Override
- public SnapshotStorage createSnapshotStorage(final String uri, final RaftOptions raftOptions) {
- Requires.requireTrue(!StringUtils.isBlank(uri), "Blank snapshot storage uri.");
-
- // TODO: IGNITE-17083 - return an in-memory store here (or get rid of SnapshotStorage)
-
- return new LocalSnapshotStorage(uri, raftOptions);
- }
-
- @Override
- public RaftMetaStorage createRaftMetaStorage(final String uri, final RaftOptions raftOptions) {
- return new VolatileRaftMetaStorage();
- }
-}
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/option/BootstrapOptions.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/option/BootstrapOptions.java
index dc1620a9f..81438c5c0 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/option/BootstrapOptions.java
+++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/option/BootstrapOptions.java
@@ -37,6 +37,9 @@ public class BootstrapOptions {
// Default: NULL
private StateMachine fsm;
+ // Describe a specific LogStorage in format ${type}://${parameters}
+ private String logUri;
+
// Describe a specific RaftMetaStorage in format ${type}://${parameters}
private String raftMetaUri;
@@ -96,6 +99,14 @@ public class BootstrapOptions {
this.fsm = fsm;
}
+ public String getLogUri() {
+ return this.logUri;
+ }
+
+ public void setLogUri(String logUri) {
+ this.logUri = logUri;
+ }
+
public String getRaftMetaUri() {
return this.raftMetaUri;
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/option/NodeOptions.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/option/NodeOptions.java
index 31e151746..e0ad684d4 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/option/NodeOptions.java
+++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/option/NodeOptions.java
@@ -19,8 +19,6 @@ package org.apache.ignite.raft.jraft.option;
import java.util.List;
import java.util.concurrent.ExecutorService;
import org.apache.ignite.internal.raft.server.RaftGroupEventsListener;
-import org.apache.ignite.raft.jraft.util.TimeoutStrategy;
-import org.apache.ignite.raft.jraft.util.NoopTimeoutStrategy;
import org.apache.ignite.raft.jraft.JRaftServiceFactory;
import org.apache.ignite.raft.jraft.StateMachine;
import org.apache.ignite.raft.jraft.conf.Configuration;
@@ -34,7 +32,9 @@ import org.apache.ignite.raft.jraft.disruptor.StripedDisruptor;
import org.apache.ignite.raft.jraft.storage.SnapshotThrottle;
import org.apache.ignite.raft.jraft.storage.impl.LogManagerImpl;
import org.apache.ignite.raft.jraft.util.Copiable;
+import org.apache.ignite.raft.jraft.util.NoopTimeoutStrategy;
import org.apache.ignite.raft.jraft.util.StringUtils;
+import org.apache.ignite.raft.jraft.util.TimeoutStrategy;
import org.apache.ignite.raft.jraft.util.Utils;
import org.apache.ignite.raft.jraft.util.concurrent.FixedThreadsExecutorGroup;
import org.apache.ignite.raft.jraft.util.timer.Timer;
@@ -109,6 +109,9 @@ public class NodeOptions extends RpcOptions implements Copiable<NodeOptions> {
// Listener for raft group reconfiguration events.
private RaftGroupEventsListener raftGrpEvtsLsnr;
+ // Describe a specific LogStorage in format ${type}://${parameters}
+ private String logUri;
+
// Describe a specific RaftMetaStorage in format ${type}://${parameters}
private String raftMetaUri;
@@ -277,8 +280,6 @@ public class NodeOptions extends RpcOptions implements Copiable<NodeOptions> {
* Service factory.
*/
public JRaftServiceFactory getServiceFactory() {
- assert this.serviceFactory != null;
-
return this.serviceFactory;
}
@@ -348,6 +349,9 @@ public class NodeOptions extends RpcOptions implements Copiable<NodeOptions> {
}
public void validate() {
+ if (StringUtils.isBlank(this.logUri)) {
+ throw new IllegalArgumentException("Blank logUri");
+ }
if (StringUtils.isBlank(this.raftMetaUri)) {
throw new IllegalArgumentException("Blank raftMetaUri");
}
@@ -445,6 +449,14 @@ public class NodeOptions extends RpcOptions implements Copiable<NodeOptions> {
this.fsm = fsm;
}
+ public String getLogUri() {
+ return this.logUri;
+ }
+
+ public void setLogUri(final String logUri) {
+ this.logUri = logUri;
+ }
+
public String getRaftMetaUri() {
return this.raftMetaUri;
}
@@ -608,7 +620,6 @@ public class NodeOptions extends RpcOptions implements Copiable<NodeOptions> {
nodeOptions.setRpcDefaultTimeout(this.getRpcDefaultTimeout());
nodeOptions.setRpcConnectTimeoutMs(this.getRpcConnectTimeoutMs());
nodeOptions.setElectionTimeoutStrategy(this.getElectionTimeoutStrategy());
- nodeOptions.setServiceFactory(this.getServiceFactory());
return nodeOptions;
}
@@ -619,7 +630,7 @@ public class NodeOptions extends RpcOptions implements Copiable<NodeOptions> {
+ ", decayPriorityGap=" + decayPriorityGap + ", leaderLeaseTimeRatio=" + leaderLeaseTimeRatio
+ ", snapshotIntervalSecs=" + snapshotIntervalSecs + ", snapshotLogIndexMargin="
+ snapshotLogIndexMargin + ", catchupMargin=" + catchupMargin + ", initialConf=" + initialConf
- + ", fsm=" + fsm + ", raftMetaUri='" + raftMetaUri + '\''
+ + ", fsm=" + fsm + ", logUri='" + logUri + '\'' + ", raftMetaUri='" + raftMetaUri + '\''
+ ", snapshotUri='" + snapshotUri + '\'' + ", filterBeforeCopyRemote=" + filterBeforeCopyRemote
+ ", disableCli=" + disableCli + ", timerPoolSize="
+ timerPoolSize + ", cliRpcThreadPoolSize=" + cliRpcThreadPoolSize + ", raftRpcThreadPoolSize="
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/impl/RocksDbSharedLogStorageAdvancedTest.java b/modules/raft/src/test/java/org/apache/ignite/internal/raft/storage/impl/RocksDbSharedLogStorageAdvancedTest.java
similarity index 86%
rename from modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/impl/RocksDbSharedLogStorageAdvancedTest.java
rename to modules/raft/src/test/java/org/apache/ignite/internal/raft/storage/impl/RocksDbSharedLogStorageAdvancedTest.java
index 46b98dcea..2cdaa049d 100644
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/impl/RocksDbSharedLogStorageAdvancedTest.java
+++ b/modules/raft/src/test/java/org/apache/ignite/internal/raft/storage/impl/RocksDbSharedLogStorageAdvancedTest.java
@@ -1,12 +1,12 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
+ * 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
+ * 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,
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.ignite.raft.jraft.storage.impl;
+package org.apache.ignite.internal.raft.storage.impl;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -71,10 +71,10 @@ public class RocksDbSharedLogStorageAdvancedTest {
@Test
public void raftGroupsWithPrefixCollision() {
- LogStorage abcdStorage = logStorageProvider.getLogStorage("abcd", new RaftOptions());
+ LogStorage abcdStorage = logStorageProvider.createLogStorage("abcd", new RaftOptions());
abcdStorage.init(logStorageOptions);
- LogStorage abStorage = logStorageProvider.getLogStorage("ab", new RaftOptions());
+ LogStorage abStorage = logStorageProvider.createLogStorage("ab", new RaftOptions());
abStorage.init(logStorageOptions);
int count = 100;
@@ -120,12 +120,12 @@ public class RocksDbSharedLogStorageAdvancedTest {
@Test
public void testCollisionWithEmptyAndNotEmptyStorage() {
- LogStorage testStorage1 = logStorageProvider.getLogStorage("test1", new RaftOptions());
+ LogStorage testStorage1 = logStorageProvider.createLogStorage("test1", new RaftOptions());
testStorage1.init(logStorageOptions);
testStorage1.appendEntry(TestUtils.mockEntry(1, 1, 1));
- LogStorage testStorage2 = logStorageProvider.getLogStorage("test2", new RaftOptions());
+ LogStorage testStorage2 = logStorageProvider.createLogStorage("test2", new RaftOptions());
testStorage2.init(logStorageOptions);
assertEquals(0, testStorage2.getLastLogIndex());
@@ -138,12 +138,12 @@ public class RocksDbSharedLogStorageAdvancedTest {
public void testIncorrectRaftGroupName() {
assertThrows(
IllegalArgumentException.class,
- () -> logStorageProvider.getLogStorage("name" + ((char) 0), new RaftOptions())
+ () -> logStorageProvider.createLogStorage("name" + ((char) 0), new RaftOptions())
);
assertThrows(
IllegalArgumentException.class,
- () -> logStorageProvider.getLogStorage("name" + ((char) 1), new RaftOptions())
+ () -> logStorageProvider.createLogStorage("name" + ((char) 1), new RaftOptions())
);
}
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/impl/RocksDbSharedLogStorageTest.java b/modules/raft/src/test/java/org/apache/ignite/internal/raft/storage/impl/RocksDbSharedLogStorageTest.java
similarity index 82%
rename from modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/impl/RocksDbSharedLogStorageTest.java
rename to modules/raft/src/test/java/org/apache/ignite/internal/raft/storage/impl/RocksDbSharedLogStorageTest.java
index 45640f3fc..183c4d76a 100644
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/impl/RocksDbSharedLogStorageTest.java
+++ b/modules/raft/src/test/java/org/apache/ignite/internal/raft/storage/impl/RocksDbSharedLogStorageTest.java
@@ -1,12 +1,12 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
+ * 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
+ * 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,
@@ -15,10 +15,11 @@
* limitations under the License.
*/
-package org.apache.ignite.raft.jraft.storage.impl;
+package org.apache.ignite.internal.raft.storage.impl;
import org.apache.ignite.raft.jraft.option.RaftOptions;
import org.apache.ignite.raft.jraft.storage.LogStorage;
+import org.apache.ignite.raft.jraft.storage.impl.BaseLogStorageTest;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -50,6 +51,6 @@ public class RocksDbSharedLogStorageTest extends BaseLogStorageTest {
/** {@inheritDoc} */
@Override
protected LogStorage newLogStorage() {
- return logStorageProvider.getLogStorage("test", new RaftOptions());
+ return logStorageProvider.createLogStorage("test", new RaftOptions());
}
}
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/impl/SharedVsNonSharedLogStorageBenchmark.java b/modules/raft/src/test/java/org/apache/ignite/internal/raft/storage/impl/SharedVsNonSharedLogStorageBenchmark.java
similarity index 95%
rename from modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/impl/SharedVsNonSharedLogStorageBenchmark.java
rename to modules/raft/src/test/java/org/apache/ignite/internal/raft/storage/impl/SharedVsNonSharedLogStorageBenchmark.java
index b163e17a9..8a60616a0 100644
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/impl/SharedVsNonSharedLogStorageBenchmark.java
+++ b/modules/raft/src/test/java/org/apache/ignite/internal/raft/storage/impl/SharedVsNonSharedLogStorageBenchmark.java
@@ -1,12 +1,12 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
+ * 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
+ * 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,
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.ignite.raft.jraft.storage.impl;
+package org.apache.ignite.internal.raft.storage.impl;
import static java.util.stream.Collectors.toList;
@@ -30,6 +30,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;
+import org.apache.ignite.internal.raft.storage.LogStorageFactory;
import org.apache.ignite.raft.jraft.Lifecycle;
import org.apache.ignite.raft.jraft.conf.ConfigurationManager;
import org.apache.ignite.raft.jraft.entity.LogEntry;
@@ -37,7 +38,7 @@ import org.apache.ignite.raft.jraft.entity.codec.v1.LogEntryV1CodecFactory;
import org.apache.ignite.raft.jraft.option.LogStorageOptions;
import org.apache.ignite.raft.jraft.option.RaftOptions;
import org.apache.ignite.raft.jraft.storage.LogStorage;
-import org.apache.ignite.raft.jraft.storage.LogStorageFactory;
+import org.apache.ignite.raft.jraft.storage.impl.RocksDBLogStorage;
import org.apache.ignite.raft.jraft.test.TestUtils;
import org.apache.ignite.raft.jraft.util.Utils;
@@ -192,7 +193,7 @@ public class SharedVsNonSharedLogStorageBenchmark {
List<LogStorage> sharedStorages = grps.stream()
.map(grp -> {
- LogStorage storage = provider.getLogStorage(grp, new RaftOptions());
+ LogStorage storage = provider.createLogStorage(grp, new RaftOptions());
LogStorageOptions opts = new LogStorageOptions();
opts.setConfigurationManager(new ConfigurationManager());
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/TestCluster.java b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/TestCluster.java
index 89412c3bd..5ebf46b67 100644
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/TestCluster.java
+++ b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/TestCluster.java
@@ -224,6 +224,7 @@ public class TestCluster {
}
String serverDataPath = this.dataPath + File.separator + listenAddr.toString().replace(':', '_');
new File(serverDataPath).mkdirs();
+ nodeOptions.setLogUri(serverDataPath + File.separator + "logs");
nodeOptions.setRaftMetaUri(serverDataPath + File.separator + "meta");
nodeOptions.setSnapshotUri(serverDataPath + File.separator + "snapshot");
nodeOptions.setElectionPriority(priority);
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/TestJRaftServiceFactory.java b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/TestJRaftServiceFactory.java
index 4766cd419..cef1c80f0 100644
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/TestJRaftServiceFactory.java
+++ b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/TestJRaftServiceFactory.java
@@ -21,13 +21,8 @@ import org.apache.ignite.raft.jraft.storage.LogStorage;
import org.apache.ignite.raft.jraft.storage.impl.LocalLogStorage;
public class TestJRaftServiceFactory extends DefaultJRaftServiceFactory {
- public TestJRaftServiceFactory() {
- super(null);
- }
-
@Override
- public LogStorage createLogStorage(final String groupId, final RaftOptions raftOptions) {
+ public LogStorage createLogStorage(final String uri, final RaftOptions raftOptions) {
return new LocalLogStorage(raftOptions);
}
-
}
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/VolatileJRaftServiceFactoryTest.java b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/VolatileJRaftServiceFactoryTest.java
deleted file mode 100644
index da74083e8..000000000
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/VolatileJRaftServiceFactoryTest.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.raft.jraft.core;
-
-import org.apache.ignite.raft.jraft.option.RaftOptions;
-import org.apache.ignite.raft.jraft.storage.VolatileStorage;
-import org.junit.jupiter.api.Test;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.instanceOf;
-import static org.hamcrest.Matchers.is;
-
-class VolatileJRaftServiceFactoryTest {
- private final VolatileJRaftServiceFactory serviceFactory = new VolatileJRaftServiceFactory();
-
- @Test
- void producesVolatileMetaStorage() {
- assertThat(serviceFactory.createRaftMetaStorage("test", new RaftOptions()), is(instanceOf(VolatileStorage.class)));
- }
-
- @Test
- void producesVolatileLogStorage() {
- assertThat(serviceFactory.createLogStorage("test", new RaftOptions()), is(instanceOf(VolatileStorage.class)));
- }
-}
\ No newline at end of file
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/SnapshotExecutorTest.java b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/SnapshotExecutorTest.java
index 94f15cf12..4d8a06dd7 100644
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/SnapshotExecutorTest.java
+++ b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/SnapshotExecutorTest.java
@@ -22,6 +22,8 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
+import org.apache.ignite.internal.raft.storage.impl.DefaultLogStorageFactory;
+import org.apache.ignite.internal.raft.storage.impl.IgniteJraftServiceFactory;
import org.apache.ignite.raft.jraft.FSMCaller;
import org.apache.ignite.raft.jraft.JRaftUtils;
import org.apache.ignite.raft.jraft.RaftMessagesFactory;
@@ -29,7 +31,6 @@ import org.apache.ignite.raft.jraft.Status;
import org.apache.ignite.raft.jraft.closure.LoadSnapshotClosure;
import org.apache.ignite.raft.jraft.closure.SaveSnapshotClosure;
import org.apache.ignite.raft.jraft.closure.SynchronizedClosure;
-import org.apache.ignite.raft.jraft.core.DefaultJRaftServiceFactory;
import org.apache.ignite.raft.jraft.core.NodeImpl;
import org.apache.ignite.raft.jraft.core.TimerManager;
import org.apache.ignite.raft.jraft.option.CopyOptions;
@@ -43,7 +44,6 @@ import org.apache.ignite.raft.jraft.rpc.RpcContext;
import org.apache.ignite.raft.jraft.rpc.RpcRequestClosure;
import org.apache.ignite.raft.jraft.rpc.RpcRequests;
import org.apache.ignite.raft.jraft.rpc.RpcResponseClosure;
-import org.apache.ignite.raft.jraft.storage.impl.DefaultLogStorageFactory;
import org.apache.ignite.raft.jraft.storage.snapshot.Snapshot;
import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotExecutorImpl;
import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotReader;
@@ -126,7 +126,7 @@ public class SnapshotExecutorTest extends BaseStorageTest {
Mockito.when(node.getOptions()).thenReturn(options);
Mockito.when(node.getRpcClientService()).thenReturn(raftClientService);
DefaultLogStorageFactory logStorageProvider = Mockito.mock(DefaultLogStorageFactory.class);
- Mockito.when(node.getServiceFactory()).thenReturn(new DefaultJRaftServiceFactory(logStorageProvider));
+ Mockito.when(node.getServiceFactory()).thenReturn(new IgniteJraftServiceFactory(logStorageProvider));
executor = new SnapshotExecutorImpl();
final SnapshotExecutorOptions opts = new SnapshotExecutorOptions();
opts.setFsmCaller(fSMCaller);
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/impl/VolatileRaftMetaStorageTest.java b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/impl/VolatileRaftMetaStorageTest.java
index 208818a22..2e9fe5071 100644
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/impl/VolatileRaftMetaStorageTest.java
+++ b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/storage/impl/VolatileRaftMetaStorageTest.java
@@ -25,7 +25,6 @@ import org.junit.jupiter.api.Test;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.nullValue;
import static org.hamcrest.Matchers.sameInstance;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertTrue;
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index 3a15ffd59..e845b84a4 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -84,6 +84,7 @@ import org.apache.ignite.internal.metastorage.client.WatchListener;
import org.apache.ignite.internal.raft.Loza;
import org.apache.ignite.internal.raft.server.RaftGroupEventsListener;
import org.apache.ignite.internal.raft.server.RaftGroupOptions;
+import org.apache.ignite.internal.raft.storage.impl.VolatileLogStorageFactory;
import org.apache.ignite.internal.schema.SchemaDescriptor;
import org.apache.ignite.internal.schema.SchemaManager;
import org.apache.ignite.internal.schema.SchemaUtils;
@@ -122,6 +123,7 @@ import org.apache.ignite.network.TopologyService;
import org.apache.ignite.raft.client.Peer;
import org.apache.ignite.raft.client.service.RaftGroupListener;
import org.apache.ignite.raft.client.service.RaftGroupService;
+import org.apache.ignite.raft.jraft.storage.impl.VolatileRaftMetaStorage;
import org.apache.ignite.raft.jraft.util.Utils;
import org.apache.ignite.raft.jraft.util.concurrent.ConcurrentHashSet;
import org.apache.ignite.table.Table;
@@ -521,7 +523,13 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
}
private RaftGroupOptions groupOptionsForInternalTable(InternalTable internalTbl) {
- return RaftGroupOptions.forTable(internalTbl.storage().isVolatile());
+ if (internalTbl.storage().isVolatile()) {
+ return RaftGroupOptions.forVolatileStores()
+ .setLogStorageFactory(new VolatileLogStorageFactory())
+ .raftMetaStorageFactory((groupId, raftOptions) -> new VolatileRaftMetaStorage());
+ } else {
+ return RaftGroupOptions.forPersistentStores();
+ }
}
/** {@inheritDoc} */