This is an automated email from the ASF dual-hosted git repository.
apolovtsev 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 6182cc78bd IGNITE-21068 Ignite node must not communicate with a node
removed from the Physical Topology (#2999)
6182cc78bd is described below
commit 6182cc78bdae11b84fc14184a211a84dfd6a5c29
Author: Roman Puchkovskiy <roman.puchkovs...@gmail.com>
AuthorDate: Fri Dec 29 17:40:34 2023 +0400
IGNITE-21068 Ignite node must not communicate with a node removed from the
Physical Topology (#2999)
---
...niteDistributionZoneManagerNodeRestartTest.java | 4 +-
modules/network/build.gradle | 1 +
.../scalecube/ItScaleCubeNetworkMessagingTest.java | 284 ++++++++++++++++-----
.../internal/network/netty/ConnectionManager.java | 29 +++
.../internal/network/netty/ConnectorKey.java | 6 +
.../internal/network/netty/HandshakeHandler.java | 4 +-
.../internal/network/netty/InNetworkObject.java | 18 +-
.../internal/network/netty/MessageHandler.java | 16 +-
.../recovery/RecoveryClientHandshakeManager.java | 2 +-
.../recovery/RecoveryServerHandshakeManager.java | 2 +-
.../{VaultStateIds.java => VaultStaleIds.java} | 6 +-
.../ignite/network/DefaultMessagingService.java | 27 +-
.../scalecube/ScaleCubeClusterServiceFactory.java | 37 ++-
...ultStateIdsTest.java => VaultStaleIdsTest.java} | 8 +-
.../network/DefaultMessagingServiceTest.java | 5 +-
.../runner/app/ItIgniteNodeRestartTest.java | 4 +-
.../org/apache/ignite/internal/app/IgniteImpl.java | 4 +-
17 files changed, 354 insertions(+), 103 deletions(-)
diff --git
a/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItIgniteDistributionZoneManagerNodeRestartTest.java
b/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItIgniteDistributionZoneManagerNodeRestartTest.java
index 6ec20de671..79fefe2e28 100644
---
a/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItIgniteDistributionZoneManagerNodeRestartTest.java
+++
b/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItIgniteDistributionZoneManagerNodeRestartTest.java
@@ -100,7 +100,7 @@ import org.apache.ignite.internal.metastorage.server.If;
import
org.apache.ignite.internal.metastorage.server.TestRocksDbKeyValueStorage;
import
org.apache.ignite.internal.metastorage.server.raft.MetaStorageWriteHandler;
import org.apache.ignite.internal.network.configuration.NetworkConfiguration;
-import org.apache.ignite.internal.network.recovery.VaultStateIds;
+import org.apache.ignite.internal.network.recovery.VaultStaleIds;
import
org.apache.ignite.internal.security.authentication.validator.AuthenticationProvidersValidatorImpl;
import org.apache.ignite.internal.testframework.TestIgnitionManager;
import org.apache.ignite.internal.util.ByteUtils;
@@ -193,7 +193,7 @@ public class ItIgniteDistributionZoneManagerNodeRestartTest
extends BaseIgniteRe
networkConfiguration,
nettyBootstrapFactory,
defaultSerializationRegistry(),
- new VaultStateIds(vault)
+ new VaultStaleIds(vault)
);
var clusterStateStorage = new TestClusterStateStorage();
diff --git a/modules/network/build.gradle b/modules/network/build.gradle
index ff153a7ff3..01d9df2b8c 100644
--- a/modules/network/build.gradle
+++ b/modules/network/build.gradle
@@ -82,4 +82,5 @@ dependencies {
integrationTestImplementation libs.netty.handler
integrationTestImplementation libs.scalecube.cluster
integrationTestImplementation libs.javapoet
+ integrationTestImplementation libs.jetbrains.annotations
}
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 529169a5a6..874d05921e 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
@@ -17,15 +17,20 @@
package org.apache.ignite.network.scalecube;
+import static java.util.Objects.requireNonNull;
import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toUnmodifiableList;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrow;
import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static
org.apache.ignite.utils.ClusterServiceTestUtils.findLocalAddresses;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -45,15 +50,18 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
import org.apache.ignite.internal.lang.IgniteBiTuple;
import org.apache.ignite.internal.lang.NodeStoppingException;
import org.apache.ignite.internal.network.NetworkMessageTypes;
import org.apache.ignite.internal.network.NetworkMessagesFactory;
+import org.apache.ignite.internal.network.handshake.HandshakeException;
import org.apache.ignite.internal.network.messages.TestMessage;
import org.apache.ignite.internal.network.messages.TestMessageTypes;
import org.apache.ignite.internal.network.messages.TestMessagesFactory;
import org.apache.ignite.internal.network.netty.ConnectionManager;
+import org.apache.ignite.internal.network.netty.NettySender;
import
org.apache.ignite.internal.network.recovery.RecoveryClientHandshakeManager;
import
org.apache.ignite.internal.network.recovery.RecoveryServerHandshakeManager;
import
org.apache.ignite.internal.network.recovery.message.HandshakeFinishMessage;
@@ -61,6 +69,7 @@ import
org.apache.ignite.internal.testframework.log4j2.LogInspector;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.network.DefaultMessagingService;
+import org.apache.ignite.network.MessagingService;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.NetworkMessage;
import org.apache.ignite.network.NodeFinder;
@@ -69,8 +78,11 @@ import org.apache.ignite.network.TopologyEventHandler;
import org.apache.ignite.utils.ClusterServiceTestUtils;
import org.apache.logging.log4j.core.LogEvent;
import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import reactor.core.publisher.Mono;
/**
@@ -90,9 +102,15 @@ class ItScaleCubeNetworkMessagingTest {
/** List of test log inspectors. */
private final List<LogInspector> logInspectors = new ArrayList<>();
- /** Tear down method. */
+ private TestInfo testInfo;
+
+ @BeforeEach
+ void saveTestInfo(TestInfo testInfo) {
+ this.testInfo = testInfo;
+ }
+
@AfterEach
- public void tearDown() {
+ void tearDown() {
testCluster.shutdown();
logInspectors.forEach(LogInspector::stop);
logInspectors.clear();
@@ -104,7 +122,7 @@ class ItScaleCubeNetworkMessagingTest {
* @throws Exception in case of errors.
*/
@Test
- public void messageWasSentToAllMembersSuccessfully(TestInfo testInfo)
throws Exception {
+ public void messageWasSentToAllMembersSuccessfully() throws Exception {
Map<String, TestMessage> messageStorage = new ConcurrentHashMap<>();
var messageReceivedLatch = new CountDownLatch(3);
@@ -146,8 +164,8 @@ class ItScaleCubeNetworkMessagingTest {
* @throws Exception If failed.
*/
@Test
- public void testShutdown(TestInfo testInfo) throws Exception {
- testShutdown0(testInfo, false);
+ public void testShutdown() throws Exception {
+ testShutdown0(false);
}
/**
@@ -156,8 +174,8 @@ class ItScaleCubeNetworkMessagingTest {
* @throws Exception If failed.
*/
@Test
- public void testForcefulShutdown(TestInfo testInfo) throws Exception {
- testShutdown0(testInfo, true);
+ public void testForcefulShutdown() throws Exception {
+ testShutdown0(true);
}
/**
@@ -166,7 +184,7 @@ class ItScaleCubeNetworkMessagingTest {
* @throws Exception in case of errors.
*/
@Test
- public void testSendMessageToSelf(TestInfo testInfo) throws Exception {
+ public void testSendMessageToSelf() throws Exception {
testCluster = new Cluster(1, testInfo);
testCluster.startAwait();
@@ -213,7 +231,7 @@ class ItScaleCubeNetworkMessagingTest {
* @throws Exception in case of errors.
*/
@Test
- public void testInvokeMessageToSelf(TestInfo testInfo) throws Exception {
+ public void testInvokeMessageToSelf() throws Exception {
testCluster = new Cluster(1, testInfo);
testCluster.startAwait();
@@ -246,7 +264,7 @@ class ItScaleCubeNetworkMessagingTest {
* exceptionally.
*/
@Test
- public void testInvokeAfterStop(TestInfo testInfo) throws
InterruptedException {
+ public void testInvokeAfterStop() throws InterruptedException {
testCluster = new Cluster(2, testInfo);
testCluster.startAwait();
@@ -282,7 +300,7 @@ class ItScaleCubeNetworkMessagingTest {
* exceptionally.
*/
@Test
- public void testInvokeDuringStop(TestInfo testInfo) throws
InterruptedException {
+ public void testInvokeDuringStop() throws InterruptedException {
testCluster = new Cluster(2, testInfo);
testCluster.startAwait();
@@ -320,7 +338,7 @@ class ItScaleCubeNetworkMessagingTest {
* exceptionally.
*/
@Test
- public void testStopDuringAwaitingForInvokeResponse(TestInfo testInfo)
throws InterruptedException {
+ public void testStopDuringAwaitingForInvokeResponse() throws
InterruptedException {
testCluster = new Cluster(2, testInfo);
testCluster.startAwait();
@@ -366,7 +384,7 @@ class ItScaleCubeNetworkMessagingTest {
* Tests that Scalecube messages are not blocked if some message handler
blocks handling of 'normal' messages.
*/
@Test
- public void scalecubeMessagesAreSentSeparatelyFromOtherMessages(TestInfo
testInfo) throws InterruptedException {
+ public void scalecubeMessagesAreSentSeparatelyFromOtherMessages() throws
InterruptedException {
testCluster = new Cluster(2, testInfo);
testCluster.startAwait();
@@ -420,7 +438,7 @@ class ItScaleCubeNetworkMessagingTest {
* @throws Exception in case of errors.
*/
@Test
- public void testMessageGroupsHandlers(TestInfo testInfo) throws Exception {
+ public void testMessageGroupsHandlers() throws Exception {
testCluster = new Cluster(2, testInfo);
testCluster.startAwait();
@@ -477,27 +495,183 @@ class ItScaleCubeNetworkMessagingTest {
* @throws Exception in case of errors.
*/
@SuppressWarnings("ConstantConditions")
- @Test
- public void nodeCannotReuseOldId(TestInfo testInfo) throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void nodeCannotReuseOldId(boolean keepPreExistingConnections)
throws Exception {
testCluster = new Cluster(3, testInfo);
testCluster.startAwait();
String outcastName =
testCluster.members.get(testCluster.members.size() - 1).nodeName();
- knockOutNode(outcastName);
+ knockOutNode(outcastName, !keepPreExistingConnections);
IgniteBiTuple<CountDownLatch, AtomicBoolean> pair =
reanimateNode(outcastName);
CountDownLatch ready = pair.get1();
AtomicBoolean reappeared = pair.get2();
- assertTrue(ready.await(10, TimeUnit.SECONDS), "Node neither
reappeared, not was rejected");
+ assertTrue(ready.await(10, TimeUnit.SECONDS), "Node neither
reappeared, nor was rejected");
assertThat(reappeared.get(), is(false));
}
- private void knockOutNode(String outcastName) throws InterruptedException {
- CountDownLatch disappeared = new CountDownLatch(2);
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void nodeCannotCommunicateAfterLeavingPhysicalTopology(boolean
keepPreExistingConnections) throws Exception {
+ testCluster = new Cluster(3, testInfo);
+
+ testCluster.startAwait();
+
+ ClusterService notOutcast = testCluster.members.get(0);
+ ClusterService outcast =
testCluster.members.get(testCluster.members.size() - 1);
+
+ ClusterNode outcastNode =
notOutcast.topologyService().getByConsistentId(outcast.nodeName());
+ ClusterNode notOutcastNode =
outcast.topologyService().getByConsistentId(notOutcast.nodeName());
+ assertNotNull(outcastNode);
+ assertNotNull(notOutcastNode);
+
+ if (keepPreExistingConnections) {
+ assertThat(notOutcast.messagingService().send(outcastNode,
messageFactory.testMessage().build()), willCompleteSuccessfully());
+ assertThat(outcast.messagingService().send(notOutcastNode,
messageFactory.testMessage().build()), willCompleteSuccessfully());
+ }
+
+ knockOutNode(outcast.nodeName(), !keepPreExistingConnections);
+
+ stopDroppingMessagesTo(outcast.nodeName());
+
+ CompletableFuture<Void> sendFromOutcast =
outcast.messagingService().send(notOutcastNode,
messageFactory.testMessage().build());
+ assertThat(sendFromOutcast, willThrow(HandshakeException.class));
+
+ CompletableFuture<?> invokeFromOutcast =
outcast.messagingService().invoke(
+ notOutcastNode,
+ messageFactory.testMessage().build(),
+ 10_000
+ );
+ assertThat(invokeFromOutcast, willThrow(HandshakeException.class));
+
+ CompletableFuture<Void> sendToOutcast =
notOutcast.messagingService().send(outcastNode,
messageFactory.testMessage().build());
+ assertThat(sendToOutcast, willThrow(HandshakeException.class));
+
+ CompletableFuture<?> invokeToOutcast =
notOutcast.messagingService().invoke(
+ outcastNode,
+ messageFactory.testMessage().build(),
+ 10_000
+ );
+ assertThat(invokeToOutcast, willThrow(HandshakeException.class));
+ }
+
+ @Test
+ public void reconnectsAfterConnectionDrop() throws Exception {
+ testCluster = new Cluster(2, testInfo);
+
+ testCluster.startAwait();
+
+ ClusterService sender = testCluster.members.get(0);
+ ClusterService receiver = testCluster.members.get(1);
+
+ receiver.messagingService().addMessageHandler(
+ TestMessageTypes.class,
+ (message, senderConsistentId, correlationId) -> {
+ receiver.messagingService().respond(
+ sender.topologyService().localMember(),
+ message,
+ requireNonNull(correlationId)
+ );
+ }
+ );
+
+ // Make an invocation to establish a connection.
+ CompletableFuture<?> firstInvoke = sender.messagingService().invoke(
+ receiver.topologyService().localMember(),
+ messageFactory.testMessage().build(),
+ 10_000
+ );
+ assertThat(firstInvoke, willCompleteSuccessfully());
+
+ closeAllChannels(sender.messagingService());
+
+ // Now try again.
+ CompletableFuture<?> secondInvoke = sender.messagingService().invoke(
+ receiver.topologyService().localMember(),
+ messageFactory.testMessage().build(),
+ 10_000
+ );
+ assertThat(secondInvoke, willCompleteSuccessfully());
+ }
+
+ @Test
+ public void doesNotDeliverMessagesWhoseSenderLeftPhysicalTopology() throws
Exception {
+ testCluster = new Cluster(2, testInfo);
+
+ testCluster.startAwait();
+
+ ClusterService sender = testCluster.members.get(0);
+ ClusterService receiver = testCluster.members.get(1);
+
+ // We are going to send 3 messages, of which 2 will arrive after the
sender has been removed from the physical topology,
+ // so we expect 2 messages to be 'skipped' and not delivered on the
receiver.
+ CountDownLatch messagesSkipped = new CountDownLatch(2);
+ logInspectors.add(
+ new LogInspector(
+ DefaultMessagingService.class.getName(),
+ event ->
event.getMessage().getFormattedMessage().contains("is stale, so skipping
message handling"),
+ messagesSkipped::countDown
+ )
+ );
+ logInspectors.forEach(LogInspector::start);
+
+ AtomicBoolean first = new AtomicBoolean(true);
+ CountDownLatch canProceed = new CountDownLatch(1);
+ CountDownLatch blockingStarted = new CountDownLatch(1);
+ AtomicInteger messagesDelivered = new AtomicInteger();
+
+ receiver.messagingService().addMessageHandler(
+ TestMessageTypes.class,
+ (message, senderConsistentId, correlationId) -> {
+ if (first.compareAndSet(true, false)) {
+ blockingStarted.countDown();
+
+ try {
+ assertTrue(canProceed.await(10, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ messagesDelivered.incrementAndGet();
+ }
+ );
+
+ TestMessage message = messageFactory.testMessage().build();
+
+ // This message will get stuck.
+
sender.messagingService().send(receiver.topologyService().localMember(),
message);
+
+ // These 2 will be handled after the sender has left the physical
topology.
+
sender.messagingService().send(receiver.topologyService().localMember(),
message);
+
sender.messagingService().invoke(receiver.topologyService().localMember(),
message, 10_000);
+
+ assertTrue(blockingStarted.await(10, TimeUnit.SECONDS));
+
+ knockOutNode(sender.nodeName(), false);
+
+ canProceed.countDown();
+
+ assertTrue(messagesSkipped.await(10, TimeUnit.SECONDS), "Messages were
not skipped");
+ assertThat(messagesDelivered.get(), is(1));
+ }
+
+ private static void closeAllChannels(MessagingService messagingService)
throws InterruptedException {
+ ConnectionManager connectionManager = ((DefaultMessagingService)
messagingService).connectionManager();
+
+ for (NettySender sender : connectionManager.channels().values()) {
+ sender.close().await(10, TimeUnit.SECONDS);
+ }
+ }
+
+ private void knockOutNode(String outcastName, boolean
closeConnectionsForcibly) throws InterruptedException {
+ CountDownLatch disappeared = new
CountDownLatch(testCluster.members.size() - 1);
TopologyEventHandler disappearListener = new TopologyEventHandler() {
@Override
@@ -524,17 +698,15 @@ class ItScaleCubeNetworkMessagingTest {
// Wait until all nodes see disappearance of the outcast.
assertTrue(disappeared.await(10, TimeUnit.SECONDS), "Node did not
disappear in time");
- DefaultMessagingService messagingService = (DefaultMessagingService)
testCluster.members.stream()
- .filter(service -> outcastName.equals(service.nodeName()))
- .findFirst()
- .get().messagingService();
-
- ConnectionManager cm = messagingService.connectionManager();
+ if (closeConnectionsForcibly) {
+ MessagingService messagingService = testCluster.members.stream()
+ .filter(service -> outcastName.equals(service.nodeName()))
+ .findFirst().orElseThrow()
+ .messagingService();
- // Forcefully close channels, so that nodes will create new channels
on reanimation of the outcast.
- cm.channels().forEach((stringConnectorKey, nettySender) -> {
- nettySender.close().awaitUninterruptibly();
- });
+ // Forcefully close channels, so that nodes will create new
channels on reanimation of the outcast.
+ closeAllChannels(messagingService);
+ }
}
private IgniteBiTuple<CountDownLatch, AtomicBoolean> reanimateNode(String
outcastName) {
@@ -557,33 +729,36 @@ class ItScaleCubeNetworkMessagingTest {
logInspectors.add(new LogInspector(
RecoveryClientHandshakeManager.class.getName(),
matcher,
- ready::countDown));
+ () -> ready.countDown()));
logInspectors.add(new LogInspector(
RecoveryServerHandshakeManager.class.getName(),
matcher,
- ready::countDown));
+ () -> ready.countDown()));
logInspectors.forEach(LogInspector::start);
+ stopDroppingMessagesTo(outcastName);
+
+ return new IgniteBiTuple<>(ready, reappeared);
+ }
+
+ private void stopDroppingMessagesTo(String outcastName) {
testCluster.members.stream()
.filter(service -> !outcastName.equals(service.nodeName()))
.forEach(service -> {
DefaultMessagingService messagingService =
(DefaultMessagingService) service.messagingService();
messagingService.stopDroppingMessages();
});
-
- return new IgniteBiTuple<>(ready, reappeared);
}
/**
* Tests shutdown.
*
- * @param testInfo Test info.
* @param forceful Whether shutdown should be forceful.
* @throws Exception If failed.
*/
- private void testShutdown0(TestInfo testInfo, boolean forceful) throws
Exception {
+ private void testShutdown0(boolean forceful) throws Exception {
testCluster = new Cluster(2, testInfo);
testCluster.startAwait();
@@ -651,9 +826,6 @@ class ItScaleCubeNetworkMessagingTest {
/** Members of the cluster. */
final List<ClusterService> members;
- /** Latch that is locked until all members are visible in the
topology. */
- private final CountDownLatch startupLatch;
-
/** Node finder. */
private final NodeFinder nodeFinder;
@@ -664,18 +836,14 @@ class ItScaleCubeNetworkMessagingTest {
* @param testInfo Test info.
*/
Cluster(int numOfNodes, TestInfo testInfo) {
- startupLatch = new CountDownLatch(numOfNodes - 1);
-
int initialPort = 3344;
List<NetworkAddress> addresses = findLocalAddresses(initialPort,
initialPort + numOfNodes);
this.nodeFinder = new StaticNodeFinder(addresses);
- var isInitial = new AtomicBoolean(true);
-
members = addresses.stream()
- .map(addr -> startNode(testInfo, addr,
isInitial.getAndSet(false)))
+ .map(addr -> startNode(testInfo, addr))
.collect(toUnmodifiableList());
}
@@ -684,25 +852,10 @@ class ItScaleCubeNetworkMessagingTest {
*
* @param testInfo Test info.
* @param addr Node address.
- * @param initial Whether this node is the first one.
* @return Started cluster node.
*/
- private ClusterService startNode(
- TestInfo testInfo, NetworkAddress addr, boolean initial
- ) {
- ClusterService clusterSvc =
ClusterServiceTestUtils.clusterService(testInfo, addr.port(), nodeFinder);
-
- if (initial) {
- clusterSvc.topologyService().addEventHandler(new
TopologyEventHandler() {
- /** {@inheritDoc} */
- @Override
- public void onAppeared(ClusterNode member) {
- startupLatch.countDown();
- }
- });
- }
-
- return clusterSvc;
+ private ClusterService startNode(TestInfo testInfo, NetworkAddress
addr) {
+ return ClusterServiceTestUtils.clusterService(testInfo,
addr.port(), nodeFinder);
}
/**
@@ -714,11 +867,18 @@ class ItScaleCubeNetworkMessagingTest {
void startAwait() throws InterruptedException {
members.forEach(ClusterService::start);
- if (!startupLatch.await(3, TimeUnit.SECONDS)) {
+ if (!waitForCondition(this::allMembersSeeEachOther,
TimeUnit.SECONDS.toMillis(3))) {
throw new AssertionError();
}
}
+ private boolean allMembersSeeEachOther() {
+ int totalMembersSeen = members.stream()
+ .mapToInt(m -> m.topologyService().allMembers().size())
+ .sum();
+ return totalMembersSeen == members.size() * members.size();
+ }
+
/**
* Stops the cluster.
*/
diff --git
a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectionManager.java
b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectionManager.java
index 6587b91bde..4651b4557c 100644
---
a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectionManager.java
+++
b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectionManager.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.network.netty;
import static java.util.function.Function.identity;
+import static java.util.stream.Collectors.toList;
import static org.apache.ignite.network.ChannelType.getChannel;
import io.netty.bootstrap.Bootstrap;
@@ -26,6 +27,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
@@ -327,6 +329,15 @@ public class ConnectionManager implements
ChannelCreationListener {
// Old channel can still be in the map, but it must be closed already
by the tie breaker in the
// handshake manager.
assert oldChannel == null || !oldChannel.isOpen() : "Incorrect channel
creation flow";
+
+ // Preventing a race between calling closeConnectionsWith() and
putting a new channel that was just opened (with the node
+ // which is already stale). If it's stale, then the stale detector
already knows it (and it knows it before
+ // closeConnectionsWith() gets called as it subscribes first).
+ // This is the only place where a new sender might be added to the map.
+ if (staleIdDetector.isIdStale(channel.launchId())) {
+ channel.close();
+ channels.remove(key, channel);
+ }
}
/**
@@ -486,4 +497,22 @@ public class ConnectionManager implements
ChannelCreationListener {
public void initiateStopping() {
stopping.set(true);
}
+
+ /**
+ * Closes physical connections with an Ignite node identified by the given
ID (it's not consistentId,
+ * it's ID that gets regenerated at each node restart).
+ *
+ * @param id ID of the node.
+ */
+ public void closeConnectionsWith(String id) {
+ List<Entry<ConnectorKey<String>, NettySender>> entriesToRemove =
channels.entrySet().stream()
+ .filter(entry -> entry.getValue().launchId().equals(id))
+ .collect(toList());
+
+ for (Entry<ConnectorKey<String>, NettySender> entry : entriesToRemove)
{
+ entry.getValue().close();
+
+ channels.remove(entry.getKey());
+ }
+ }
}
diff --git
a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectorKey.java
b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectorKey.java
index 9b0f39e4a9..8ec2a86411 100644
---
a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectorKey.java
+++
b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectorKey.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.network.netty;
+import org.apache.ignite.internal.tostring.S;
import org.apache.ignite.network.ChannelType;
/**
@@ -69,4 +70,9 @@ public class ConnectorKey<T> {
result = 31 * result + (type != null ? type.hashCode() : 0);
return result;
}
+
+ @Override
+ public String toString() {
+ return S.toString(ConnectorKey.class, this);
+ }
}
diff --git
a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/HandshakeHandler.java
b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/HandshakeHandler.java
index aebd477a27..6ab57d9e2f 100644
---
a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/HandshakeHandler.java
+++
b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/HandshakeHandler.java
@@ -117,7 +117,7 @@ public class HandshakeHandler extends
ChannelInboundHandlerAdapter {
* @param connectionIndex Index of the connection (corresponds to the
channel ID).
* @return Message handler.
*/
- public MessageHandler createMessageHandler(String remoteConsistentId,
short connectionIndex) {
- return new MessageHandler(messageListener, remoteConsistentId,
connectionIndex, serializationService);
+ public MessageHandler createMessageHandler(String remoteLaunchId, String
remoteConsistentId, short connectionIndex) {
+ return new MessageHandler(messageListener, remoteLaunchId,
remoteConsistentId, connectionIndex, serializationService);
}
}
diff --git
a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/InNetworkObject.java
b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/InNetworkObject.java
index 739fd07e5d..f7693685fe 100644
---
a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/InNetworkObject.java
+++
b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/InNetworkObject.java
@@ -28,6 +28,8 @@ public class InNetworkObject {
/** Message. */
private final NetworkMessage message;
+ private final String launchId;
+
/** Sender's consistent id. */
private final String consistentId;
@@ -37,8 +39,15 @@ public class InNetworkObject {
private final DescriptorRegistry registry;
/** Constructor. */
- public InNetworkObject(NetworkMessage message, String consistentId, short
connectionIndex, DescriptorRegistry registry) {
+ public InNetworkObject(
+ NetworkMessage message,
+ String launchId,
+ String consistentId,
+ short connectionIndex,
+ DescriptorRegistry registry
+ ) {
this.message = message;
+ this.launchId = launchId;
this.consistentId = consistentId;
this.connectionIndex = connectionIndex;
this.registry = registry;
@@ -53,6 +62,13 @@ public class InNetworkObject {
return message;
}
+ /**
+ * Returns node ID of the sender that does not survive node restart (aka
launch ID).
+ */
+ public String launchId() {
+ return launchId;
+ }
+
/**
* Returns consistent id.
*
diff --git
a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/MessageHandler.java
b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/MessageHandler.java
index e07633270f..80b0e6ec6b 100644
---
a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/MessageHandler.java
+++
b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/MessageHandler.java
@@ -34,6 +34,8 @@ public class MessageHandler extends
ChannelInboundHandlerAdapter {
/** Message listener. */
private final Consumer<InNetworkObject> messageListener;
+ private final String launchId;
+
/** Consistent id of the remote node. */
private final String consistentId;
@@ -45,12 +47,20 @@ public class MessageHandler extends
ChannelInboundHandlerAdapter {
* Constructor.
*
* @param messageListener Message listener.
+ * @param launchId Launch ID (this is the ID that does not survive a
restart) of the remote node.
* @param consistentId Consistent id of the remote node.
+ * @param connectionIndex Connection index (aka channel ID).
* @param serializationService Serialization service.
*/
- public MessageHandler(Consumer<InNetworkObject> messageListener, String
consistentId, short connectionIndex,
- PerSessionSerializationService serializationService) {
+ public MessageHandler(
+ Consumer<InNetworkObject> messageListener,
+ String launchId,
+ String consistentId,
+ short connectionIndex,
+ PerSessionSerializationService serializationService
+ ) {
this.messageListener = messageListener;
+ this.launchId = launchId;
this.consistentId = consistentId;
this.connectionIndex = connectionIndex;
this.serializationService = serializationService;
@@ -66,7 +76,7 @@ public class MessageHandler extends
ChannelInboundHandlerAdapter {
}
messageListener.accept(
- new InNetworkObject(message, consistentId, connectionIndex,
serializationService.compositeDescriptorRegistry())
+ new InNetworkObject(message, launchId, consistentId,
connectionIndex, serializationService.compositeDescriptorRegistry())
);
}
}
diff --git
a/modules/network/src/main/java/org/apache/ignite/internal/network/recovery/RecoveryClientHandshakeManager.java
b/modules/network/src/main/java/org/apache/ignite/internal/network/recovery/RecoveryClientHandshakeManager.java
index d44f082c00..9fd6fd6696 100644
---
a/modules/network/src/main/java/org/apache/ignite/internal/network/recovery/RecoveryClientHandshakeManager.java
+++
b/modules/network/src/main/java/org/apache/ignite/internal/network/recovery/RecoveryClientHandshakeManager.java
@@ -381,7 +381,7 @@ public class RecoveryClientHandshakeManager implements
HandshakeManager {
* @return New message handler.
*/
private MessageHandler createMessageHandler() {
- return handler.createMessageHandler(remoteConsistentId, connectionId);
+ return handler.createMessageHandler(remoteLaunchId.toString(),
remoteConsistentId, connectionId);
}
/**
diff --git
a/modules/network/src/main/java/org/apache/ignite/internal/network/recovery/RecoveryServerHandshakeManager.java
b/modules/network/src/main/java/org/apache/ignite/internal/network/recovery/RecoveryServerHandshakeManager.java
index af5aa45b4c..54edec00d3 100644
---
a/modules/network/src/main/java/org/apache/ignite/internal/network/recovery/RecoveryServerHandshakeManager.java
+++
b/modules/network/src/main/java/org/apache/ignite/internal/network/recovery/RecoveryServerHandshakeManager.java
@@ -389,7 +389,7 @@ public class RecoveryServerHandshakeManager implements
HandshakeManager {
* @return New message handler.
*/
private MessageHandler createMessageHandler() {
- return handler.createMessageHandler(remoteConsistentId,
remoteChannelId);
+ return handler.createMessageHandler(remoteLaunchId.toString(),
remoteConsistentId, remoteChannelId);
}
/** {@inheritDoc} */
diff --git
a/modules/network/src/main/java/org/apache/ignite/internal/network/recovery/VaultStateIds.java
b/modules/network/src/main/java/org/apache/ignite/internal/network/recovery/VaultStaleIds.java
similarity index 94%
rename from
modules/network/src/main/java/org/apache/ignite/internal/network/recovery/VaultStateIds.java
rename to
modules/network/src/main/java/org/apache/ignite/internal/network/recovery/VaultStaleIds.java
index e011b30978..2ad20206e4 100644
---
a/modules/network/src/main/java/org/apache/ignite/internal/network/recovery/VaultStateIds.java
+++
b/modules/network/src/main/java/org/apache/ignite/internal/network/recovery/VaultStaleIds.java
@@ -30,7 +30,7 @@ import org.apache.ignite.internal.vault.VaultManager;
/**
* {@link StaleIds} implementating using Vault as a persistent storage.
*/
-public class VaultStateIds implements StaleIds {
+public class VaultStaleIds implements StaleIds {
private static final ByteArray STALE_IDS_KEY = new
ByteArray("network.staleIds");
private static final int DEFAULT_MAX_IDS_TO_REMEMBER = 10_000;
@@ -41,11 +41,11 @@ public class VaultStateIds implements StaleIds {
private Set<String> staleIds;
- public VaultStateIds(VaultManager vaultManager) {
+ public VaultStaleIds(VaultManager vaultManager) {
this(vaultManager, DEFAULT_MAX_IDS_TO_REMEMBER);
}
- public VaultStateIds(VaultManager vaultManager, int maxIdsToRemember) {
+ public VaultStaleIds(VaultManager vaultManager, int maxIdsToRemember) {
this.vaultManager = vaultManager;
this.maxIdsToRemember = maxIdsToRemember;
}
diff --git
a/modules/network/src/main/java/org/apache/ignite/network/DefaultMessagingService.java
b/modules/network/src/main/java/org/apache/ignite/network/DefaultMessagingService.java
index 22115aa56f..63c0ac9b62 100644
---
a/modules/network/src/main/java/org/apache/ignite/network/DefaultMessagingService.java
+++
b/modules/network/src/main/java/org/apache/ignite/network/DefaultMessagingService.java
@@ -48,6 +48,7 @@ import
org.apache.ignite.internal.network.message.ScaleCubeMessage;
import org.apache.ignite.internal.network.netty.ConnectionManager;
import org.apache.ignite.internal.network.netty.InNetworkObject;
import org.apache.ignite.internal.network.netty.NettySender;
+import org.apache.ignite.internal.network.recovery.StaleIdDetector;
import
org.apache.ignite.internal.network.serialization.ClassDescriptorRegistry;
import org.apache.ignite.internal.network.serialization.DescriptorRegistry;
import
org.apache.ignite.internal.network.serialization.marshal.UserObjectMarshaller;
@@ -67,6 +68,8 @@ public class DefaultMessagingService extends
AbstractMessagingService {
/** Topology service. */
private final TopologyService topologyService;
+ private final StaleIdDetector staleIdDetector;
+
/** User object marshaller. */
private final UserObjectMarshaller marshaller;
@@ -97,6 +100,7 @@ public class DefaultMessagingService extends
AbstractMessagingService {
*
* @param factory Network messages factory.
* @param topologyService Topology service.
+ * @param staleIdDetector Used to detect stale node IDs.
* @param classDescriptorRegistry Descriptor registry.
* @param marshaller Marshaller.
*/
@@ -104,11 +108,13 @@ public class DefaultMessagingService extends
AbstractMessagingService {
String nodeName,
NetworkMessagesFactory factory,
TopologyService topologyService,
+ StaleIdDetector staleIdDetector,
ClassDescriptorRegistry classDescriptorRegistry,
UserObjectMarshaller marshaller
) {
this.factory = factory;
this.topologyService = topologyService;
+ this.staleIdDetector = staleIdDetector;
this.classDescriptorRegistry = classDescriptorRegistry;
this.marshaller = marshaller;
@@ -220,7 +226,7 @@ public class DefaultMessagingService extends
AbstractMessagingService {
NetworkMessage message = correlationId != null ?
responseFromMessage(msg, correlationId) : msg;
- return sendMessage0(recipient.name(), type, recipientAddress, message);
+ return sendViaNetwork(recipient.name(), type, recipientAddress,
message);
}
private boolean shouldDropMessage(ClusterNode recipient, NetworkMessage
msg) {
@@ -264,7 +270,7 @@ public class DefaultMessagingService extends
AbstractMessagingService {
InvokeRequest message = requestFromMessage(msg, correlationId);
- return sendMessage0(recipient.name(), type, recipientAddress,
message).thenCompose(unused -> responseFuture);
+ return sendViaNetwork(recipient.name(), type, recipientAddress,
message).thenCompose(unused -> responseFuture);
}
/**
@@ -277,21 +283,21 @@ public class DefaultMessagingService extends
AbstractMessagingService {
*
* @return Future of the send operation.
*/
- private CompletableFuture<Void> sendMessage0(
+ private CompletableFuture<Void> sendViaNetwork(
@Nullable String consistentId,
ChannelType type,
InetSocketAddress addr,
NetworkMessage message
) {
if (isInNetworkThread()) {
- return CompletableFuture.supplyAsync(() ->
sendMessage0(consistentId, type, addr, message), outboundExecutor)
+ return CompletableFuture.supplyAsync(() ->
sendViaNetwork(consistentId, type, addr, message), outboundExecutor)
.thenCompose(Function.identity());
}
List<ClassDescriptorMessage> descriptors;
try {
- descriptors = beforeRead(message);
+ descriptors = prepareMarshal(message);
} catch (Exception e) {
return failedFuture(new IgniteException("Failed to marshal
message: " + e.getMessage(), e));
}
@@ -300,7 +306,7 @@ public class DefaultMessagingService extends
AbstractMessagingService {
.thenComposeToCompletable(sender -> sender.send(new
OutNetworkObject(message, descriptors)));
}
- private List<ClassDescriptorMessage> beforeRead(NetworkMessage msg) throws
Exception {
+ private List<ClassDescriptorMessage> prepareMarshal(NetworkMessage msg)
throws Exception {
IntSet ids = new IntOpenHashSet();
msg.prepareMarshal(ids, marshaller);
@@ -345,6 +351,11 @@ public class DefaultMessagingService extends
AbstractMessagingService {
}
private void handleIncomingMessage(InNetworkObject obj) {
+ if (senderIdIsStale(obj)) {
+ LOG.info("Sender ID {} ({}) is stale, so skipping message
handling: {}", obj.launchId(), obj.consistentId(), obj.message());
+ return;
+ }
+
NetworkMessage msg = obj.message();
DescriptorRegistry registry = obj.registry();
try {
@@ -380,6 +391,10 @@ public class DefaultMessagingService extends
AbstractMessagingService {
}
}
+ private boolean senderIdIsStale(InNetworkObject obj) {
+ return staleIdDetector.isIdStale(obj.launchId());
+ }
+
private static void logAndRethrowIfError(InNetworkObject obj, Throwable e)
{
if (e instanceof UnresolvableConsistentIdException && obj.message()
instanceof InvokeRequest) {
LOG.info("onMessage() failed while processing {} from {} as the
sender has left the topology",
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 68e0de4097..f8124e835b 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
@@ -83,16 +83,26 @@ public class ScaleCubeClusterServiceFactory {
MessageSerializationRegistry serializationRegistry,
StaleIds staleIds
) {
- var messageFactory = new NetworkMessagesFactory();
-
var topologyService = new ScaleCubeTopologyService();
+ // Adding this handler as the first handler to make sure that StaleIds
is at least up-to-date as any
+ // other component that watches topology events.
+ topologyService.addEventHandler(new TopologyEventHandler() {
+ @Override
+ public void onDisappeared(ClusterNode member) {
+ staleIds.markAsStale(member.id());
+ }
+ });
+
+ var messageFactory = new NetworkMessagesFactory();
+
UserObjectSerializationContext userObjectSerialization =
createUserObjectSerializationContext();
var messagingService = new DefaultMessagingService(
consistentId,
messageFactory,
topologyService,
+ staleIds,
userObjectSerialization.descriptorRegistry(),
userObjectSerialization.marshaller()
);
@@ -114,7 +124,7 @@ public class ScaleCubeClusterServiceFactory {
NetworkView configView = networkConfiguration.value();
- connectionMgr = new ConnectionManager(
+ ConnectionManager connectionMgr = new ConnectionManager(
configView,
serializationService,
launchId,
@@ -125,6 +135,13 @@ public class ScaleCubeClusterServiceFactory {
connectionMgr.start();
+ topologyService.addEventHandler(new TopologyEventHandler() {
+ @Override
+ public void onDisappeared(ClusterNode member) {
+ connectionMgr.closeConnectionsWith(member.id());
+ }
+ });
+
var transport = new ScaleCubeDirectMarshallerTransport(
connectionMgr.localAddress(),
messagingService,
@@ -133,7 +150,7 @@ public class ScaleCubeClusterServiceFactory {
);
NodeFinder finder =
NodeFinderFactory.createNodeFinder(configView.nodeFinder());
- cluster = new
ClusterImpl(clusterConfig(configView.membership()))
+ ClusterImpl cluster = new
ClusterImpl(clusterConfig(configView.membership()))
.handler(cl -> new ClusterMessageHandler() {
/** {@inheritDoc} */
@Override
@@ -149,24 +166,20 @@ public class ScaleCubeClusterServiceFactory {
.transport(opts ->
opts.transportFactory(transportConfig -> transport))
.membership(opts ->
opts.seedMembers(parseAddresses(finder.findNodes())));
- shutdownFuture = cluster.onShutdown().toFuture();
+ this.shutdownFuture = cluster.onShutdown().toFuture();
// resolve cyclic dependencies
topologyService.setCluster(cluster);
messagingService.setConnectionManager(connectionMgr);
- topologyService.addEventHandler(new TopologyEventHandler() {
- @Override
- public void onDisappeared(ClusterNode member) {
- staleIds.markAsStale(member.id());
- }
- });
-
cluster.startAwait();
// emit an artificial event as if the local member has joined
the topology (ScaleCube doesn't do that)
var localMembershipEvent = createAdded(cluster.member(), null,
System.currentTimeMillis());
topologyService.onMembershipEvent(localMembershipEvent);
+
+ this.cluster = cluster;
+ this.connectionMgr = connectionMgr;
}
/** {@inheritDoc} */
diff --git
a/modules/network/src/test/java/org/apache/ignite/internal/network/recovery/VaultStateIdsTest.java
b/modules/network/src/test/java/org/apache/ignite/internal/network/recovery/VaultStaleIdsTest.java
similarity index 95%
rename from
modules/network/src/test/java/org/apache/ignite/internal/network/recovery/VaultStateIdsTest.java
rename to
modules/network/src/test/java/org/apache/ignite/internal/network/recovery/VaultStaleIdsTest.java
index e0ed786b8a..c3c4b9b38c 100644
---
a/modules/network/src/test/java/org/apache/ignite/internal/network/recovery/VaultStateIdsTest.java
+++
b/modules/network/src/test/java/org/apache/ignite/internal/network/recovery/VaultStaleIdsTest.java
@@ -42,17 +42,17 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
@ExtendWith(MockitoExtension.class)
-class VaultStateIdsTest extends BaseIgniteAbstractTest {
+class VaultStaleIdsTest extends BaseIgniteAbstractTest {
@Mock
private VaultManager vaultManager;
private final ByteArray staleIdsKey = new ByteArray("network.staleIds");
- private VaultStateIds staleIds;
+ private VaultStaleIds staleIds;
@BeforeEach
void createObjectToTest() {
- staleIds = new VaultStateIds(vaultManager);
+ staleIds = new VaultStaleIds(vaultManager);
}
@Test
@@ -92,7 +92,7 @@ class VaultStateIdsTest extends BaseIgniteAbstractTest {
@Test
void respectsMaxIdsLimit() {
- staleIds = new VaultStateIds(vaultManager, 2);
+ staleIds = new VaultStaleIds(vaultManager, 2);
doReturn(nullCompletedFuture()).when(vaultManager).get(staleIdsKey);
diff --git
a/modules/network/src/test/java/org/apache/ignite/network/DefaultMessagingServiceTest.java
b/modules/network/src/test/java/org/apache/ignite/network/DefaultMessagingServiceTest.java
index c72990c4c9..c5e722155e 100644
---
a/modules/network/src/test/java/org/apache/ignite/network/DefaultMessagingServiceTest.java
+++
b/modules/network/src/test/java/org/apache/ignite/network/DefaultMessagingServiceTest.java
@@ -436,6 +436,8 @@ class DefaultMessagingServiceTest extends
BaseIgniteAbstractTest {
Runnable beforeHandshake,
MessageSerializationRegistry registry
) {
+ StaleIdDetector staleIdDetector = new AllIdsAreFresh();
+
ClassDescriptorRegistry classDescriptorRegistry = new
ClassDescriptorRegistry();
ClassDescriptorFactory classDescriptorFactory = new
ClassDescriptorFactory(classDescriptorRegistry);
UserObjectMarshaller marshaller = new
DefaultUserObjectMarshaller(classDescriptorRegistry, classDescriptorFactory);
@@ -444,6 +446,7 @@ class DefaultMessagingServiceTest extends
BaseIgniteAbstractTest {
node.name(),
networkMessagesFactory,
topologyService,
+ staleIdDetector,
classDescriptorRegistry,
marshaller
);
@@ -458,8 +461,6 @@ class DefaultMessagingServiceTest extends
BaseIgniteAbstractTest {
NettyBootstrapFactory bootstrapFactory = new
NettyBootstrapFactory(networkConfig, eventLoopGroupNamePrefix);
bootstrapFactory.start();
- StaleIdDetector staleIdDetector = new AllIdsAreFresh();
-
UUID launchId = UUID.randomUUID();
ConnectionManager connectionManager = new ConnectionManager(
networkConfig.value(),
diff --git
a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
index 6cbfcb7eea..5fce588f45 100644
---
a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
+++
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
@@ -106,7 +106,7 @@ import
org.apache.ignite.internal.metastorage.server.persistence.RocksDbKeyValue
import org.apache.ignite.internal.metastorage.server.raft.MetastorageGroupId;
import org.apache.ignite.internal.metrics.MetricManager;
import org.apache.ignite.internal.network.configuration.NetworkConfiguration;
-import org.apache.ignite.internal.network.recovery.VaultStateIds;
+import org.apache.ignite.internal.network.recovery.VaultStaleIds;
import org.apache.ignite.internal.placementdriver.PlacementDriverManager;
import org.apache.ignite.internal.raft.Loza;
import org.apache.ignite.internal.raft.Peer;
@@ -250,7 +250,7 @@ public class ItIgniteNodeRestartTest extends
BaseIgniteRestartTest {
networkConfiguration,
nettyBootstrapFactory,
defaultSerializationRegistry(),
- new VaultStateIds(vault)
+ new VaultStaleIds(vault)
);
var hybridClock = new HybridClockImpl();
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 fe896756fa..882429c919 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
@@ -109,7 +109,7 @@ import
org.apache.ignite.internal.metrics.configuration.MetricConfiguration;
import org.apache.ignite.internal.metrics.sources.JvmMetricSource;
import org.apache.ignite.internal.network.configuration.NetworkConfiguration;
import
org.apache.ignite.internal.network.configuration.NetworkConfigurationSchema;
-import org.apache.ignite.internal.network.recovery.VaultStateIds;
+import org.apache.ignite.internal.network.recovery.VaultStaleIds;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.PlacementDriverManager;
import org.apache.ignite.internal.raft.Loza;
@@ -380,7 +380,7 @@ public class IgniteImpl implements Ignite {
networkConfiguration,
nettyBootstrapFactory,
serializationRegistry,
- new VaultStateIds(vaultMgr)
+ new VaultStaleIds(vaultMgr)
);
clock = new HybridClockImpl();
