You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sk...@apache.org on 2021/07/29 10:36:19 UTC
[ignite-3] branch main updated: IGNITE-15148 Implemented top level
node stop logic. Fixes #230
This is an automated email from the ASF dual-hosted git repository.
sk0x50 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 ce83394 IGNITE-15148 Implemented top level node stop logic. Fixes #230
ce83394 is described below
commit ce833945ea8fa17a0acf5a54b1d4b39a7f821289
Author: Alexander Lapin <la...@gmail.com>
AuthorDate: Thu Jul 29 13:35:39 2021 +0300
IGNITE-15148 Implemented top level node stop logic. Fixes #230
Signed-off-by: Slava Koptilin <sl...@gmail.com>
---
.../ignite/internal/affinity/AffinityManager.java | 11 +-
.../internal/affinity/AffinityManagerTest.java | 44 ++-
.../main/java/org/apache/ignite/app/Ignite.java | 7 +
.../main/java/org/apache/ignite/app/Ignition.java | 10 +
.../org/apache/ignite/app/IgnitionManager.java | 71 +++-
.../ignite/internal/baseline/BaselineManager.java | 13 +-
.../query/calcite/SqlQueryProcessor.java | 26 +-
.../handler/ClientHandlerIntegrationTest.java | 2 +
.../ignite/internal/client/TcpIgniteClient.java | 6 +
.../apache/ignite/client/AbstractClientTest.java | 2 +
.../org/apache/ignite/client/fakes/FakeIgnite.java | 5 +
.../configuration/hocon/HoconConverterTest.java | 2 +
.../configuration/json/JsonConverterTest.java | 2 +
.../notifications/ConfigurationListenerTest.java | 2 +
.../internal/configuration/sample/UsageTest.java | 4 +
.../configuration/ConfigurationManager.java | 14 +-
.../configuration/ConfigurationRegistry.java | 55 ++--
.../ignite/internal/manager/IgniteComponent.java | 46 +++
.../apache/ignite/lang/NodeStoppingException.java | 75 +++++
.../ITMetaStorageServicePersistenceTest.java | 14 +-
.../client/ITMetaStorageServiceTest.java | 6 +-
.../internal/metastorage/client/CursorImpl.java | 7 +
.../metastorage/client/MetaStorageServiceImpl.java | 3 +-
.../internal/metastorage/MetaStorageManager.java | 40 +--
.../ignite/network/AbstractClusterService.java | 2 +-
.../org/apache/ignite/network/ClusterService.java | 18 +-
.../network/scalecube/ITNodeRestartsTest.java | 6 +-
.../scalecube/ITScaleCubeNetworkMessagingTest.java | 4 +-
.../scalecube/ScaleCubeClusterServiceFactory.java | 12 +-
.../scalecube/ScaleCubeMessagingService.java | 37 ++-
.../ignite/raft/jraft/core/ITCliServiceTest.java | 2 +-
.../apache/ignite/raft/jraft/core/ITNodeTest.java | 2 +-
.../raft/server/ITJRaftCounterServerTest.java | 16 +-
.../raft/server/ITSimpleCounterServerTest.java | 14 +-
.../java/org/apache/ignite/internal/raft/Loza.java | 15 +-
.../ignite/internal/raft/server/RaftServer.java | 10 +-
.../internal/raft/server/impl/JRaftServerImpl.java | 19 +-
.../internal/raft/server/impl/RaftServerImpl.java | 44 ++-
.../apache/ignite/raft/jraft/core/TestCluster.java | 2 +-
.../ignite/raft/jraft/rpc/IgniteRpcTest.java | 4 +-
modules/runner/README.md | 41 ++-
.../ignite/internal/runner/app/IgnitionTest.java | 11 +-
.../org/apache/ignite/internal/app/IgniteImpl.java | 33 +-
.../apache/ignite/internal/app/IgnitionImpl.java | 366 ++++++++++++++++-----
.../ignite/internal/schema/SchemaManager.java | 11 +-
.../SchemaConfigurationConverterTest.java | 8 +
.../ignite/distributed/ITDistributedTableTest.java | 17 +-
.../internal/table/distributed/TableManager.java | 18 +-
.../ignite/internal/table/TableManagerTest.java | 101 +++---
.../PersistencePropertiesVaultServiceTest.java | 28 +-
.../apache/ignite/internal/vault/VaultManager.java | 19 +-
.../apache/ignite/internal/vault/VaultService.java | 3 +-
.../vault/inmemory/InMemoryVaultService.java | 20 +-
.../vault/persistence/PersistentVaultService.java | 35 +-
.../ignite/internal/vault/VaultManagerTest.java | 2 +-
.../ignite/internal/vault/VaultServiceTest.java | 6 +-
56 files changed, 1075 insertions(+), 318 deletions(-)
diff --git a/modules/affinity/src/main/java/org/apache/ignite/internal/affinity/AffinityManager.java b/modules/affinity/src/main/java/org/apache/ignite/internal/affinity/AffinityManager.java
index 6fa6524..316687c 100644
--- a/modules/affinity/src/main/java/org/apache/ignite/internal/affinity/AffinityManager.java
+++ b/modules/affinity/src/main/java/org/apache/ignite/internal/affinity/AffinityManager.java
@@ -26,6 +26,7 @@ import org.apache.ignite.internal.affinity.event.AffinityEvent;
import org.apache.ignite.internal.affinity.event.AffinityEventParameters;
import org.apache.ignite.internal.baseline.BaselineManager;
import org.apache.ignite.internal.configuration.ConfigurationManager;
+import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.manager.Producer;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.client.Conditions;
@@ -42,7 +43,7 @@ import org.jetbrains.annotations.NotNull;
/**
* Affinity manager is responsible for affinity function related logic including calculating affinity assignments.
*/
-public class AffinityManager extends Producer<AffinityEvent, AffinityEventParameters> {
+public class AffinityManager extends Producer<AffinityEvent, AffinityEventParameters> implements IgniteComponent {
/** The logger. */
private static final IgniteLogger LOG = IgniteLogger.forClass(AffinityManager.class);
@@ -76,7 +77,10 @@ public class AffinityManager extends Producer<AffinityEvent, AffinityEventParame
this.configurationMgr = configurationMgr;
this.metaStorageMgr = metaStorageMgr;
this.baselineMgr = baselineMgr;
+ }
+ /** {@inheritDoc} */
+ @Override public void start() {
metaStorageMgr.registerWatchByPrefix(new ByteArray(INTERNAL_PREFIX), new WatchListener() {
@Override public boolean onUpdate(@NotNull WatchEvent watchEvt) {
for (EntryEvent evt : watchEvt.entryEvents()) {
@@ -111,6 +115,11 @@ public class AffinityManager extends Producer<AffinityEvent, AffinityEventParame
});
}
+ /** {@inheritDoc} */
+ @Override public void stop() {
+ // TODO: IGNITE-15161 Implement component's stop.
+ }
+
/**
* Calculates an assignment for a table which was specified by id.
*
diff --git a/modules/affinity/src/test/java/org/apache/ignite/internal/affinity/AffinityManagerTest.java b/modules/affinity/src/test/java/org/apache/ignite/internal/affinity/AffinityManagerTest.java
index ed480fd..19f31f1 100644
--- a/modules/affinity/src/test/java/org/apache/ignite/internal/affinity/AffinityManagerTest.java
+++ b/modules/affinity/src/test/java/org/apache/ignite/internal/affinity/AffinityManagerTest.java
@@ -45,6 +45,7 @@ import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteLogger;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -76,11 +77,13 @@ public class AffinityManagerTest {
/** Before all test scenarios. */
@BeforeEach
- private void before() {
+ private void setUp() {
try {
cfrMgr = new ConfigurationManager(rootConfigurationKeys(), Arrays.asList(
new TestConfigurationStorage(ConfigurationType.DISTRIBUTED)));
+ cfrMgr.start();
+
cfrMgr.bootstrap("{\n" +
" \"table\":{\n" +
" \"tables\":{\n" +
@@ -123,6 +126,12 @@ public class AffinityManagerTest {
}
+ /** Stop configuration manager. */
+ @AfterEach
+ private void tearDown() {
+ cfrMgr.stop();
+ }
+
/**
* Gets a list of configuration keys to use in the test scenario.
*
@@ -189,21 +198,28 @@ public class AffinityManagerTest {
AffinityManager affinityManager = new AffinityManager(cfrMgr, mm, bm);
- CompletableFuture<Boolean> assignmentCalculated = new CompletableFuture<>();
+ try {
+ affinityManager.start();
- affinityManager.listen(AffinityEvent.CALCULATED, new EventListener<AffinityEventParameters>() {
- @Override public boolean notify(@NotNull AffinityEventParameters parameters, @Nullable Throwable e) {
- return assignmentCalculated.complete(e == null);
- }
+ CompletableFuture<Boolean> assignmentCalculated = new CompletableFuture<>();
- @Override public void remove(@NotNull Throwable e) {
- assignmentCalculated.completeExceptionally(e);
- }
- });
+ affinityManager.listen(AffinityEvent.CALCULATED, new EventListener<AffinityEventParameters>() {
+ @Override public boolean notify(@NotNull AffinityEventParameters parameters, @Nullable Throwable e) {
+ return assignmentCalculated.complete(e == null);
+ }
- affinityManager.calculateAssignments(tblId, STATIC_TABLE_NAME);
+ @Override public void remove(@NotNull Throwable e) {
+ assignmentCalculated.completeExceptionally(e);
+ }
+ });
- assertTrue(assignmentCalculated.join());
+ affinityManager.calculateAssignments(tblId, STATIC_TABLE_NAME);
+
+ assertTrue(assignmentCalculated.join());
+ }
+ finally {
+ affinityManager.stop();
+ }
}
/**
@@ -252,6 +268,8 @@ public class AffinityManagerTest {
AffinityManager affinityManager = new AffinityManager(cfrMgr, mm, bm);
+ affinityManager.start();
+
CompletableFuture<Boolean> assignmentRemoved = new CompletableFuture<>();
affinityManager.listen(AffinityEvent.REMOVED, new EventListener<AffinityEventParameters>() {
@@ -267,5 +285,7 @@ public class AffinityManagerTest {
affinityManager.removeAssignment(tblId);
assertTrue(assignmentRemoved.join());
+
+ affinityManager.stop();
}
}
diff --git a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java b/modules/api/src/main/java/org/apache/ignite/app/Ignite.java
index 87e585a..b682be2 100644
--- a/modules/api/src/main/java/org/apache/ignite/app/Ignite.java
+++ b/modules/api/src/main/java/org/apache/ignite/app/Ignite.java
@@ -25,6 +25,13 @@ import org.apache.ignite.tx.IgniteTransactions;
*/
public interface Ignite extends AutoCloseable {
/**
+ * Returns ignite node name.
+ *
+ * @return Ignite node name.
+ */
+ String name();
+
+ /**
* Gets an object for manipulate Ignite tables.
*
* @return Ignite tables.
diff --git a/modules/api/src/main/java/org/apache/ignite/app/Ignition.java b/modules/api/src/main/java/org/apache/ignite/app/Ignition.java
index 244f663..e0e6a41 100644
--- a/modules/api/src/main/java/org/apache/ignite/app/Ignition.java
+++ b/modules/api/src/main/java/org/apache/ignite/app/Ignition.java
@@ -55,4 +55,14 @@ public interface Ignition {
* @return Started Ignite node.
*/
public Ignite start(@NotNull String name, @NotNull Path workDir);
+
+ /**
+ * Stops the node with given {@code name}.
+ * It's possible to stop both already started node or node that is currently starting.
+ * Has no effect if node with specified name doesn't exist.
+ *
+ * @param name Node name to stop.
+ * @throws IllegalArgumentException if null is specified instead of node name.
+ */
+ public void stop(@NotNull String name);
}
diff --git a/modules/api/src/main/java/org/apache/ignite/app/IgnitionManager.java b/modules/api/src/main/java/org/apache/ignite/app/IgnitionManager.java
index 835fc2c..1b4c066 100644
--- a/modules/api/src/main/java/org/apache/ignite/app/IgnitionManager.java
+++ b/modules/api/src/main/java/org/apache/ignite/app/IgnitionManager.java
@@ -24,6 +24,7 @@ import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.util.ServiceLoader;
import org.apache.ignite.lang.IgniteException;
+import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
@@ -44,10 +45,16 @@ public class IgnitionManager {
* @throws IgniteException If error occurs while reading node configuration.
*/
// TODO IGNITE-14580 Add exception handling logic to IgnitionProcessor.
- public static synchronized Ignite start(String nodeName, @Nullable String configStr, Path workDir) {
- if (ignition == null) {
- ServiceLoader<Ignition> ldr = ServiceLoader.load(Ignition.class);
- ignition = ldr.iterator().next();
+ public static Ignite start(
+ @NotNull String nodeName,
+ @Nullable String configStr,
+ @NotNull Path workDir
+ ) {
+ synchronized (IgnitionManager.class) {
+ if (ignition == null) {
+ ServiceLoader<Ignition> ldr = ServiceLoader.load(Ignition.class);
+ ignition = ldr.iterator().next();
+ }
}
if (configStr == null)
@@ -74,14 +81,60 @@ public class IgnitionManager {
* @return Started Ignite node.
*/
// TODO IGNITE-14580 Add exception handling logic to IgnitionProcessor.
- public static synchronized Ignite start(
- String nodeName, @Nullable Path cfgPath, Path workDir, @Nullable ClassLoader clsLdr
+ public static Ignite start(
+ @NotNull String nodeName,
+ @Nullable Path cfgPath,
+ @NotNull Path workDir,
+ @Nullable ClassLoader clsLdr
) {
- if (ignition == null) {
- ServiceLoader<Ignition> ldr = ServiceLoader.load(Ignition.class, clsLdr);
- ignition = ldr.iterator().next();
+ synchronized (IgnitionManager.class) {
+ if (ignition == null) {
+ ServiceLoader<Ignition> ldr = ServiceLoader.load(Ignition.class, clsLdr);
+ ignition = ldr.iterator().next();
+ }
}
return ignition.start(nodeName, cfgPath, workDir);
}
+
+ /**
+ * Stops the node with given {@code name}.
+ * It's possible to stop both already started node or node that is currently starting.
+ * Has no effect if node with specified name doesn't exist.
+ *
+ * @param name Node name to stop.
+ * @throws IllegalArgumentException if null is specified instead of node name.
+ */
+ public static void stop(@NotNull String name) {
+ synchronized (IgnitionManager.class) {
+ if (ignition == null) {
+ ServiceLoader<Ignition> ldr = ServiceLoader.load(Ignition.class);
+ ignition = ldr.iterator().next();
+ }
+ }
+
+ ignition.stop(name);
+ }
+
+ /**
+ * Stops the node with given {@code name}.
+ * It's possible to stop both already started node or node that is currently starting.
+ * Has no effect if node with specified name doesn't exist.
+ *
+ * @param name Node name to stop.
+ * @param clsLdr The class loader to be used to load provider-configuration files
+ * and provider classes, or {@code null} if the system class loader
+ * (or, failing that, the bootstrap class loader) is to be used
+ * @throws IllegalArgumentException if null is specified instead of node name.
+ */
+ public static void stop(@NotNull String name, @Nullable ClassLoader clsLdr) {
+ synchronized (IgnitionManager.class) {
+ if (ignition == null) {
+ ServiceLoader<Ignition> ldr = ServiceLoader.load(Ignition.class, clsLdr);
+ ignition = ldr.iterator().next();
+ }
+ }
+
+ ignition.stop(name);
+ }
}
diff --git a/modules/baseline/src/main/java/org/apache/ignite/internal/baseline/BaselineManager.java b/modules/baseline/src/main/java/org/apache/ignite/internal/baseline/BaselineManager.java
index 01e76c7..b844fb1 100644
--- a/modules/baseline/src/main/java/org/apache/ignite/internal/baseline/BaselineManager.java
+++ b/modules/baseline/src/main/java/org/apache/ignite/internal/baseline/BaselineManager.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.baseline;
import java.util.Collection;
import org.apache.ignite.internal.configuration.ConfigurationManager;
+import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.ClusterService;
@@ -28,7 +29,7 @@ import org.apache.ignite.network.ClusterService;
*/
// TODO: IGNITE-14586 Remove @SuppressWarnings when implementation provided.
// TODO: https://issues.apache.org/jira/browse/IGNITE-14716 Adapt concept of baseline topology IEP-4.
-@SuppressWarnings({"FieldCanBeLocal", "unused"}) public class BaselineManager {
+@SuppressWarnings({"FieldCanBeLocal", "unused"}) public class BaselineManager implements IgniteComponent {
/** Configuration manager in order to handle and listen baseline specific configuration.*/
private final ConfigurationManager configurationMgr;
@@ -58,6 +59,16 @@ import org.apache.ignite.network.ClusterService;
this.clusterSvc = clusterSvc;
}
+ /** {@inheritDoc} */
+ @Override public void start() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void stop() {
+ // TODO: IGNITE-15161 Implement component's stop.
+ }
+
/**
* Gets all nodes which participant in baseline and may process user data.
*
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/SqlQueryProcessor.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/SqlQueryProcessor.java
index f3a8955..e9d5a97 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/SqlQueryProcessor.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/SqlQueryProcessor.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.query.calcite;
import java.util.List;
import org.apache.ignite.internal.manager.EventListener;
+import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.processors.query.calcite.exec.ArrayRowHandler;
import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionService;
import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionServiceImpl;
@@ -33,24 +34,35 @@ import org.apache.ignite.internal.table.distributed.TableManager;
import org.apache.ignite.internal.table.event.TableEvent;
import org.apache.ignite.internal.table.event.TableEventParameters;
import org.apache.ignite.internal.util.Cursor;
+import org.apache.ignite.lang.NodeStoppingException;
import org.apache.ignite.network.ClusterService;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
-public class SqlQueryProcessor {
+public class SqlQueryProcessor implements IgniteComponent {
/** Default Ignite thread keep alive time. */
public static final long DFLT_THREAD_KEEP_ALIVE_TIME = 60_000L;
- private final ExecutionService executionSrvc;
+ private volatile ExecutionService executionSrvc;
- private final MessageService msgSrvc;
+ private volatile MessageService msgSrvc;
- private final QueryTaskExecutor taskExecutor;
+ private volatile QueryTaskExecutor taskExecutor;
+
+ private final ClusterService clusterSrvc;
+
+ private final TableManager tableManager;
public SqlQueryProcessor(
ClusterService clusterSrvc,
TableManager tableManager
) {
+ this.clusterSrvc = clusterSrvc;
+ this.tableManager = tableManager;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void start() {
taskExecutor = new QueryTaskExecutorImpl(
new StripedThreadPoolExecutor(
4,
@@ -83,6 +95,12 @@ public class SqlQueryProcessor {
tableManager.listen(TableEvent.DROP, new TableDroppedListener(schemaHolder));
}
+
+ /** {@inheritDoc} */
+ @Override public void stop() throws NodeStoppingException {
+ // TODO: IGNITE-15161 Implement component's stop.
+ }
+
public List<Cursor<List<?>>> query(String schemaName, String qry, Object... params) {
return executionSrvc.executeQuery(schemaName, qry, params);
}
diff --git a/modules/client-handler/src/test/java/org/apache/ignite/client/handler/ClientHandlerIntegrationTest.java b/modules/client-handler/src/test/java/org/apache/ignite/client/handler/ClientHandlerIntegrationTest.java
index fa73707..ad482d2 100644
--- a/modules/client-handler/src/test/java/org/apache/ignite/client/handler/ClientHandlerIntegrationTest.java
+++ b/modules/client-handler/src/test/java/org/apache/ignite/client/handler/ClientHandlerIntegrationTest.java
@@ -176,6 +176,8 @@ public class ClientHandlerIntegrationTest {
Collections.singletonList(new TestConfigurationStorage(ConfigurationType.LOCAL))
);
+ registry.start();
+
var module = new ClientHandlerModule(mock(Ignite.class), NOPLogger.NOP_LOGGER);
module.prepareStart(registry);
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
index 36a9c5d..85a3294 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
@@ -93,4 +93,10 @@ public class TcpIgniteClient implements Ignite {
@Override public void close() throws Exception {
ch.close();
}
+
+ /** {@inheritDoc} */
+ @Override public String name() {
+ // TODO: improve and finalize IGNITE-15164.
+ return null;
+ }
}
diff --git a/modules/client/src/test/java/org/apache/ignite/client/AbstractClientTest.java b/modules/client/src/test/java/org/apache/ignite/client/AbstractClientTest.java
index f1a562f..a226ef6 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/AbstractClientTest.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/AbstractClientTest.java
@@ -83,6 +83,8 @@ public abstract class AbstractClientTest {
Collections.singletonList(new TestConfigurationStorage(ConfigurationType.LOCAL))
);
+ registry.start();
+
server = new FakeIgnite();
var module = new ClientHandlerModule(server, NOPLogger.NOP_LOGGER);
diff --git a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeIgnite.java b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeIgnite.java
index 4f5504f..0119afd 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeIgnite.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeIgnite.java
@@ -42,4 +42,9 @@ public class FakeIgnite implements Ignite {
@Override public void close() throws Exception {
// No-op.
}
+
+ /** {@inheritDoc} */
+ @Override public String name() {
+ return null;
+ }
}
diff --git a/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/hocon/HoconConverterTest.java b/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/hocon/HoconConverterTest.java
index e0a8fe6..39c475e 100644
--- a/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/hocon/HoconConverterTest.java
+++ b/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/hocon/HoconConverterTest.java
@@ -163,6 +163,8 @@ public class HoconConverterTest {
Collections.singletonList(new TestConfigurationStorage())
);
+ registry.start();
+
configuration = registry.getConfiguration(HoconRootConfiguration.KEY);
}
diff --git a/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/json/JsonConverterTest.java b/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/json/JsonConverterTest.java
index deeb07c..b81d82d 100644
--- a/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/json/JsonConverterTest.java
+++ b/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/json/JsonConverterTest.java
@@ -163,6 +163,8 @@ public class JsonConverterTest {
Collections.singletonList(new TestConfigurationStorage())
);
+ registry.start();
+
configuration = registry.getConfiguration(JsonRootConfiguration.KEY);
}
diff --git a/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/notifications/ConfigurationListenerTest.java b/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/notifications/ConfigurationListenerTest.java
index ec1bbcb..111e9c5 100644
--- a/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/notifications/ConfigurationListenerTest.java
+++ b/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/notifications/ConfigurationListenerTest.java
@@ -81,6 +81,8 @@ public class ConfigurationListenerTest {
Collections.singletonList(testConfigurationStorage)
);
+ registry.start();
+
registry.startStorageConfigurations(testConfigurationStorage.type());
configuration = registry.getConfiguration(ParentConfiguration.KEY);
diff --git a/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/sample/UsageTest.java b/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/sample/UsageTest.java
index 124f471..6756cb9 100644
--- a/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/sample/UsageTest.java
+++ b/modules/configuration-annotation-processor/src/test/java/org/apache/ignite/internal/configuration/sample/UsageTest.java
@@ -54,6 +54,8 @@ public class UsageTest {
Collections.singletonList(new TestConfigurationStorage())
);
+ registry.start();
+
LocalConfiguration root = registry.getConfiguration(LocalConfiguration.KEY);
root.change(local ->
@@ -110,6 +112,8 @@ public class UsageTest {
Collections.singletonList(new TestConfigurationStorage())
);
+ registry.start();
+
registry.getConfiguration(LocalConfiguration.KEY).change(local ->
local.changeBaseline(baseline ->
baseline.changeAutoAdjust(autoAdjust ->
diff --git a/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationManager.java b/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationManager.java
index 9885e4a..f38da2c 100644
--- a/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationManager.java
+++ b/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationManager.java
@@ -31,11 +31,12 @@ import org.apache.ignite.configuration.annotation.ConfigurationType;
import org.apache.ignite.configuration.validation.Validator;
import org.apache.ignite.internal.configuration.hocon.HoconConverter;
import org.apache.ignite.internal.configuration.storage.ConfigurationStorage;
+import org.apache.ignite.internal.manager.IgniteComponent;
/**
* Configuration manager is responsible for handling configuration lifecycle and provides configuration API.
*/
-public class ConfigurationManager {
+public class ConfigurationManager implements IgniteComponent {
/** Configuration registry. */
private final ConfigurationRegistry confRegistry;
@@ -67,6 +68,17 @@ public class ConfigurationManager {
confRegistry = new ConfigurationRegistry(rootKeys, validators, configurationStorages);
}
+ /** {@inheritDoc} */
+ @Override public void start() {
+ confRegistry.start();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void stop() {
+ // TODO: IGNITE-15161 Implement component's stop.
+ confRegistry.stop();
+ }
+
/**
* Constructor.
*
diff --git a/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationRegistry.java b/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationRegistry.java
index daf41a7..4875e5f 100644
--- a/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationRegistry.java
+++ b/modules/configuration/src/main/java/org/apache/ignite/internal/configuration/ConfigurationRegistry.java
@@ -45,6 +45,7 @@ import org.apache.ignite.internal.configuration.util.KeyNotFoundException;
import org.apache.ignite.internal.configuration.validation.ImmutableValidator;
import org.apache.ignite.internal.configuration.validation.MaxValidator;
import org.apache.ignite.internal.configuration.validation.MinValidator;
+import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.lang.IgniteLogger;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -53,31 +54,28 @@ import static org.apache.ignite.internal.configuration.util.ConfigurationNotific
import static org.apache.ignite.internal.configuration.util.ConfigurationUtil.innerNodeVisitor;
/** */
-public class ConfigurationRegistry {
+public class ConfigurationRegistry implements IgniteComponent {
/** The logger. */
private static final IgniteLogger LOG = IgniteLogger.forClass(ConfigurationRegistry.class);
/** */
private final Map<String, DynamicConfiguration<?, ?>> configs = new HashMap<>();
+ /** Root keys. */
+ private final Collection<RootKey<?, ?>> rootKeys;
+
+ /** Validators. */
+ private final Map<Class<? extends Annotation>, Set<Validator<? extends Annotation, ?>>> validators;
+
+ /** Configuration storages. */
+ private final Collection<ConfigurationStorage> configurationStorages;
+
/** */
- private final ConfigurationChanger changer = new ConfigurationChanger(this::notificator) {
- /** {@inheritDoc} */
- @Override public InnerNode createRootNode(RootKey<?, ?> rootKey) {
- return cgen.instantiateNode(rootKey.schemaClass());
- }
- };
+ private volatile ConfigurationChanger changer;
/** */
private final ConfigurationAsmGenerator cgen = new ConfigurationAsmGenerator();
- {
- // Default vaildators implemented in current module.
- changer.addValidator(Min.class, new MinValidator());
- changer.addValidator(Max.class, new MaxValidator());
- changer.addValidator(Immutable.class, new ImmutableValidator());
- }
-
/**
* Constructor.
*
@@ -90,6 +88,24 @@ public class ConfigurationRegistry {
Map<Class<? extends Annotation>, Set<Validator<? extends Annotation, ?>>> validators,
Collection<ConfigurationStorage> configurationStorages
) {
+ this.rootKeys = rootKeys;
+ this.validators = validators;
+ this.configurationStorages = configurationStorages;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void start() {
+ this.changer = new ConfigurationChanger(this::notificator) {
+ /** {@inheritDoc} */
+ @Override public InnerNode createRootNode(RootKey<?, ?> rootKey) {
+ return cgen.instantiateNode(rootKey.schemaClass());
+ }
+ };
+
+ changer.addValidator(Min.class, new MinValidator());
+ changer.addValidator(Max.class, new MaxValidator());
+ changer.addValidator(Immutable.class, new ImmutableValidator());
+
rootKeys.forEach(rootKey -> {
cgen.compileRootSchema(rootKey.schemaClass());
@@ -105,6 +121,12 @@ public class ConfigurationRegistry {
configurationStorages.forEach(changer::register);
}
+ /** {@inheritDoc} */
+ @Override public void stop() {
+ if (changer != null)
+ changer.stop();
+ }
+
/**
* Starts storage configurations.
* @param storageType Storage type.
@@ -165,11 +187,6 @@ public class ConfigurationRegistry {
}
/** */
- public void stop() {
- changer.stop();
- }
-
- /** */
private @NotNull CompletableFuture<Void> notificator(SuperRoot oldSuperRoot, SuperRoot newSuperRoot, long storageRevision) {
List<CompletableFuture<?>> futures = new ArrayList<>();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/manager/IgniteComponent.java b/modules/core/src/main/java/org/apache/ignite/internal/manager/IgniteComponent.java
new file mode 100644
index 0000000..91c4919
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/manager/IgniteComponent.java
@@ -0,0 +1,46 @@
+/*
+ * 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.manager;
+
+/**
+ * Common interface for ignite components that provides entry points for component lifecycle flow.
+ */
+public interface IgniteComponent {
+ /**
+ * Starts the component. Depending on component flow both configuration properties listeners,
+ * meta storage watch registration, starting thread pools and threads goes here.
+ */
+ void start();
+
+ /**
+ * Triggers running before node stop logic. It's guaranteed that during beforeNodeStop all components
+ * beneath given one are still running and node is a part of topology.
+ */
+ default void beforeNodeStop() {
+ // No-op.
+ }
+
+ /**
+ * Stops the component. It's guaranteed that during {@code IgniteComponent#stop())} all components
+ * beneath given one are still running, however the node is no longer part of the topology and, accordingly,
+ * network interaction is impossible.
+ *
+ * @throws Exception If this component cannot be closed
+ */
+ void stop() throws Exception;
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/lang/NodeStoppingException.java b/modules/core/src/main/java/org/apache/ignite/lang/NodeStoppingException.java
new file mode 100644
index 0000000..02349ca
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/lang/NodeStoppingException.java
@@ -0,0 +1,75 @@
+/*
+ * 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.lang;
+
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * This exception is used in order to indicate that Ignite node is stopping (already stopped) for some reason.
+ */
+public class NodeStoppingException extends IgniteInternalCheckedException {
+ /** Serial version UID. */
+ private static final long serialVersionUID = 0L;
+
+ /**
+ * Creates an empty node stopping exception.
+ */
+ public NodeStoppingException() {
+ super("Node is stopping.");
+ }
+
+ /**
+ * Creates a new exception with the given error message.
+ *
+ * @param msg Error message.
+ */
+ public NodeStoppingException(String msg) {
+ super(msg);
+ }
+
+ /**
+ * Creates a new node stopping exception with the given throwable as a cause and
+ * source of error message.
+ *
+ * @param cause Non-null throwable cause.
+ */
+ public NodeStoppingException(Throwable cause) {
+ this(cause.getMessage(), cause);
+ }
+
+ /**
+ * Creates a new node stopping exception with the given error message and optional nested exception.
+ *
+ * @param msg Error message.
+ * @param cause Optional nested exception (can be {@code null}).
+ * @param writableStackTrace Whether or not the stack trace should be writable.
+ */
+ public NodeStoppingException(String msg, @Nullable Throwable cause, boolean writableStackTrace) {
+ super(msg, cause, writableStackTrace);
+ }
+
+ /**
+ * Creates a new node stopping exception with the given error message and optional nested exception.
+ *
+ * @param msg Error message.
+ * @param cause Optional nested exception (can be {@code null}).
+ */
+ public NodeStoppingException(String msg, @Nullable Throwable cause) {
+ super(msg, cause);
+ }
+}
diff --git a/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ITMetaStorageServicePersistenceTest.java b/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ITMetaStorageServicePersistenceTest.java
index 46fcdce..e912b98 100644
--- a/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ITMetaStorageServicePersistenceTest.java
+++ b/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ITMetaStorageServicePersistenceTest.java
@@ -116,7 +116,7 @@ public class ITMetaStorageServicePersistenceTest {
client.shutdown();
for (JRaftServerImpl server : servers)
- server.shutdown();
+ server.stop();
}
/**
@@ -194,7 +194,7 @@ public class ITMetaStorageServicePersistenceTest {
servers.remove(stopIdx);
// Shutdown that node
- toStop.shutdown();
+ toStop.stop();
// Create a raft snapshot of the metastorage service
metaStorageSvc.snapshot(metaStorageSvc.leader()).get();
@@ -373,13 +373,15 @@ public class ITMetaStorageServicePersistenceTest {
Path jraft = workDir.resolve("jraft" + idx);
JRaftServerImpl server = new JRaftServerImpl(service, jraft.toString()) {
- @Override public void shutdown() throws Exception {
- super.shutdown();
+ @Override public void stop() {
+ super.stop();
- service.shutdown();
+ service.stop();
}
};
+ server.start();
+
server.startRaftGroup(
METASTORAGE_RAFT_GROUP_NAME,
new MetaStorageListener(storage),
@@ -417,7 +419,7 @@ public class ITMetaStorageServicePersistenceTest {
@Override public void shutdown() {
super.shutdown();
- clientNode.shutdown();
+ clientNode.stop();
}
};
diff --git a/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ITMetaStorageServiceTest.java b/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ITMetaStorageServiceTest.java
index b4b210d..e9eefc5 100644
--- a/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ITMetaStorageServiceTest.java
+++ b/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ITMetaStorageServiceTest.java
@@ -196,10 +196,10 @@ public class ITMetaStorageServiceTest {
*/
@AfterEach
public void afterTest() throws Exception {
- metaStorageRaftSrv.shutdown();
+ metaStorageRaftSrv.stop();
for (ClusterService node : cluster)
- node.shutdown();
+ node.stop();
}
/**
@@ -1097,6 +1097,8 @@ public class ITMetaStorageServiceTest {
metaStorageRaftSrv = new RaftServerImpl(cluster.get(0), FACTORY);
+ metaStorageRaftSrv.start();
+
metaStorageRaftSrv.
startRaftGroup(METASTORAGE_RAFT_GROUP_NAME, new MetaStorageListener(keyValStorageMock), peers);
diff --git a/modules/metastorage-client/src/main/java/org/apache/ignite/internal/metastorage/client/CursorImpl.java b/modules/metastorage-client/src/main/java/org/apache/ignite/internal/metastorage/client/CursorImpl.java
index 736aca5..79c2daf 100644
--- a/modules/metastorage-client/src/main/java/org/apache/ignite/internal/metastorage/client/CursorImpl.java
+++ b/modules/metastorage-client/src/main/java/org/apache/ignite/internal/metastorage/client/CursorImpl.java
@@ -29,6 +29,7 @@ import org.apache.ignite.internal.util.Cursor;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.lang.NodeStoppingException;
import org.apache.ignite.raft.client.service.RaftGroupService;
import org.jetbrains.annotations.NotNull;
@@ -100,6 +101,9 @@ public class CursorImpl<T> implements Cursor<T> {
cursorId -> metaStorageRaftGrpSvc.<Boolean>run(new CursorHasNextCommand(cursorId))).get();
}
catch (InterruptedException | ExecutionException e) {
+ if (e.getCause().getClass().equals(NodeStoppingException.class))
+ return false;
+
LOG.error("Unable to evaluate cursor hasNext command", e);
throw new IgniteInternalException(e);
@@ -118,6 +122,9 @@ public class CursorImpl<T> implements Cursor<T> {
return fn.apply(res);
}
catch (InterruptedException | ExecutionException e) {
+ if (e.getCause().getClass().equals(NodeStoppingException.class))
+ throw new NoSuchElementException();
+
LOG.error("Unable to evaluate cursor hasNext command", e);
throw new IgniteInternalException(e);
diff --git a/modules/metastorage-client/src/main/java/org/apache/ignite/internal/metastorage/client/MetaStorageServiceImpl.java b/modules/metastorage-client/src/main/java/org/apache/ignite/internal/metastorage/client/MetaStorageServiceImpl.java
index b1927d4..efdf85b 100644
--- a/modules/metastorage-client/src/main/java/org/apache/ignite/internal/metastorage/client/MetaStorageServiceImpl.java
+++ b/modules/metastorage-client/src/main/java/org/apache/ignite/internal/metastorage/client/MetaStorageServiceImpl.java
@@ -51,6 +51,7 @@ import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.lang.NodeStoppingException;
import org.apache.ignite.raft.client.service.RaftGroupService;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -460,7 +461,7 @@ public class MetaStorageServiceImpl implements MetaStorageService {
Thread.sleep(10);
}
catch (Throwable e) {
- if (e instanceof InterruptedException || e.getCause() instanceof InterruptedException)
+ if (e instanceof NodeStoppingException || e.getCause() instanceof NodeStoppingException)
break;
else {
// TODO: IGNITE-14693 Implement Meta storage exception handling logic.
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java
index 9dc383d..ec8fa1b 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java
@@ -29,6 +29,7 @@ import java.util.function.Predicate;
import java.util.stream.Collectors;
import org.apache.ignite.configuration.schemas.runner.NodeConfiguration;
import org.apache.ignite.internal.configuration.ConfigurationManager;
+import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.metastorage.client.CompactedException;
import org.apache.ignite.internal.metastorage.client.Condition;
import org.apache.ignite.internal.metastorage.client.Entry;
@@ -69,7 +70,7 @@ import static org.apache.ignite.internal.util.ByteUtils.longToBytes;
*/
// TODO: IGNITE-14586 Remove @SuppressWarnings when implementation provided.
@SuppressWarnings("unused")
-public class MetaStorageManager {
+public class MetaStorageManager implements IgniteComponent {
/** Meta storage raft group name. */
private static final String METASTORAGE_RAFT_GROUP_NAME = "metastorage_raft_group";
@@ -95,7 +96,7 @@ public class MetaStorageManager {
private final Loza raftMgr;
/** Meta storage service. */
- private final CompletableFuture<MetaStorageService> metaStorageSvcFut;
+ private volatile CompletableFuture<MetaStorageService> metaStorageSvcFut;
/**
* Aggregator of multiple watches to deploy them as one batch.
@@ -123,11 +124,6 @@ public class MetaStorageManager {
private boolean metaStorageNodesOnStart;
/**
- * Lock for the read-then-update logic in the {@link #storeEntries} method.
- */
- private final Object revisionLock = new Object();
-
- /**
* The constructor.
*
* @param vaultMgr Vault manager.
@@ -147,8 +143,11 @@ public class MetaStorageManager {
this.raftMgr = raftMgr;
watchAggregator = new WatchAggregator();
deployFut = new CompletableFuture<>();
+ }
- String[] metastorageNodes = locCfgMgr.configurationRegistry().getConfiguration(NodeConfiguration.KEY)
+ /** {@inheritDoc} */
+ @Override public void start() {
+ String[] metastorageNodes = this.locCfgMgr.configurationRegistry().getConfiguration(NodeConfiguration.KEY)
.metastorageNodes().value();
Predicate<ClusterNode> metaStorageNodesContainsLocPred =
@@ -196,6 +195,11 @@ public class MetaStorageManager {
// clusterNetSvc.messagingService().addMessageHandler((message, senderAddr, correlationId) -> {});
}
+ /** {@inheritDoc} */
+ @Override public void stop() {
+ // TODO: IGNITE-15161 Implement component's stop.
+ }
+
/**
* Deploy all registered watches.
*/
@@ -556,20 +560,18 @@ public class MetaStorageManager {
entries.forEach(e -> batch.put(e.getKey(), e.getValue()));
- synchronized (revisionLock) {
- byte[] appliedRevisionBytes = vaultMgr.get(APPLIED_REV).join().value();
-
- long appliedRevision = appliedRevisionBytes == null ? 0L : bytesToLong(appliedRevisionBytes);
+ byte[] appliedRevisionBytes = vaultMgr.get(APPLIED_REV).join().value();
- if (revision <= appliedRevision) {
- throw new IgniteInternalException(String.format(
- "Current revision (%d) must be greater than the revision in the Vault (%d)",
- revision, appliedRevision
- ));
- }
+ long appliedRevision = appliedRevisionBytes == null ? 0L : bytesToLong(appliedRevisionBytes);
- vaultMgr.putAll(batch).join();
+ if (revision <= appliedRevision) {
+ throw new IgniteInternalException(String.format(
+ "Current revision (%d) must be greater than the revision in the Vault (%d)",
+ revision, appliedRevision
+ ));
}
+
+ vaultMgr.putAll(batch).join();
}
/**
diff --git a/modules/network-api/src/main/java/org/apache/ignite/network/AbstractClusterService.java b/modules/network-api/src/main/java/org/apache/ignite/network/AbstractClusterService.java
index 8185ab4..52a1982 100644
--- a/modules/network-api/src/main/java/org/apache/ignite/network/AbstractClusterService.java
+++ b/modules/network-api/src/main/java/org/apache/ignite/network/AbstractClusterService.java
@@ -20,7 +20,7 @@ package org.apache.ignite.network;
/**
* Default implementation of a {@link ClusterService}.
* <p>
- * Extending classes should use {@link #start()} and {@link #shutdown()} to allocate and free any network-related
+ * Extending classes should use {@link #start()} and {@link #stop()} to allocate and free any network-related
* resources.
*/
public abstract class AbstractClusterService implements ClusterService {
diff --git a/modules/network-api/src/main/java/org/apache/ignite/network/ClusterService.java b/modules/network-api/src/main/java/org/apache/ignite/network/ClusterService.java
index bc5db45..a03d031 100644
--- a/modules/network-api/src/main/java/org/apache/ignite/network/ClusterService.java
+++ b/modules/network-api/src/main/java/org/apache/ignite/network/ClusterService.java
@@ -17,11 +17,13 @@
package org.apache.ignite.network;
+import org.apache.ignite.internal.manager.IgniteComponent;
+
/**
* Class that represents the network-related resources of a node and provides entry points for working with the
* network members of a cluster.
*/
-public interface ClusterService {
+public interface ClusterService extends IgniteComponent {
/**
* @return {@link TopologyService} for working with the cluster topology.
*/
@@ -37,13 +39,15 @@ public interface ClusterService {
*/
ClusterLocalConfiguration localConfiguration();
- /**
- * Starts the current node, allowing it to join the cluster and start receiving messages.
- */
- void start();
+ /** {@inheritDoc} */
+ @Override default void stop() {
+ // TODO: IGNITE-15161 Implement component's stop.
+ }
/**
- * Stops the current node, gracefully freeing the encapsulated resources.
+ * Checks whether cluster service was stopped.
+ *
+ * @return {@code true} if cluster service is stopped, {@code false} otherwise.
*/
- void shutdown();
+ public boolean isStopped();
}
diff --git a/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ITNodeRestartsTest.java b/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ITNodeRestartsTest.java
index d09afc8..63f2592 100644
--- a/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ITNodeRestartsTest.java
+++ b/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ITNodeRestartsTest.java
@@ -53,7 +53,7 @@ class ITNodeRestartsTest {
@AfterEach
void after() {
for (ClusterService service : services)
- service.shutdown();
+ service.stop();
}
/** */
@@ -78,10 +78,10 @@ class ITNodeRestartsTest {
List<NetworkAddress> addresses = nodeFinder.findNodes();
LOG.info("Shutdown {}", addresses.get(idx0));
- services.get(idx0).shutdown();
+ services.get(idx0).stop();
LOG.info("Shutdown {}", addresses.get(idx1));
- services.get(idx1).shutdown();
+ services.get(idx1).stop();
LOG.info("Starting {}", addresses.get(idx0));
ClusterService svc0 = startNetwork(addresses.get(idx0), nodeFinder);
diff --git a/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ITScaleCubeNetworkMessagingTest.java b/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ITScaleCubeNetworkMessagingTest.java
index 96d6155..1ed62c3 100644
--- a/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ITScaleCubeNetworkMessagingTest.java
+++ b/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ITScaleCubeNetworkMessagingTest.java
@@ -313,7 +313,7 @@ class ITScaleCubeNetworkMessagingTest {
if (forceful)
stopForcefully(alice);
else
- alice.shutdown();
+ alice.stop();
boolean aliceShutdownReceived = aliceShutdownLatch.await(forceful ? 10 : 3, TimeUnit.SECONDS);
assertTrue(aliceShutdownReceived);
@@ -420,7 +420,7 @@ class ITScaleCubeNetworkMessagingTest {
* Shutdown cluster.
*/
void shutdown() {
- members.forEach(ClusterService::shutdown);
+ members.forEach(ClusterService::stop);
}
}
}
diff --git a/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeClusterServiceFactory.java b/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeClusterServiceFactory.java
index b029b75..b0b9627 100644
--- a/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeClusterServiceFactory.java
+++ b/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeClusterServiceFactory.java
@@ -102,7 +102,7 @@ public class ScaleCubeClusterServiceFactory implements ClusterServiceFactory {
}
/** {@inheritDoc} */
- @Override public void shutdown() {
+ @Override public void stop() {
// local member will be null, if cluster has not been started
if (cluster.member() == null)
return;
@@ -114,6 +114,16 @@ public class ScaleCubeClusterServiceFactory implements ClusterServiceFactory {
connectionManager.stop();
}
+ /** {@inheritDoc} */
+ @Override public void beforeNodeStop() {
+ stop();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean isStopped() {
+ return cluster.isShutdown();
+ }
+
/**
* Removes the JMX MBean registered by the "io.scalecube.cluster.ClusterImpl#startJmxMonitor()" method.
* Current ScaleCube implementation does not do that which leads to memory leaks.
diff --git a/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeMessagingService.java b/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeMessagingService.java
index cb9e185..0974fdf 100644
--- a/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeMessagingService.java
+++ b/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeMessagingService.java
@@ -20,9 +20,11 @@ package org.apache.ignite.network.scalecube;
import java.time.Duration;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import io.scalecube.cluster.Cluster;
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.net.Address;
+import org.apache.ignite.lang.NodeStoppingException;
import org.apache.ignite.network.AbstractMessagingService;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.MessagingService;
@@ -73,6 +75,15 @@ final class ScaleCubeMessagingService extends AbstractMessagingService {
/** {@inheritDoc} */
@Override public CompletableFuture<Void> send(ClusterNode recipient, NetworkMessage msg) {
+ // TODO: IGNITE-15161 Temporarly, probably should be removed after the implementation
+ // TODO of stopping the clusterService cause some sort of stop thread-safety logic will be implemented.
+ if (cluster.isShutdown()) {
+ CompletableFuture nodeStoppingRes = new CompletableFuture<NetworkMessage>();
+ nodeStoppingRes.completeExceptionally(new NodeStoppingException());
+
+ return nodeStoppingRes;
+ }
+
return cluster
.send(fromNetworkAddress(recipient.address()), Message.fromData(msg))
.toFuture();
@@ -85,6 +96,15 @@ final class ScaleCubeMessagingService extends AbstractMessagingService {
/** {@inheritDoc} */
@Override public CompletableFuture<Void> send(NetworkAddress addr, NetworkMessage msg, String correlationId) {
+ // TODO: IGNITE-15161 Temporarly, probably should be removed after the implementation
+ // TODO of stopping the clusterService cause some sort of stop thread-safety logic will be implemented.
+ if (cluster.isShutdown()) {
+ CompletableFuture nodeStoppingRes = new CompletableFuture<NetworkMessage>();
+ nodeStoppingRes.completeExceptionally(new NodeStoppingException());
+
+ return nodeStoppingRes;
+ }
+
var message = Message
.withData(msg)
.correlationId(correlationId)
@@ -102,16 +122,31 @@ final class ScaleCubeMessagingService extends AbstractMessagingService {
/** {@inheritDoc} */
@Override public CompletableFuture<NetworkMessage> invoke(NetworkAddress addr, NetworkMessage msg, long timeout) {
+ // TODO: IGNITE-15161 Temporarly, probably should be removed after the implementation
+ // TODO of stopping the clusterService cause some sort of stop thread-safety logic will be implemented.
+ if (cluster.isShutdown()) {
+ CompletableFuture nodeStoppingRes = new CompletableFuture<NetworkMessage>();
+ nodeStoppingRes.completeExceptionally(new NodeStoppingException());
+
+ return nodeStoppingRes;
+ }
+
var message = Message
.withData(msg)
.correlationId(UUID.randomUUID().toString())
.build();
+ // TODO: IGNITE-15196 Null seems to be an unexpected result on node stopping.
return cluster
.requestResponse(fromNetworkAddress(addr), message)
.timeout(Duration.ofMillis(timeout))
.toFuture()
- .thenApply(m -> m == null ? null : m.data()); // The result can be null on node stopping.
+ .thenApply(m -> {
+ if (m == null)
+ throw new CompletionException(new NodeStoppingException());
+ else
+ return m.data();
+ }); // The result can be null on node stopping.
}
/**
diff --git a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ITCliServiceTest.java b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ITCliServiceTest.java
index e573529..c9de172 100644
--- a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ITCliServiceTest.java
+++ b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ITCliServiceTest.java
@@ -134,7 +134,7 @@ public class ITCliServiceTest {
@Override public void shutdown() {
super.shutdown();
- clientSvc.shutdown();
+ clientSvc.stop();
}
};
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 98a3528..f325312 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
@@ -3410,7 +3410,7 @@ public class ITNodeTest {
@Override public synchronized void shutdown() {
super.shutdown();
- clusterService.shutdown();
+ clusterService.stop();
}
};
diff --git a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/server/ITJRaftCounterServerTest.java b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/server/ITJRaftCounterServerTest.java
index a1c719a..5897045 100644
--- a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/server/ITJRaftCounterServerTest.java
+++ b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/server/ITJRaftCounterServerTest.java
@@ -151,7 +151,7 @@ class ITJRaftCounterServerTest extends RaftServerAbstractTest {
iterSrv.remove();
- server.shutdown();
+ server.stop();
}
LOG.info(">>>>>>>>>>>>>>> End test method: {}", testInfo.getTestMethod().orElseThrow().getName());
@@ -167,15 +167,17 @@ class ITJRaftCounterServerTest extends RaftServerAbstractTest {
ClusterService service = clusterService("server" + idx, PORT + idx, List.of(addr), true);
JRaftServerImpl server = new JRaftServerImpl(service, dataPath.toString()) {
- @Override public void shutdown() throws Exception {
+ @Override public void stop() {
servers.remove(this);
- super.shutdown();
+ super.stop();
- service.shutdown();
+ service.stop();
}
};
+ server.start();
+
clo.accept(server);
servers.add(server);
@@ -202,7 +204,7 @@ class ITJRaftCounterServerTest extends RaftServerAbstractTest {
super.shutdown();
- clientNode.shutdown();
+ clientNode.stop();
}
};
@@ -526,7 +528,7 @@ class ITJRaftCounterServerTest extends RaftServerAbstractTest {
int stopIdx = servers.indexOf(toStop);
- toStop.shutdown();
+ toStop.stop();
applyIncrements(client1, 11, 20);
applyIncrements(client2, 21, 30);
@@ -549,7 +551,7 @@ class ITJRaftCounterServerTest extends RaftServerAbstractTest {
waitForCondition(() -> validateStateMachine(sum(20), svc2, COUNTER_GROUP_0), 5_000);
waitForCondition(() -> validateStateMachine(sum(30), svc2, COUNTER_GROUP_1), 5_000);
- svc2.shutdown();
+ svc2.stop();
var svc3 = startServer(stopIdx, r -> {
r.startRaftGroup(COUNTER_GROUP_0, listenerFactory.get(), INITIAL_CONF);
diff --git a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/server/ITSimpleCounterServerTest.java b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/server/ITSimpleCounterServerTest.java
index 1f87528..357da90 100644
--- a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/server/ITSimpleCounterServerTest.java
+++ b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/server/ITSimpleCounterServerTest.java
@@ -87,13 +87,15 @@ class ITSimpleCounterServerTest extends RaftServerAbstractTest {
ClusterService service = clusterService(addr.toString(), PORT, List.of(), true);
server = new JRaftServerImpl(service, dataPath.toString()) {
- @Override public synchronized void shutdown() throws Exception {
- super.shutdown();
+ @Override public synchronized void stop() {
+ super.stop();
- service.shutdown();
+ service.stop();
}
};
+ server.start();
+
ClusterNode serverNode = server.clusterService().topologyService().localMember();
server.startRaftGroup(COUNTER_GROUP_ID_0, new CounterListener(), List.of(new Peer(serverNode.address())));
@@ -106,7 +108,7 @@ class ITSimpleCounterServerTest extends RaftServerAbstractTest {
@Override public void shutdown() {
super.shutdown();
- clientNode1.shutdown();
+ clientNode1.stop();
}
};
@@ -117,7 +119,7 @@ class ITSimpleCounterServerTest extends RaftServerAbstractTest {
@Override public void shutdown() {
super.shutdown();
- clientNode2.shutdown();
+ clientNode2.stop();
}
};
@@ -131,7 +133,7 @@ class ITSimpleCounterServerTest extends RaftServerAbstractTest {
*/
@AfterEach
void after() throws Exception {
- server.shutdown();
+ server.stop();
client1.shutdown();
client2.shutdown();
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java
index 638c2ab..b49209d 100644
--- a/modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java
@@ -21,6 +21,8 @@ import java.util.List;
import java.util.stream.Collectors;
import org.apache.ignite.internal.raft.server.RaftServer;
import org.apache.ignite.internal.raft.server.impl.JRaftServerImpl;
+import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.lang.NodeStoppingException;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.raft.client.Peer;
@@ -32,7 +34,7 @@ import org.apache.ignite.raft.client.service.impl.RaftGroupServiceImpl;
/**
* Best raft manager ever since 1982.
*/
-public class Loza {
+public class Loza implements IgniteComponent {
/** Factory. */
private static final RaftClientMessagesFactory FACTORY = new RaftClientMessagesFactory();
@@ -59,6 +61,17 @@ public class Loza {
this.raftServer = new JRaftServerImpl(clusterNetSvc, dataPath);
}
+ /** {@inheritDoc} */
+ @Override public void start() {
+ raftServer.start();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void stop() throws NodeStoppingException {
+ // TODO: IGNITE-15161 Implement component's stop.
+// raftServer.stop();
+ }
+
/**
* Creates a raft group service.
* If {@code nodes} contains the current node, then raft group starts on the current node.
diff --git a/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/RaftServer.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/RaftServer.java
index 65477ce..4b15e49 100644
--- a/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/RaftServer.java
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/RaftServer.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.raft.server;
import java.util.List;
+import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.raft.client.Peer;
import org.apache.ignite.raft.client.service.RaftGroupListener;
@@ -28,7 +29,7 @@ import org.jetbrains.annotations.Nullable;
* The server listens for client commands, submits them to a replicated log and calls
* {@link RaftGroupListener} {@code onRead} and {@code onWrite} methods after the command was committed to the log.
*/
-public interface RaftServer {
+public interface RaftServer extends IgniteComponent {
/**
* @return Cluster service.
*/
@@ -59,11 +60,4 @@ public interface RaftServer {
* @return Local peer or null if the group is not started.
*/
@Nullable Peer localPeer(String groupId);
-
- /**
- * Shutdown a server.
- *
- * @throws Exception If failed.
- */
- void shutdown() throws Exception;
}
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 5d46122..2517712 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
@@ -102,7 +102,10 @@ public class JRaftServerImpl implements RaftServer {
if (opts.getServerName() == null)
opts.setServerName(service.localConfiguration().getName());
+ }
+ /** {@inheritDoc} */
+ @Override public void start() {
if (opts.getCommonExecutor() == null)
opts.setCommonExecutor(JRaftUtils.createCommonExecutor(opts));
@@ -127,6 +130,14 @@ public class JRaftServerImpl implements RaftServer {
}
/** {@inheritDoc} */
+ @Override public void stop() {
+ for (RaftGroupService groupService : groups.values())
+ groupService.shutdown();
+
+ rpcServer.shutdown();
+ }
+
+ /** {@inheritDoc} */
@Override public ClusterService clusterService() {
return service;
}
@@ -214,14 +225,6 @@ public class JRaftServerImpl implements RaftServer {
return groups.get(groupId);
}
- /** {@inheritDoc} */
- @Override public void shutdown() throws Exception {
- for (RaftGroupService groupService : groups.values())
- groupService.shutdown();
-
- rpcServer.shutdown();
- }
-
/**
*
*/
diff --git a/modules/raft/src/test/java/org/apache/ignite/internal/raft/server/impl/RaftServerImpl.java b/modules/raft/src/test/java/org/apache/ignite/internal/raft/server/impl/RaftServerImpl.java
index 767bddc..d148224 100644
--- a/modules/raft/src/test/java/org/apache/ignite/internal/raft/server/impl/RaftServerImpl.java
+++ b/modules/raft/src/test/java/org/apache/ignite/internal/raft/server/impl/RaftServerImpl.java
@@ -28,6 +28,7 @@ import java.util.concurrent.ConcurrentMap;
import java.util.function.BiConsumer;
import org.apache.ignite.internal.raft.server.RaftServer;
import org.apache.ignite.lang.IgniteLogger;
+import org.apache.ignite.lang.NodeStoppingException;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.raft.client.Command;
@@ -72,10 +73,10 @@ public class RaftServerImpl implements RaftServer {
private final BlockingQueue<CommandClosureEx<WriteCommand>> writeQueue;
/** */
- private final Thread readWorker;
+ private volatile Thread readWorker;
/** */
- private final Thread writeWorker;
+ private volatile Thread writeWorker;
/**
* @param service Network service.
@@ -90,7 +91,10 @@ public class RaftServerImpl implements RaftServer {
readQueue = new ArrayBlockingQueue<>(QUEUE_SIZE);
writeQueue = new ArrayBlockingQueue<>(QUEUE_SIZE);
+ }
+ /** {@inheritDoc} */
+ @Override public void start() {
service.messagingService().addMessageHandler(
RaftClientMessageGroup.class,
(message, senderAddr, correlationId) -> {
@@ -133,6 +137,31 @@ public class RaftServerImpl implements RaftServer {
}
/** {@inheritDoc} */
+ @Override public void stop() throws NodeStoppingException {
+ if (readWorker != null) {
+ readWorker.interrupt();
+ try {
+ readWorker.join();
+ }
+ catch (InterruptedException e) {
+ throw new NodeStoppingException("Unable to stop read worker.", e);
+ }
+ }
+
+ if (writeWorker != null) {
+ writeWorker.interrupt();
+ try {
+ writeWorker.join();
+ }
+ catch (InterruptedException e) {
+ throw new NodeStoppingException("Unable to stop write worker.", e);
+ }
+ }
+
+ LOG.info("Stopped replication server [node={}]", service);
+ }
+
+ /** {@inheritDoc} */
@Override public ClusterService clusterService() {
return service;
}
@@ -158,17 +187,6 @@ public class RaftServerImpl implements RaftServer {
return new Peer(service.topologyService().localMember().address());
}
- /** {@inheritDoc} */
- @Override public synchronized void shutdown() throws Exception {
- readWorker.interrupt();
- readWorker.join();
-
- writeWorker.interrupt();
- writeWorker.join();
-
- LOG.info("Stopped replication server [node={}]", service);
- }
-
/**
* @param sender The sender.
* @param req The request.
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 24e2172..868d4f4 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
@@ -239,7 +239,7 @@ public class TestCluster {
rpcServer.shutdown();
- clusterService.shutdown();
+ clusterService.stop();
}
};
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/rpc/IgniteRpcTest.java b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/rpc/IgniteRpcTest.java
index 3827230..d41a90d 100644
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/rpc/IgniteRpcTest.java
+++ b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/rpc/IgniteRpcTest.java
@@ -47,7 +47,7 @@ public class IgniteRpcTest extends AbstractRpcTest {
@Override public void shutdown() {
super.shutdown();
- service.shutdown();
+ service.stop();
}
};
@@ -66,7 +66,7 @@ public class IgniteRpcTest extends AbstractRpcTest {
@Override public void shutdown() {
super.shutdown();
- service.shutdown();
+ service.stop();
}
};
diff --git a/modules/runner/README.md b/modules/runner/README.md
index 47a1927..d0aecd7 100644
--- a/modules/runner/README.md
+++ b/modules/runner/README.md
@@ -163,4 +163,43 @@ local.tables.partition.<ID>.<PARTITION_ID>.storagepath=/path/to/storage/file
Once the projected keys are synced to the vault, the partition manager can create partition Raft servers (initialize
the log and storage, write hard state, register message handlers, etc). Upon startup, the node checks the existing
projected keys (finishing the raft log and storage initialization in case it crashed in the midst) and starts the Raft
-servers.
\ No newline at end of file
+servers.
+
+### Stop node flow.
+It's possible to stop node both when node was already started or during the startup process, in that case
+node stop will prevent any new components startup and stop already started ones.
+
+Following method was added to Ignition interface:
+```
+ /**
+ * Stops the node with given {@code name}.
+ * It's possible to stop both already started node or node that is currently starting.
+ * Has no effect if node with specified name doesn't exist.
+ *
+ * @param name Node name to stop.
+ * @throws IllegalArgumentException if null is specified instead of node name.
+ */
+ public void stop(@NotNull String name);
+```
+It's also possible to stop a node by calling ``close()`` on an already started Ignite instance.
+
+As a starting point stop process checks node status:
+ * If it's STOPPING - nothing happens, cause previous intention to stop node was already detected.
+ * If it's STARTING (means that node is somewhere in the middle of the startup process) - node status will be updated to
+STOPPING and later on startup process will detect status change on attempt of starting next component, prevent further
+startup process and stop already started managers and corresponding inner components.
+ * if it's STARTED - explicit stop will take an action. All components will be stopped.
+
+In all cases the node stop process consists of two phases:
+ * At phase one ``onNodeStop()`` will be called on all started components in reverse order, meaning that the last
+ started component will run ``onNodeStop()`` first. For most components ``onNodeStop()`` will be No-op. Core idea here
+ is to stop network communication on ``onNodeStop()`` in order to terminate distributed operations gracefully:
+ no network communication is allowed but node local logic still remains consistent.
+ * At phase two within a write busy lock, ``stop()`` will be called on all started components also in reverse order,
+ here thread stopping logic, inner structures cleanup and other related logic takes it time. Please pay attention that
+ at this point network communication isn't possible.
+
+Besides local node stopping logic two more actions took place on a cluster as a result of node left event:
+ * Both range and watch cursors will be removed on server side. Given process is linearized with meta storage
+ operations by using a meta storage raft.
+ * Baseline update and corresponding baseline recalculation with ongoing partition raft groups redeployment.
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/IgnitionTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/IgnitionTest.java
index b04106a..937a28f 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/IgnitionTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/IgnitionTest.java
@@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import com.google.common.collect.Lists;
import org.apache.ignite.app.Ignite;
import org.apache.ignite.app.IgnitionManager;
import org.apache.ignite.internal.testframework.WorkDirectory;
@@ -79,8 +80,8 @@ class IgnitionTest {
/** */
@AfterEach
- void tearDown() throws Exception {
- IgniteUtils.closeAll(startedNodes);
+ private void tearDown() throws Exception {
+ IgniteUtils.closeAll(Lists.reverse(startedNodes));
}
/**
@@ -102,8 +103,8 @@ class IgnitionTest {
*/
@Test
void testNodeStartWithoutBootstrapConfiguration() throws Exception {
- try (Ignite ignite = IgnitionManager.start("node0", null, workDir)) {
- Assertions.assertNotNull(ignite);
- }
+ startedNodes.add(IgnitionManager.start("node0", null, workDir));
+
+ Assertions.assertNotNull(startedNodes.get(0));
}
}
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
index 54a93b9..c73a0b7 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -18,7 +18,7 @@
package org.apache.ignite.internal.app;
import org.apache.ignite.app.Ignite;
-import org.apache.ignite.internal.vault.VaultManager;
+import org.apache.ignite.app.IgnitionManager;
import org.apache.ignite.internal.processors.query.calcite.SqlQueryProcessor;
import org.apache.ignite.table.manager.IgniteTables;
import org.apache.ignite.tx.IgniteTransactions;
@@ -28,27 +28,31 @@ import org.apache.ignite.tx.IgniteTransactions;
*/
public class IgniteImpl implements Ignite {
/** Distributed table manager. */
- private final IgniteTables distributedTableManager;
+ private final IgniteTables distributedTblMgr;
- /** Vault manager */
- private final VaultManager vaultManager;
+ /** Ignite node name. */
+ private final String name;
private final SqlQueryProcessor qryEngine;
/**
- * @param tableManager Table manager.
- * @param vaultManager Vault manager.
+ * @param name Ignite node name.
+ * @param tblMgr Table manager.
* @param qryEngine Query processor.
*/
- IgniteImpl(IgniteTables tableManager, VaultManager vaultManager, SqlQueryProcessor qryEngine) {
- this.distributedTableManager = tableManager;
- this.vaultManager = vaultManager;
+ IgniteImpl(
+ String name,
+ IgniteTables tblMgr,
+ SqlQueryProcessor qryEngine
+ ) {
+ this.name = name;
+ this.distributedTblMgr = tblMgr;
this.qryEngine = qryEngine;
}
/** {@inheritDoc} */
@Override public IgniteTables tables() {
- return distributedTableManager;
+ return distributedTblMgr;
}
public SqlQueryProcessor queryEngine() {
@@ -61,7 +65,12 @@ public class IgniteImpl implements Ignite {
}
/** {@inheritDoc} */
- @Override public void close() throws Exception {
- vaultManager.close();
+ @Override public void close() {
+ IgnitionManager.stop(name);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String name() {
+ return name;
}
}
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgnitionImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgnitionImpl.java
index 11e5aac..d1c1b73 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgnitionImpl.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgnitionImpl.java
@@ -27,6 +27,10 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.ListIterator;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.ignite.app.Ignite;
import org.apache.ignite.app.Ignition;
import org.apache.ignite.configuration.RootKey;
@@ -40,6 +44,7 @@ import org.apache.ignite.internal.affinity.AffinityManager;
import org.apache.ignite.internal.baseline.BaselineManager;
import org.apache.ignite.internal.configuration.ConfigurationManager;
import org.apache.ignite.internal.configuration.storage.ConfigurationStorage;
+import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.processors.query.calcite.SqlQueryProcessor;
import org.apache.ignite.internal.raft.Loza;
@@ -50,9 +55,11 @@ import org.apache.ignite.internal.table.distributed.TableManager;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.internal.vault.VaultService;
import org.apache.ignite.internal.vault.persistence.PersistentVaultService;
+import org.apache.ignite.lang.IgniteException;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.lang.LoggerMessageHelper;
+import org.apache.ignite.lang.NodeStoppingException;
import org.apache.ignite.network.ClusterLocalConfiguration;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.network.MessageSerializationRegistryImpl;
@@ -94,6 +101,17 @@ public class IgnitionImpl implements Ignition {
/** */
private static final String VER_KEY = "version";
+ /** Mapping of a node name to a node status, e.g. node0 -> starting or node1 -> stopping. */
+ private static Map<String, NodeState> nodesStatus = new ConcurrentHashMap<>();
+
+ /**
+ * Mapping of a node name to a started node components list.
+ * Given map helps to stop node by stopping all it's components in an appropriate order both
+ * when node is already started which means that all components are ready and
+ * if node is in a middle of a startup process which means that only part of it's components are prepared.
+ */
+ private static Map<String, List<IgniteComponent>> nodesStartedComponents = new ConcurrentHashMap<>();
+
/** {@inheritDoc} */
@Override public Ignite start(@NotNull String nodeName, @Nullable Path cfgPath, @NotNull Path workDir) {
try {
@@ -129,6 +147,24 @@ public class IgnitionImpl implements Ignition {
return doStart(name, null, workDir);
}
+ /** {@inheritDoc} */
+ @Override public void stop(@NotNull String name) {
+ AtomicBoolean explicitStop = new AtomicBoolean();
+
+ nodesStatus.computeIfPresent(name, (nodeName, state) -> {
+ if (state == NodeState.STARTED)
+ explicitStop.set(true);
+
+ return NodeState.STOPPING;
+ });
+
+ if (explicitStop.get()) {
+ List<IgniteComponent> startedComponents = nodesStartedComponents.get(name);
+
+ doStopNode(name, startedComponents);
+ }
+ }
+
/**
* Starts an Ignite node with an optional bootstrap configuration from a HOCON file.
*
@@ -141,110 +177,208 @@ public class IgnitionImpl implements Ignition {
if (nodeName.isEmpty())
throw new IllegalArgumentException("Node name must not be null or empty.");
+ NodeState prevNodeState = nodesStatus.putIfAbsent(nodeName, NodeState.STARTING);
+
+ if (prevNodeState != null) {
+ String errMsg = "Node with name=[" + nodeName + "] already exists in state=[" + prevNodeState + "].";
+
+ LOG.error(errMsg);
+
+ throw new IgniteException(errMsg);
+ }
+
ackBanner();
- VaultManager vaultMgr = createVault(nodeName, workDir);
+ List<IgniteComponent> startedComponents = new ArrayList<>();
- boolean cfgBootstrappedFromPds = vaultMgr.bootstrapped();
+ try {
+ // Vault startup.
+ VaultManager vaultMgr = doStartComponent(
+ nodeName,
+ startedComponents,
+ createVault(workDir)
+ );
- List<RootKey<?, ?>> rootKeys = Arrays.asList(
- NetworkConfiguration.KEY,
- NodeConfiguration.KEY,
- ClusterConfiguration.KEY,
- TablesConfiguration.KEY
- );
+ vaultMgr.putName(nodeName).join();
- List<ConfigurationStorage> cfgStorages =
- new ArrayList<>(Collections.singletonList(new LocalConfigurationStorage(vaultMgr)));
+ boolean cfgBootstrappedFromPds = vaultMgr.bootstrapped();
- // Bootstrap local configuration manager.
- ConfigurationManager locConfigurationMgr = new ConfigurationManager(rootKeys, cfgStorages);
+ List<RootKey<?, ?>> rootKeys = Arrays.asList(
+ NetworkConfiguration.KEY,
+ NodeConfiguration.KEY,
+ ClusterConfiguration.KEY,
+ TablesConfiguration.KEY
+ );
- if (!cfgBootstrappedFromPds && cfgContent != null)
- try {
- locConfigurationMgr.bootstrap(cfgContent, ConfigurationType.LOCAL);
- }
- catch (Exception e) {
- LOG.warn("Unable to parse user-specific configuration, default configuration will be used: {}", e.getMessage());
- }
- else if (cfgContent != null)
- LOG.warn("User specific configuration will be ignored, cause vault was bootstrapped with pds configuration");
- else
- locConfigurationMgr.configurationRegistry().startStorageConfigurations(ConfigurationType.LOCAL);
+ List<ConfigurationStorage> cfgStorages =
+ new ArrayList<>(Collections.singletonList(new LocalConfigurationStorage(vaultMgr)));
+
+ // Bootstrap local configuration manager.
+ ConfigurationManager locConfigurationMgr = doStartComponent(
+ nodeName,
+ startedComponents,
+ new ConfigurationManager(rootKeys, cfgStorages)
+ );
+
+ if (!cfgBootstrappedFromPds && cfgContent != null)
+ try {
+ locConfigurationMgr.bootstrap(cfgContent, ConfigurationType.LOCAL);
+ }
+ catch (Exception e) {
+ LOG.warn("Unable to parse user-specific configuration, default configuration will be used: {}", e.getMessage());
+ }
+ else if (cfgContent != null)
+ LOG.warn("User specific configuration will be ignored, cause vault was bootstrapped with pds configuration");
+ else
+ locConfigurationMgr.configurationRegistry().startStorageConfigurations(ConfigurationType.LOCAL);
+
+ NetworkView netConfigurationView =
+ locConfigurationMgr.configurationRegistry().getConfiguration(NetworkConfiguration.KEY).value();
+
+ var serializationRegistry = new MessageSerializationRegistryImpl();
+
+ var nodeFinder = StaticNodeFinder.fromConfiguration(netConfigurationView);
+
+ // Network startup.
+ ClusterService clusterNetSvc = doStartComponent(
+ nodeName,
+ startedComponents,
+ new ScaleCubeClusterServiceFactory().createClusterService(
+ new ClusterLocalConfiguration(
+ nodeName,
+ netConfigurationView.port(),
+ nodeFinder,
+ serializationRegistry
+ )
+ )
+ );
+
+ // Raft Component startup.
+ Loza raftMgr = doStartComponent(
+ nodeName,
+ startedComponents,
+ new Loza(clusterNetSvc, workDir.toString())
+ );
- NetworkView netConfigurationView =
- locConfigurationMgr.configurationRegistry().getConfiguration(NetworkConfiguration.KEY).value();
+ // Meta storage Component startup.
+ MetaStorageManager metaStorageMgr = doStartComponent(
+ nodeName,
+ startedComponents,
+ new MetaStorageManager(
+ vaultMgr,
+ locConfigurationMgr,
+ clusterNetSvc,
+ raftMgr
+ )
+ );
- var serializationRegistry = new MessageSerializationRegistryImpl();
+ // TODO IGNITE-14578 Bootstrap configuration manager with distributed configuration.
+ cfgStorages.add(new DistributedConfigurationStorage(metaStorageMgr, vaultMgr));
+
+ // Start configuration manager.
+ ConfigurationManager configurationMgr = doStartComponent(
+ nodeName,
+ startedComponents,
+ new ConfigurationManager(rootKeys, cfgStorages)
+ );
+
+ // Baseline manager startup.
+ BaselineManager baselineMgr = doStartComponent(
+ nodeName,
+ startedComponents,
+ new BaselineManager(
+ configurationMgr,
+ metaStorageMgr,
+ clusterNetSvc
+ )
+ );
- var nodeFinder = StaticNodeFinder.fromConfiguration(netConfigurationView);
+ // Affinity manager startup.
+ AffinityManager affinityMgr = doStartComponent(
+ nodeName,
+ startedComponents,
+ new AffinityManager(
+ configurationMgr,
+ metaStorageMgr,
+ baselineMgr
+ )
+ );
- // Network startup.
- ClusterService clusterNetSvc = new ScaleCubeClusterServiceFactory().createClusterService(
- new ClusterLocalConfiguration(
+ // Schema manager startup.
+ SchemaManager schemaMgr = doStartComponent(
nodeName,
- netConfigurationView.port(),
- nodeFinder,
- serializationRegistry
- )
- );
+ startedComponents,
+ new SchemaManager(
+ configurationMgr,
+ metaStorageMgr,
+ vaultMgr
+ )
+ );
- clusterNetSvc.start();
+ // Distributed table manager startup.
+ TableManager distributedTblMgr = doStartComponent(
+ nodeName,
+ startedComponents,
+ new TableManager(
+ configurationMgr,
+ metaStorageMgr,
+ schemaMgr,
+ affinityMgr,
+ raftMgr
+ )
+ );
- // Raft Component startup.
- Loza raftMgr = new Loza(clusterNetSvc, workDir.toString());
+ SqlQueryProcessor qryProc = doStartComponent(
+ nodeName,
+ startedComponents,
+ new SqlQueryProcessor(
+ clusterNetSvc,
+ distributedTblMgr
+ )
+ );
- // Meta storage Component startup.
- MetaStorageManager metaStorageMgr = new MetaStorageManager(
- vaultMgr,
- locConfigurationMgr,
- clusterNetSvc,
- raftMgr
- );
+ // TODO IGNITE-14579 Start rest manager.
- // TODO IGNITE-14578 Bootstrap configuration manager with distributed configuration.
- cfgStorages.add(new DistributedConfigurationStorage(metaStorageMgr, vaultMgr));
+ // Deploy all resisted watches cause all components are ready and have registered their listeners.
+ metaStorageMgr.deployWatches();
- // Start configuration manager.
- ConfigurationManager configurationMgr = new ConfigurationManager(rootKeys, cfgStorages);
+ AtomicBoolean explicitStop = new AtomicBoolean();
- // Baseline manager startup.
- BaselineManager baselineMgr = new BaselineManager(configurationMgr, metaStorageMgr, clusterNetSvc);
+ nodesStatus.computeIfPresent(nodeName, (name, state) -> {
+ switch (state) {
+ case STARTING:
+ nodesStartedComponents.put(name, startedComponents);
- // Affinity manager startup.
- AffinityManager affinityMgr = new AffinityManager(configurationMgr, metaStorageMgr, baselineMgr);
+ return NodeState.STARTED;
+ case STOPPING:
+ explicitStop.set(true);
+ }
- SchemaManager schemaMgr = new SchemaManager(configurationMgr, metaStorageMgr, vaultMgr);
+ return state;
+ });
- // Distributed table manager startup.
- TableManager distributedTblMgr = new TableManager(
- configurationMgr,
- metaStorageMgr,
- schemaMgr,
- affinityMgr,
- raftMgr,
- vaultMgr
- );
+ if (explicitStop.get())
+ throw new NodeStoppingException();
- SqlQueryProcessor qryProc = new SqlQueryProcessor(
- clusterNetSvc,
- distributedTblMgr
- );
+ ackSuccessStart();
- // TODO IGNITE-14579 Start rest manager.
+ return new IgniteImpl(nodeName, distributedTblMgr, qryProc);
+ }
+ catch (Exception e) {
+ String errMsg = "Unable to start node=[" + nodeName + "].";
- // Deploy all resisted watches cause all components are ready and have registered their listeners.
- metaStorageMgr.deployWatches();
+ LOG.error(errMsg, e);
- ackSuccessStart();
+ doStopNode(nodeName, startedComponents);
- return new IgniteImpl(distributedTblMgr, vaultMgr, qryProc);
+ throw new IgniteException(errMsg, e);
+ }
}
/**
* Starts the Vault component.
*/
- private static VaultManager createVault(String nodeName, Path workDir) {
+ private static VaultManager createVault(Path workDir) {
Path vaultPath = workDir.resolve(VAULT_DB_PATH);
try {
@@ -256,11 +390,79 @@ public class IgnitionImpl implements Ignition {
var vaultMgr = new VaultManager(new PersistentVaultService(vaultPath));
- vaultMgr.putName(nodeName).join();
-
return vaultMgr;
}
+ /**
+ * Checks node status. If it's STOPPING then prevents further starting and throws NodeStoppingException
+ * that will lead to stopping already started components later on,
+ * otherwise starts component and add it to started components list.
+ *
+ * @param nodeName Node name.
+ * @param startedComponents List of already started components for given node.
+ * @param component Ignite component to start.
+ * @param <T> Ignite component type.
+ * @return Started ignite component.
+ * @throws NodeStoppingException If node stopping intention was detected.
+ */
+ private static <T extends IgniteComponent> T doStartComponent(
+ @NotNull String nodeName,
+ @NotNull List<IgniteComponent> startedComponents,
+ @NotNull T component
+ ) throws NodeStoppingException {
+ if (nodesStatus.get(nodeName) == NodeState.STOPPING)
+ throw new NodeStoppingException("Node=[" + nodeName + "] was stopped.");
+ else {
+ startedComponents.add(component);
+
+ component.start();
+
+ return component;
+ }
+ }
+
+ /**
+ * Calls {@link IgniteComponent#beforeNodeStop()} and then {@link IgniteComponent#stop()} for all components
+ * in start-reverse-order. Cleanups node started components map and node status map.
+ *
+ * @param nodeName Node name.
+ * @param startedComponents List of already started components for given node.
+ */
+ private static void doStopNode(@NotNull String nodeName, @NotNull List<IgniteComponent> startedComponents) {
+ ListIterator<IgniteComponent> beforeStopIter =
+ startedComponents.listIterator(startedComponents.size() - 1);
+
+ while (beforeStopIter.hasPrevious()) {
+ IgniteComponent componentToExecuteBeforeNodeStop = beforeStopIter.previous();
+
+ try {
+ componentToExecuteBeforeNodeStop.beforeNodeStop();
+ }
+ catch (Exception e) {
+ LOG.error("Unable to execute before node stop on the component=[" +
+ componentToExecuteBeforeNodeStop + "] within node=[" + nodeName + ']', e);
+ }
+ }
+
+ ListIterator<IgniteComponent> stopIter =
+ startedComponents.listIterator(startedComponents.size() - 1);
+
+ while (stopIter.hasPrevious()) {
+ IgniteComponent componentToStop = stopIter.previous();
+
+ try {
+ componentToStop.stop();
+ }
+ catch (Exception e) {
+ LOG.error("Unable to stop component=[" + componentToStop + "] within node=[" + nodeName + ']', e);
+ }
+ }
+
+ nodesStartedComponents.remove(nodeName);
+
+ nodesStatus.remove(nodeName);
+ }
+
/** */
private static void ackSuccessStart() {
LOG.info("Apache Ignite started successfully!");
@@ -276,4 +478,18 @@ public class IgnitionImpl implements Ignition {
LoggerMessageHelper.format("{}\n" + " ".repeat(22) + "Apache Ignite ver. {}\n", banner, ver),
null);
}
+
+ /**
+ * Node state.
+ */
+ private enum NodeState {
+ /** */
+ STARTING,
+
+ /** */
+ STARTED,
+
+ /** */
+ STOPPING
+ }
}
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java
index 90f6b32..7e24972 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java
@@ -31,6 +31,7 @@ import org.apache.ignite.configuration.schemas.table.TableConfiguration;
import org.apache.ignite.configuration.schemas.table.TableView;
import org.apache.ignite.configuration.schemas.table.TablesConfiguration;
import org.apache.ignite.internal.configuration.ConfigurationManager;
+import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.manager.Producer;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.client.Conditions;
@@ -70,7 +71,7 @@ import org.jetbrains.annotations.NotNull;
* @implSpec Initial schema history MAY be registered without the first outdated versions
* that could be cleaned up earlier.
*/
-public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters> {
+public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters> implements IgniteComponent {
/** The logger. */
private static final IgniteLogger LOG = IgniteLogger.forClass(SchemaManager.class);
@@ -107,7 +108,10 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
this.configurationMgr = configurationMgr;
this.metaStorageMgr = metaStorageMgr;
this.vaultMgr = vaultMgr;
+ }
+ /** {@inheritDoc} */
+ @Override public void start() {
metaStorageMgr.registerWatchByPrefix(new ByteArray(INTERNAL_PREFIX), new WatchListener() {
@Override public boolean onUpdate(@NotNull WatchEvent events) {
for (EntryEvent evt : events.entryEvents()) {
@@ -168,6 +172,11 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
});
}
+ /** {@inheritDoc} */
+ @Override public void stop() {
+ // TODO: IGNITE-15161 Implement component's stop.
+ }
+
/**
* Creates schema registry for the table with existed schema or
* registers initial schema from configuration.
diff --git a/modules/schema/src/test/java/org/apache/ignite/internal/schema/configuration/SchemaConfigurationConverterTest.java b/modules/schema/src/test/java/org/apache/ignite/internal/schema/configuration/SchemaConfigurationConverterTest.java
index 0237f60..f46c584 100644
--- a/modules/schema/src/test/java/org/apache/ignite/internal/schema/configuration/SchemaConfigurationConverterTest.java
+++ b/modules/schema/src/test/java/org/apache/ignite/internal/schema/configuration/SchemaConfigurationConverterTest.java
@@ -39,6 +39,7 @@ import org.apache.ignite.schema.builder.PartialIndexBuilder;
import org.apache.ignite.schema.builder.PrimaryIndexBuilder;
import org.apache.ignite.schema.builder.SchemaTableBuilder;
import org.apache.ignite.schema.builder.SortedIndexBuilder;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -69,6 +70,8 @@ public class SchemaConfigurationConverterTest {
Collections.singletonMap(TableValidator.class, Collections.singleton(SchemaTableValidatorImpl.INSTANCE)),
Collections.singleton(new TestConfigurationStorage()));
+ confRegistry.start();
+
tblBuilder = SchemaBuilders.tableBuilder("SNAME","TNAME")
.columns(
SchemaBuilders.column("COL1", ColumnType.DOUBLE).build(),
@@ -88,6 +91,11 @@ public class SchemaConfigurationConverterTest {
}).get();
}
+ @AfterEach
+ private void tearDown() {
+ confRegistry.stop();
+ }
+
/**
* Add/remove HashIndex into configuration and read it back.
*/
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ITDistributedTableTest.java b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ITDistributedTableTest.java
index 524d4a5..c3d3021 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ITDistributedTableTest.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ITDistributedTableTest.java
@@ -148,10 +148,10 @@ public class ITDistributedTableTest {
@AfterEach
public void afterTest() throws Exception {
for (ClusterService node : cluster) {
- node.shutdown();
+ node.stop();
}
- client.shutdown();
+ client.stop();
}
/**
@@ -165,6 +165,8 @@ public class ITDistributedTableTest {
RaftServer partSrv = new JRaftServerImpl(cluster.get(0), dataPath.toString());
+ partSrv.start();
+
List<Peer> conf = List.of(new Peer(cluster.get(0).topologyService().localMember().address()));
partSrv.startRaftGroup(grpId, new PartitionListener(), conf);
@@ -186,7 +188,7 @@ public class ITDistributedTableTest {
assertEquals(testRow.longValue(1), new Row(SCHEMA, getFut.get().getValue()).longValue(1));
- partSrv.shutdown();
+ partSrv.stop();
}
/**
@@ -223,8 +225,13 @@ public class ITDistributedTableTest {
public void partitionedTable() {
HashMap<ClusterNode, RaftServer> raftServers = new HashMap<>(NODES);
- for (int i = 0; i < NODES; i++)
- raftServers.put(cluster.get(i).topologyService().localMember(), new JRaftServerImpl(cluster.get(i), dataPath.toString()));
+ for (int i = 0; i < NODES; i++) {
+ var raftSrv = new JRaftServerImpl(cluster.get(i), dataPath.toString());
+
+ raftSrv.start();
+
+ raftServers.put(cluster.get(i).topologyService().localMember(), raftSrv);
+ }
List<List<ClusterNode>> assignment = RendezvousAffinityFunction.assignPartitions(
cluster.stream().map(node -> node.topologyService().localMember()).collect(Collectors.toList()),
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 c84b5e3..91d3004 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
@@ -44,6 +44,7 @@ import org.apache.ignite.internal.affinity.event.AffinityEventParameters;
import org.apache.ignite.internal.configuration.ConfigurationManager;
import org.apache.ignite.internal.configuration.util.ConfigurationUtil;
import org.apache.ignite.internal.manager.EventListener;
+import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.manager.Producer;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.client.Conditions;
@@ -62,7 +63,6 @@ import org.apache.ignite.internal.table.distributed.storage.InternalTableImpl;
import org.apache.ignite.internal.table.event.TableEvent;
import org.apache.ignite.internal.table.event.TableEventParameters;
import org.apache.ignite.internal.util.Cursor;
-import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteInternalCheckedException;
@@ -79,7 +79,7 @@ import org.jetbrains.annotations.Nullable;
/**
* Table manager.
*/
-public class TableManager extends Producer<TableEvent, TableEventParameters> implements IgniteTables, IgniteTablesInternal {
+public class TableManager extends Producer<TableEvent, TableEventParameters> implements IgniteTables, IgniteTablesInternal, IgniteComponent {
/** The logger. */
private static final IgniteLogger LOG = IgniteLogger.forClass(TableManager.class);
@@ -95,7 +95,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
/** Configuration manager. */
private final ConfigurationManager configurationMgr;
- /** Raft manmager. */
+ /** Raft manager. */
private final Loza raftMgr;
/** Schema manager. */
@@ -118,28 +118,34 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
* @param schemaMgr Schema manager.
* @param affMgr Affinity manager.
* @param raftMgr Raft manager.
- * @param vaultManager Vault manager.
*/
public TableManager(
ConfigurationManager configurationMgr,
MetaStorageManager metaStorageMgr,
SchemaManager schemaMgr,
AffinityManager affMgr,
- Loza raftMgr,
- VaultManager vaultManager
+ Loza raftMgr
) {
this.configurationMgr = configurationMgr;
this.metaStorageMgr = metaStorageMgr;
this.affMgr = affMgr;
this.raftMgr = raftMgr;
this.schemaMgr = schemaMgr;
+ }
+ /** {@inheritDoc} */
+ @Override public void start() {
//TODO: IGNITE-14652 Change a metastorage update in listeners to multi-invoke
configurationMgr.configurationRegistry().getConfiguration(TablesConfiguration.KEY).tables().listen(ctx -> {
return onConfigurationChanged(ctx.storageRevision(), ctx.oldValue(), ctx.newValue());
});
}
+ /** {@inheritDoc} */
+ @Override public void stop() {
+ // TODO: IGNITE-15161 Implement component's stop.
+ }
+
/**
* Creates local structures for a table.
*
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/TableManagerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/TableManagerTest.java
index 75e3e36..d13b2c4 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/TableManagerTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/TableManagerTest.java
@@ -50,7 +50,6 @@ import org.apache.ignite.internal.schema.event.SchemaEvent;
import org.apache.ignite.internal.schema.event.SchemaEventParameters;
import org.apache.ignite.internal.table.distributed.TableManager;
import org.apache.ignite.internal.util.Cursor;
-import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.network.ClusterNode;
@@ -60,6 +59,7 @@ import org.apache.ignite.schema.SchemaBuilders;
import org.apache.ignite.schema.SchemaTable;
import org.apache.ignite.table.Table;
import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
@@ -131,10 +131,6 @@ public class TableManagerTest {
@Mock(lenient = true)
private Loza rm;
- /** Vault manager. */
- @Mock(lenient = true)
- private VaultManager vm;
-
/** Test node. */
private final ClusterNode node = new ClusterNode(
UUID.randomUUID().toString(),
@@ -144,12 +140,14 @@ public class TableManagerTest {
/** Before all test scenarios. */
@BeforeEach
- private void before() {
+ private void setUp() {
try {
cfrMgr = new ConfigurationManager(rootConfigurationKeys(), Arrays.asList(
new TestConfigurationStorage(ConfigurationType.LOCAL),
new TestConfigurationStorage(ConfigurationType.DISTRIBUTED)));
+ cfrMgr.start();
+
cfrMgr.bootstrap("{\n" +
" \"node\":{\n" +
" \"metastorageNodes\":[\n" +
@@ -204,13 +202,19 @@ public class TableManagerTest {
}
}
+ /** Stop configuration manager. */
+ @AfterEach
+ private void tearDown() {
+ cfrMgr.stop();
+ }
+
/**
* Tests a table which was defined before start through bootstrap configuration.
*/
@Disabled("https://issues.apache.org/jira/browse/IGNITE-14578")
@Test
public void testStaticTableConfigured() {
- TableManager tableManager = new TableManager(cfrMgr, mm, sm, am, rm, vm);
+ TableManager tableManager = new TableManager(cfrMgr, mm, sm, am, rm);
assertEquals(1, tableManager.tables().size());
@@ -443,64 +447,73 @@ public class TableManagerTest {
return null;
}).when(am).listen(same(AffinityEvent.CALCULATED), any());
- TableManager tableManager = new TableManager(cfrMgr, mm, sm, am, rm, vm);
+ TableManager tableManager = new TableManager(cfrMgr, mm, sm, am, rm);
- tblManagerFut.complete(tableManager);
+ TableImpl tbl2;
- when(mm.range(eq(new ByteArray(PUBLIC_PREFIX)), any())).thenAnswer(invocation -> {
- Cursor<Entry> cursor = mock(Cursor.class);
+ try {
+ tableManager.start();
- when(cursor.hasNext()).thenReturn(false);
+ tblManagerFut.complete(tableManager);
- return cursor;
- });
+ when(mm.range(eq(new ByteArray(PUBLIC_PREFIX)), any())).thenAnswer(invocation -> {
+ Cursor<Entry> cursor = mock(Cursor.class);
- int tablesBeforeCreation = tableManager.tables().size();
+ when(cursor.hasNext()).thenReturn(false);
- cfrMgr.configurationRegistry().getConfiguration(TablesConfiguration.KEY).tables().listen(ctx -> {
- boolean createTbl = ctx.newValue().get(schemaTable.canonicalName()) != null &&
- ctx.oldValue().get(schemaTable.canonicalName()) == null;
+ return cursor;
+ });
- boolean dropTbl = ctx.oldValue().get(schemaTable.canonicalName()) != null &&
- ctx.newValue().get(schemaTable.canonicalName()) == null;
+ int tablesBeforeCreation = tableManager.tables().size();
- if (!createTbl && !dropTbl)
- return CompletableFuture.completedFuture(null);
+ cfrMgr.configurationRegistry().getConfiguration(TablesConfiguration.KEY).tables().listen(ctx -> {
+ boolean createTbl = ctx.newValue().get(schemaTable.canonicalName()) != null &&
+ ctx.oldValue().get(schemaTable.canonicalName()) == null;
- tableCreatedFlag.set(createTbl);
+ boolean dropTbl = ctx.oldValue().get(schemaTable.canonicalName()) != null &&
+ ctx.newValue().get(schemaTable.canonicalName()) == null;
- when(mm.range(eq(new ByteArray(PUBLIC_PREFIX)), any())).thenAnswer(invocation -> {
- AtomicBoolean firstRecord = new AtomicBoolean(createTbl);
+ if (!createTbl && !dropTbl)
+ return CompletableFuture.completedFuture(null);
- Cursor<Entry> cursor = mock(Cursor.class);
+ tableCreatedFlag.set(createTbl);
- when(cursor.hasNext()).thenAnswer(hasNextInvocation ->
- firstRecord.compareAndSet(true, false));
+ when(mm.range(eq(new ByteArray(PUBLIC_PREFIX)), any())).thenAnswer(invocation -> {
+ AtomicBoolean firstRecord = new AtomicBoolean(createTbl);
- Entry mockEntry = mock(Entry.class);
+ Cursor<Entry> cursor = mock(Cursor.class);
- when(mockEntry.key()).thenReturn(new ByteArray(PUBLIC_PREFIX +
- ConfigurationUtil.escape(schemaTable.canonicalName()) + ".name"));
+ when(cursor.hasNext()).thenAnswer(hasNextInvocation ->
+ firstRecord.compareAndSet(true, false));
- when(cursor.next()).thenReturn(mockEntry);
+ Entry mockEntry = mock(Entry.class);
- return cursor;
- });
+ when(mockEntry.key()).thenReturn(new ByteArray(PUBLIC_PREFIX +
+ ConfigurationUtil.escape(schemaTable.canonicalName()) + ".name"));
- if (phaser != null)
- phaser.arriveAndAwaitAdvance();
+ when(cursor.next()).thenReturn(mockEntry);
- return CompletableFuture.completedFuture(null);
- });
+ return cursor;
+ });
- TableImpl tbl2 = (TableImpl)tableManager.createTable(schemaTable.canonicalName(), tblCh -> SchemaConfigurationConverter.convert(schemaTable, tblCh)
- .changeReplicas(1)
- .changePartitions(10)
- );
+ if (phaser != null)
+ phaser.arriveAndAwaitAdvance();
- assertNotNull(tbl2);
+ return CompletableFuture.completedFuture(null);
+ });
+
+ tbl2 = (TableImpl)tableManager.createTable(schemaTable.canonicalName(), tblCh -> SchemaConfigurationConverter.convert(schemaTable, tblCh)
+ .changeReplicas(1)
+ .changePartitions(10)
+ );
- assertEquals(tablesBeforeCreation + 1, tableManager.tables().size());
+ assertNotNull(tbl2);
+
+ assertEquals(tablesBeforeCreation + 1, tableManager.tables().size());
+ }
+ finally {
+ tableManager.stop();
+ }
return tbl2;
}
diff --git a/modules/vault/src/integrationTest/java/org/apache/ignite/internal/vault/persistence/PersistencePropertiesVaultServiceTest.java b/modules/vault/src/integrationTest/java/org/apache/ignite/internal/vault/persistence/PersistencePropertiesVaultServiceTest.java
index a871fda..42f7e4c 100644
--- a/modules/vault/src/integrationTest/java/org/apache/ignite/internal/vault/persistence/PersistencePropertiesVaultServiceTest.java
+++ b/modules/vault/src/integrationTest/java/org/apache/ignite/internal/vault/persistence/PersistencePropertiesVaultServiceTest.java
@@ -61,30 +61,36 @@ class PersistencePropertiesVaultServiceTest {
);
try (var vaultService = new PersistentVaultService(vaultDir)) {
+ vaultService.start();
+
vaultService.putAll(data).get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
}
try (var vaultService = new PersistentVaultService(vaultDir)) {
+ vaultService.start();
+
assertThat(
vaultService.get(new ByteArray("key" + 1)),
willBe(equalTo(new VaultEntry(new ByteArray("key" + 1), fromString("value" + 1))))
);
}
- try (
- var vaultService = new PersistentVaultService(vaultDir);
- var cursor = vaultService.range(new ByteArray("key" + 1), new ByteArray("key" + 4))
- ) {
- var actualData = new ArrayList<VaultEntry>();
+ try (var vaultService = new PersistentVaultService(vaultDir)) {
+ vaultService.start();
+
+ try (var cursor = vaultService.range(new ByteArray("key" + 1), new ByteArray("key" + 4))) {
+
+ var actualData = new ArrayList<VaultEntry>();
- cursor.forEachRemaining(actualData::add);
+ cursor.forEachRemaining(actualData::add);
- List<VaultEntry> expectedData = data.entrySet().stream()
- .map(e -> new VaultEntry(e.getKey(), e.getValue()))
- .sorted(Comparator.comparing(VaultEntry::key))
- .collect(Collectors.toList());
+ List<VaultEntry> expectedData = data.entrySet().stream()
+ .map(e -> new VaultEntry(e.getKey(), e.getValue()))
+ .sorted(Comparator.comparing(VaultEntry::key))
+ .collect(Collectors.toList());
- assertThat(actualData, is(expectedData));
+ assertThat(actualData, is(expectedData));
+ }
}
}
diff --git a/modules/vault/src/main/java/org/apache/ignite/internal/vault/VaultManager.java b/modules/vault/src/main/java/org/apache/ignite/internal/vault/VaultManager.java
index 6092608..5102d92 100644
--- a/modules/vault/src/main/java/org/apache/ignite/internal/vault/VaultManager.java
+++ b/modules/vault/src/main/java/org/apache/ignite/internal/vault/VaultManager.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.vault;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.util.Cursor;
import org.apache.ignite.lang.ByteArray;
import org.jetbrains.annotations.NotNull;
@@ -29,7 +30,7 @@ import org.jetbrains.annotations.Nullable;
* VaultManager is responsible for handling {@link VaultService} lifecycle
* and providing interface for managing local keys.
*/
-public class VaultManager implements AutoCloseable {
+public class VaultManager implements IgniteComponent {
/** Special key, which reserved for storing the name of the current node. */
private static final ByteArray NODE_NAME = new ByteArray("node_name");
@@ -44,6 +45,17 @@ public class VaultManager implements AutoCloseable {
this.vaultSvc = vaultSvc;
}
+ /** {@inheritDoc} */
+ @Override public void start() {
+ vaultSvc.start();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void stop() throws Exception {
+ // TODO: IGNITE-15161 Implement component's stop.
+ vaultSvc.close();
+ }
+
/**
* @return {@code true} if VaultService beneath given VaultManager was bootstrapped with data
* either from PDS or from user initial bootstrap configuration.
@@ -130,9 +142,4 @@ public class VaultManager implements AutoCloseable {
.thenApply(VaultEntry::value)
.thenApply(name -> name == null ? null : new String(name, StandardCharsets.UTF_8));
}
-
- /** {@inheritDoc} */
- @Override public void close() throws Exception {
- vaultSvc.close();
- }
}
diff --git a/modules/vault/src/main/java/org/apache/ignite/internal/vault/VaultService.java b/modules/vault/src/main/java/org/apache/ignite/internal/vault/VaultService.java
index 37d4b0a..be01836 100644
--- a/modules/vault/src/main/java/org/apache/ignite/internal/vault/VaultService.java
+++ b/modules/vault/src/main/java/org/apache/ignite/internal/vault/VaultService.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.vault;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.util.Cursor;
import org.apache.ignite.lang.ByteArray;
import org.jetbrains.annotations.NotNull;
@@ -27,7 +28,7 @@ import org.jetbrains.annotations.Nullable;
/**
* Defines interface for accessing to a vault service.
*/
-public interface VaultService extends AutoCloseable {
+public interface VaultService extends AutoCloseable, IgniteComponent {
/**
* Retrieves an entry for the given key.
*
diff --git a/modules/vault/src/main/java/org/apache/ignite/internal/vault/inmemory/InMemoryVaultService.java b/modules/vault/src/main/java/org/apache/ignite/internal/vault/inmemory/InMemoryVaultService.java
index df976a4..d617b7a 100644
--- a/modules/vault/src/main/java/org/apache/ignite/internal/vault/inmemory/InMemoryVaultService.java
+++ b/modules/vault/src/main/java/org/apache/ignite/internal/vault/inmemory/InMemoryVaultService.java
@@ -42,6 +42,22 @@ public class InMemoryVaultService implements VaultService {
private final Object mux = new Object();
/** {@inheritDoc} */
+ @Override public void start() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void stop() {
+ // TODO: IGNITE-15161 Implement component's stop.
+ close();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void close() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
@Override @NotNull public CompletableFuture<VaultEntry> get(@NotNull ByteArray key) {
synchronized (mux) {
return CompletableFuture.completedFuture(new VaultEntry(key, storage.get(key)));
@@ -112,8 +128,4 @@ public class InMemoryVaultService implements VaultService {
return CompletableFuture.completedFuture(null);
}
}
-
- /** {@inheritDoc} */
- @Override public void close() {
- }
}
diff --git a/modules/vault/src/main/java/org/apache/ignite/internal/vault/persistence/PersistentVaultService.java b/modules/vault/src/main/java/org/apache/ignite/internal/vault/persistence/PersistentVaultService.java
index 794f8e7..0de6d2b 100644
--- a/modules/vault/src/main/java/org/apache/ignite/internal/vault/persistence/PersistentVaultService.java
+++ b/modules/vault/src/main/java/org/apache/ignite/internal/vault/persistence/PersistentVaultService.java
@@ -60,14 +60,24 @@ public class PersistentVaultService implements VaultService {
private final Options options = new Options();
/** */
- private final RocksDB db;
+ private volatile RocksDB db;
+
+ /** Base path for RocksDB. */
+ private final Path path;
/**
- * Creates and starts the RocksDB instance using the recommended options on the given {@code path}.
+ * Creates persistent vault service.
*
* @param path base path for RocksDB
*/
public PersistentVaultService(Path path) {
+ this.path = path;
+ }
+
+ /**
+ * Creates and starts the RocksDB instance using the recommended options on the given {@code path}.
+ */
+ @Override public void start() {
// using the recommended options from https://github.com/facebook/rocksdb/wiki/Setup-Options-and-Basic-Tuning
options
.setCreateIfMissing(true)
@@ -95,13 +105,26 @@ public class PersistentVaultService implements VaultService {
}
/** {@inheritDoc} */
+ @Override public void stop() {
+ // TODO: IGNITE-15161 Implement component's stop.
+ try {
+ close();
+ }
+ catch (RocksDBException e) {
+ throw new IgniteInternalException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
@Override
public void close() throws RocksDBException {
- try (options; db) {
- db.syncWal();
+ db.syncWal();
- IgniteUtils.shutdownAndAwaitTermination(threadPool, 10, TimeUnit.SECONDS);
- }
+ IgniteUtils.shutdownAndAwaitTermination(threadPool, 10, TimeUnit.SECONDS);
+
+ options.close();
+
+ db.close();
}
/** {@inheritDoc} */
diff --git a/modules/vault/src/test/java/org/apache/ignite/internal/vault/VaultManagerTest.java b/modules/vault/src/test/java/org/apache/ignite/internal/vault/VaultManagerTest.java
index 89c61d5..51ba155 100644
--- a/modules/vault/src/test/java/org/apache/ignite/internal/vault/VaultManagerTest.java
+++ b/modules/vault/src/test/java/org/apache/ignite/internal/vault/VaultManagerTest.java
@@ -41,7 +41,7 @@ public class VaultManagerTest {
/** */
@AfterEach
void tearDown() throws Exception {
- vaultManager.close();
+ vaultManager.stop();
}
/**
diff --git a/modules/vault/src/test/java/org/apache/ignite/internal/vault/VaultServiceTest.java b/modules/vault/src/test/java/org/apache/ignite/internal/vault/VaultServiceTest.java
index e2337c9..919f2ce 100644
--- a/modules/vault/src/test/java/org/apache/ignite/internal/vault/VaultServiceTest.java
+++ b/modules/vault/src/test/java/org/apache/ignite/internal/vault/VaultServiceTest.java
@@ -55,12 +55,14 @@ public abstract class VaultServiceTest {
@BeforeEach
public void setUp() throws IOException {
vaultService = getVaultService();
+
+ vaultService.start();
}
/** */
@AfterEach
- public void tearDown() throws Exception {
- vaultService.close();
+ private void tearDown() throws Exception {
+ vaultService.stop();
}
/**