sashapolo commented on code in PR #2237:
URL: https://github.com/apache/ignite-3/pull/2237#discussion_r1241811950
##########
modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java:
##########
@@ -691,14 +691,24 @@ public CompletableFuture<Ignite> start(Path configPath) {
LOG.info("Components started, joining the cluster");
return cmgMgr.joinFuture()
- // using the default executor to avoid blocking the CMG
Manager threads
+ // Using the default executor to avoid blocking the CMG
Manager threads.
Review Comment:
I think that this comment is no longer relevant and can be removed
##########
modules/runner/src/main/java/org/apache/ignite/internal/recovery/ConfigurationCatchUpListener.java:
##########
@@ -84,6 +84,7 @@ private boolean isConfigurationUpToDate(long targetRevision,
long appliedRevisio
private CompletableFuture<?> checkRevisionUpToDate(long appliedRevision) {
return cfgStorage.lastRevision().thenAccept(rev -> {
synchronized (targetRevisionUpdateMutex) {
+ // TODO: actual metastorage revision can be higher than
configuration revision
Review Comment:
TODO without a ticket, shame on you
##########
modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageManagerImpl.java:
##########
@@ -148,6 +158,60 @@ public MetaStorageManagerImpl(
this.clusterTime = new ClusterTimeImpl(busyLock, clock);
}
+ private CompletableFuture<Long> recover(MetaStorageServiceImpl service) {
+ if (!busyLock.enterBusy()) {
+ return CompletableFuture.failedFuture(new NodeStoppingException());
+ }
+
+ try {
+ CompletableFuture<Long> res = new CompletableFuture<>();
+
+ ExecutorService recoveryExecutor =
Executors.newSingleThreadExecutor(
+ NamedThreadFactory.create(clusterService.nodeName(),
"ms-start", LOG)
+ );
+
+ service.currentRevision().whenCompleteAsync((revision, throwable)
-> {
+ if (throwable != null) {
+ res.completeExceptionally(throwable);
+
+ return;
+ }
+
+ LOG.info("Performing MetaStorage recovery from revision {} to
{}", storage.revision(), revision);
+
+ assert revision != null;
+
+ // Busy wait is ok here, because other threads are not doing
any real work,
+ // and for node to start we must wait until storage is up to
this revision.
+ while (storage.revision() < revision) {
Review Comment:
I don't like this approach, can we introduce an async mechanism to wait for
a particular revision? Like an ability to register revision listeners.
##########
modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageManagerImpl.java:
##########
@@ -148,6 +158,60 @@ public MetaStorageManagerImpl(
this.clusterTime = new ClusterTimeImpl(busyLock, clock);
}
+ private CompletableFuture<Long> recover(MetaStorageServiceImpl service) {
+ if (!busyLock.enterBusy()) {
+ return CompletableFuture.failedFuture(new NodeStoppingException());
+ }
+
+ try {
+ CompletableFuture<Long> res = new CompletableFuture<>();
Review Comment:
This explicit future can be avoided
##########
modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/persistence/RocksDbKeyValueStorage.java:
##########
@@ -364,11 +364,6 @@ public void restoreSnapshot(Path path) {
} finally {
rwLock.writeLock().unlock();
}
-
- // Replay updates if startWatches() has already been called.
- if (recoveryStatus.compareAndSet(RecoveryStatus.PENDING,
RecoveryStatus.IN_PROGRESS)) {
Review Comment:
Looks like we need to get rid of `RecoveryStatus.PENDING` altogether
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java:
##########
@@ -719,11 +732,66 @@ public void nodeWithDataTest() throws
InterruptedException {
checkTableWithData(ignite, TABLE_NAME);
}
+ /**
+ * Restarts the node which stores some data.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void metastorageRecoveryTest(boolean useSnapshot) throws
InterruptedException {
+ List<IgniteImpl> nodes = startNodes(2);
+ IgniteImpl main = nodes.get(0);
+
+ createTableWithData(List.of(main), TABLE_NAME, 1);
+
+ stopNode(1);
+
+ MetaStorageManager metaStorageManager = main.metaStorageManager();
+
+ CompletableFuture[] futs = new CompletableFuture[10];
+
+ for (int i = 0; i < 10; i++) {
+ ByteArray key = ByteArray.fromString("some-test-key-" + i);
+ futs[i] = metaStorageManager.put(key, new byte[]{(byte) 0});
+ }
+
+ CompletableFuture.allOf(futs).join();
+
+ if (useSnapshot) {
+ // Force log truncation, so that restarting node would request a
snapshot.
+ JraftServerImpl server = (JraftServerImpl)
main.raftManager().server();
Review Comment:
Please extract this logic into a method
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java:
##########
@@ -719,11 +732,66 @@ public void nodeWithDataTest() throws
InterruptedException {
checkTableWithData(ignite, TABLE_NAME);
}
+ /**
+ * Restarts the node which stores some data.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void metastorageRecoveryTest(boolean useSnapshot) throws
InterruptedException {
+ List<IgniteImpl> nodes = startNodes(2);
+ IgniteImpl main = nodes.get(0);
+
+ createTableWithData(List.of(main), TABLE_NAME, 1);
+
+ stopNode(1);
+
+ MetaStorageManager metaStorageManager = main.metaStorageManager();
+
+ CompletableFuture[] futs = new CompletableFuture[10];
+
+ for (int i = 0; i < 10; i++) {
+ ByteArray key = ByteArray.fromString("some-test-key-" + i);
Review Comment:
What's the point of inserting data here? You don't check it anywhere
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java:
##########
@@ -719,11 +732,66 @@ public void nodeWithDataTest() throws
InterruptedException {
checkTableWithData(ignite, TABLE_NAME);
}
+ /**
+ * Restarts the node which stores some data.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
Review Comment:
I think you should simply split this test in two
##########
modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/impl/MetaStorageManagerRecoveryTest.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.metastorage.impl;
+
+import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedFast;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import
org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
+import
org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.hlc.HybridClockImpl;
+import
org.apache.ignite.internal.metastorage.command.GetCurrentRevisionCommand;
+import org.apache.ignite.internal.metastorage.server.KeyValueStorage;
+import
org.apache.ignite.internal.metastorage.server.SimpleInMemoryKeyValueStorage;
+import org.apache.ignite.internal.raft.RaftManager;
+import org.apache.ignite.internal.raft.service.RaftGroupService;
+import org.apache.ignite.internal.vault.VaultManager;
+import org.apache.ignite.internal.vault.inmemory.InMemoryVaultService;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.MessagingService;
+import org.apache.ignite.network.NodeMetadata;
+import org.apache.ignite.network.TopologyService;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+import org.junit.jupiter.api.Test;
+
+/** Tests MetaStorage manager recovery basics. */
+public class MetaStorageManagerRecoveryTest {
Review Comment:
Can we use `ItMetaStorageManagerImplTest` instead?
##########
modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageManagerImpl.java:
##########
@@ -148,6 +158,60 @@ public MetaStorageManagerImpl(
this.clusterTime = new ClusterTimeImpl(busyLock, clock);
}
+ private CompletableFuture<Long> recover(MetaStorageServiceImpl service) {
+ if (!busyLock.enterBusy()) {
+ return CompletableFuture.failedFuture(new NodeStoppingException());
+ }
+
+ try {
+ CompletableFuture<Long> res = new CompletableFuture<>();
+
+ ExecutorService recoveryExecutor =
Executors.newSingleThreadExecutor(
+ NamedThreadFactory.create(clusterService.nodeName(),
"ms-start", LOG)
+ );
+
+ service.currentRevision().whenCompleteAsync((revision, throwable)
-> {
+ if (throwable != null) {
+ res.completeExceptionally(throwable);
+
+ return;
+ }
+
+ LOG.info("Performing MetaStorage recovery from revision {} to
{}", storage.revision(), revision);
+
+ assert revision != null;
+
+ // Busy wait is ok here, because other threads are not doing
any real work,
+ // and for node to start we must wait until storage is up to
this revision.
+ while (storage.revision() < revision) {
+ if (!busyLock.enterBusy()) {
+ res.completeExceptionally(new NodeStoppingException());
+
+ return;
+ }
+
+ try {
+ Thread.sleep(10);
+ } catch (InterruptedException ignored) {
+ // Ignore. If we are being shut down, then busy lock
will stop us.
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ LOG.info("Finished MetaStorage recovery");
+
+ res.complete(revision);
+ }, recoveryExecutor);
+
+ return res.whenComplete((s, t) -> {
+ recoveryExecutor.shutdown();
Review Comment:
1. Why not use a method from `IgniteUtils`?
2. This approach is dangerous, because you actually leak this executor
outside (by exposing a future)
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java:
##########
@@ -719,11 +732,66 @@ public void nodeWithDataTest() throws
InterruptedException {
checkTableWithData(ignite, TABLE_NAME);
}
+ /**
+ * Restarts the node which stores some data.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void metastorageRecoveryTest(boolean useSnapshot) throws
InterruptedException {
+ List<IgniteImpl> nodes = startNodes(2);
+ IgniteImpl main = nodes.get(0);
+
+ createTableWithData(List.of(main), TABLE_NAME, 1);
+
+ stopNode(1);
+
+ MetaStorageManager metaStorageManager = main.metaStorageManager();
+
+ CompletableFuture[] futs = new CompletableFuture[10];
+
+ for (int i = 0; i < 10; i++) {
+ ByteArray key = ByteArray.fromString("some-test-key-" + i);
+ futs[i] = metaStorageManager.put(key, new byte[]{(byte) 0});
+ }
+
+ CompletableFuture.allOf(futs).join();
Review Comment:
don't use `join` please
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
