This is an automated email from the ASF dual-hosted git repository.
rpuch 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 a5de6e179e IGNITE-22881 Handle logical topology event version reset in
DistributionZoneManager (#4318)
a5de6e179e is described below
commit a5de6e179e6069938479949223e611d774ac1393
Author: Roman Puchkovskiy <[email protected]>
AuthorDate: Wed Sep 4 20:34:28 2024 +0400
IGNITE-22881 Handle logical topology event version reset in
DistributionZoneManager (#4318)
1. clusterId is included in LogicalTopologySnapshot
2. In DistributionZoneManager, the clusterId is stored when a snapshot with
version=1 is stored, and checked for equality with the already-stored clusterId
if there is an entry with the same version
3. There is no need to check explicitly which clusterId is 'older' or
'newer' because, if a 'newer' clusterId is written to the Metastorage, the
'older' one cannot be written anymore (due to guarantees given by the CMG
repair logic)
This allows to handle version reset gracefully while keeping the 'only
first attempt to write for the same event actually writes' semantics.
---
.../management/raft/ItCmgRaftServiceTest.java | 6 +-
.../management/topology/LogicalTopologyImpl.java | 21 ++++--
.../topology/api/LogicalTopologySnapshot.java | 28 +++++++-
.../management/raft/CmgRaftGroupListenerTest.java | 4 +-
.../topology/LogicalTopologyImplTest.java | 4 +-
.../internal/cluster/management/MockNode.java | 4 +-
.../org/apache/ignite/internal/util/ByteUtils.java | 23 +++++++
.../apache/ignite/internal/util/ByteUtilsTest.java | 11 ++++
modules/distribution-zones/build.gradle | 1 +
...niteDistributionZoneManagerNodeRestartTest.java | 4 +-
.../distributionzones/DistributionZoneManager.java | 39 +++++++----
.../distributionzones/DistributionZonesUtil.java | 61 ++++++++++++++---
.../BaseDistributionZoneManagerTest.java | 4 +-
...butionZoneManagerLogicalTopologyEventsTest.java | 7 +-
.../DistributionZoneCausalityDataNodesTest.java | 2 +-
.../ItMetaStorageMultipleNodesAbstractTest.java | 4 +-
.../metastorage/impl/ItMetaStorageWatchTest.java | 4 +-
.../replicator/ItReplicaLifecycleTest.java | 4 +-
.../ItPlacementDriverReplicaSideTest.java | 4 +-
.../AbstractTopologyAwareGroupServiceTest.java | 6 +-
...stImpl.java => TestLogicalTopologyService.java} | 16 +++--
.../ItDistributedConfigurationPropertiesTest.java | 4 +-
.../ItDistributedConfigurationStorageTest.java | 4 +-
.../runner/app/ItIgniteNodeRestartTest.java | 2 +-
.../org/apache/ignite/internal/app/IgniteImpl.java | 2 +-
.../java/org/apache/ignite/internal/Cluster.java | 16 ++++-
modules/system-disaster-recovery/build.gradle | 3 +
.../disaster/system/ItCmgDisasterRecoveryTest.java | 77 ++++++++++++++++++++++
.../rebalance/ItRebalanceDistributedTest.java | 2 +-
.../apache/ignite/distributed/ItTxTestCluster.java | 11 +++-
30 files changed, 317 insertions(+), 61 deletions(-)
diff --git
a/modules/cluster-management/src/integrationTest/java/org/apache/ignite/internal/cluster/management/raft/ItCmgRaftServiceTest.java
b/modules/cluster-management/src/integrationTest/java/org/apache/ignite/internal/cluster/management/raft/ItCmgRaftServiceTest.java
index 5dd9e57670..3993118d97 100644
---
a/modules/cluster-management/src/integrationTest/java/org/apache/ignite/internal/cluster/management/raft/ItCmgRaftServiceTest.java
+++
b/modules/cluster-management/src/integrationTest/java/org/apache/ignite/internal/cluster/management/raft/ItCmgRaftServiceTest.java
@@ -38,6 +38,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import org.apache.ignite.internal.cluster.management.ClusterState;
@@ -59,6 +60,7 @@ import
org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.lang.NodeStoppingException;
import org.apache.ignite.internal.manager.ComponentContext;
import org.apache.ignite.internal.network.ClusterService;
+import org.apache.ignite.internal.network.ConstantClusterIdSupplier;
import org.apache.ignite.internal.network.NodeFinder;
import org.apache.ignite.internal.network.StaticNodeFinder;
import org.apache.ignite.internal.properties.IgniteProductVersion;
@@ -119,7 +121,7 @@ public class ItCmgRaftServiceTest extends
BaseIgniteAbstractTest {
workingDir.raftLogPath()
);
this.raftManager = TestLozaFactory.create(clusterService,
raftConfiguration, new HybridClockImpl());
- this.logicalTopology = new
LogicalTopologyImpl(clusterStateStorage);
+ this.logicalTopology = new
LogicalTopologyImpl(clusterStateStorage, new
ConstantClusterIdSupplier(UUID.randomUUID()));
}
void start() {
@@ -153,7 +155,7 @@ public class ItCmgRaftServiceTest extends
BaseIgniteAbstractTest {
configuration,
new CmgRaftGroupListener(
clusterStateStorageMgr,
- new
LogicalTopologyImpl(clusterStateStorage),
+ logicalTopology,
new
ValidationManager(clusterStateStorageMgr, logicalTopology),
term -> {}
),
diff --git
a/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/topology/LogicalTopologyImpl.java
b/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/topology/LogicalTopologyImpl.java
index 7af82b7039..ec7ab64b8b 100644
---
a/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/topology/LogicalTopologyImpl.java
+++
b/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/topology/LogicalTopologyImpl.java
@@ -29,6 +29,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;
import org.apache.ignite.internal.cluster.management.raft.ClusterStateStorage;
@@ -37,6 +38,7 @@ import
org.apache.ignite.internal.cluster.management.topology.api.LogicalTopolog
import
org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologySnapshot;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.network.ClusterIdSupplier;
/**
* Implementation of {@link LogicalTopology}.
@@ -49,10 +51,13 @@ public class LogicalTopologyImpl implements LogicalTopology
{
private final ClusterStateStorage storage;
+ private final ClusterIdSupplier clusterIdSupplier;
+
private final List<LogicalTopologyEventListener> listeners = new
CopyOnWriteArrayList<>();
- public LogicalTopologyImpl(ClusterStateStorage storage) {
+ public LogicalTopologyImpl(ClusterStateStorage storage, ClusterIdSupplier
clusterIdSupplier) {
this.storage = storage;
+ this.clusterIdSupplier = clusterIdSupplier;
}
@Override
@@ -94,7 +99,7 @@ public class LogicalTopologyImpl implements LogicalTopology {
}
// This is an update. First simulate disappearance, then
appearance will be fired.
- snapshot = new LogicalTopologySnapshot(snapshot.version() + 1,
mapByName.values());
+ snapshot = new LogicalTopologySnapshot(snapshot.version() + 1,
mapByName.values(), requiredClusterId());
if (LOG.isInfoEnabled()) {
LOG.info("Node removed from logical topology [node={},
topology={}]", nodeToPut, snapshot);
@@ -106,7 +111,7 @@ public class LogicalTopologyImpl implements LogicalTopology
{
mapByName.put(nodeToPut.name(), nodeToPut);
- snapshot = new LogicalTopologySnapshot(snapshot.version() + 1,
mapByName.values());
+ snapshot = new LogicalTopologySnapshot(snapshot.version() + 1,
mapByName.values(), requiredClusterId());
if (LOG.isInfoEnabled()) {
LOG.info("Node added to logical topology [node={}, topology={}]",
nodeToPut, snapshot);
@@ -122,6 +127,14 @@ public class LogicalTopologyImpl implements
LogicalTopology {
fireNodeJoined(nodeToPut, snapshot);
}
+ private UUID requiredClusterId() {
+ UUID clusterId = clusterIdSupplier.clusterId();
+
+ assert clusterId != null : "clusterId cannot be null when commands are
already being executed by the CMG state machine";
+
+ return clusterId;
+ }
+
private void saveSnapshotToStorage(LogicalTopologySnapshot newTopology) {
storage.put(LOGICAL_TOPOLOGY_KEY, toBytes(newTopology));
}
@@ -144,7 +157,7 @@ public class LogicalTopologyImpl implements LogicalTopology
{
LogicalNode removedNode = mapById.remove(nodeToRemove.id());
if (removedNode != null) {
- snapshot = new LogicalTopologySnapshot(snapshot.version() + 1,
mapById.values());
+ snapshot = new LogicalTopologySnapshot(snapshot.version() + 1,
mapById.values(), requiredClusterId());
if (LOG.isInfoEnabled()) {
LOG.info("Node removed from logical topology [node={},
topology={}]", removedNode, snapshot);
diff --git
a/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/topology/api/LogicalTopologySnapshot.java
b/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/topology/api/LogicalTopologySnapshot.java
index e423a90f9b..ed2823b5dc 100644
---
a/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/topology/api/LogicalTopologySnapshot.java
+++
b/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/topology/api/LogicalTopologySnapshot.java
@@ -22,8 +22,10 @@ import static java.util.Collections.emptySet;
import java.io.Serializable;
import java.util.Collection;
import java.util.Set;
+import java.util.UUID;
import org.apache.ignite.internal.tostring.IgniteToStringInclude;
import org.apache.ignite.internal.tostring.S;
+import org.jetbrains.annotations.TestOnly;
/**
* A snapshot of a logical topology as seen locally. Includes nodes
participating in the logical topology and the version
@@ -34,17 +36,32 @@ import org.apache.ignite.internal.tostring.S;
public class LogicalTopologySnapshot implements Serializable {
private static final long serialVersionUID = 0L;
+ /** Version that first topology snapshot in history will have. */
+ public static final long FIRST_VERSION = 1;
+
/** Initial 'topology' for an empty cluster (before any node has joined).
*/
- public static final LogicalTopologySnapshot INITIAL = new
LogicalTopologySnapshot(0, emptySet());
+ public static final LogicalTopologySnapshot INITIAL = new
LogicalTopologySnapshot(FIRST_VERSION - 1, emptySet(), new UUID(0, 0));
private final long version;
@IgniteToStringInclude
private final Set<LogicalNode> nodes;
- public LogicalTopologySnapshot(long version, Collection<LogicalNode>
nodes) {
+ private final UUID clusterId;
+
+ /** Constructor. */
+ public LogicalTopologySnapshot(long version, Collection<LogicalNode>
nodes, UUID clusterId) {
this.version = version;
this.nodes = Set.copyOf(nodes);
+ this.clusterId = clusterId;
+ }
+
+ /**
+ * Creates a snapshot with a random cluster ID (only suitable for tests
that don't care about clusterIds).
+ */
+ @TestOnly
+ public LogicalTopologySnapshot(long version, Collection<LogicalNode>
nodes) {
+ this(version, nodes, UUID.randomUUID());
}
/**
@@ -61,6 +78,13 @@ public class LogicalTopologySnapshot implements Serializable
{
return nodes;
}
+ /**
+ * Returns ID of the cluster in which this topology snapshot was created.
+ */
+ public UUID clusterId() {
+ return clusterId;
+ }
+
@Override
public String toString() {
return S.toString(LogicalTopologySnapshot.class, this);
diff --git
a/modules/cluster-management/src/test/java/org/apache/ignite/internal/cluster/management/raft/CmgRaftGroupListenerTest.java
b/modules/cluster-management/src/test/java/org/apache/ignite/internal/cluster/management/raft/CmgRaftGroupListenerTest.java
index 42a9b77ff0..5d6ca0a4b8 100644
---
a/modules/cluster-management/src/test/java/org/apache/ignite/internal/cluster/management/raft/CmgRaftGroupListenerTest.java
+++
b/modules/cluster-management/src/test/java/org/apache/ignite/internal/cluster/management/raft/CmgRaftGroupListenerTest.java
@@ -40,6 +40,7 @@ import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
+import java.util.UUID;
import java.util.function.LongConsumer;
import org.apache.ignite.internal.cluster.management.ClusterState;
import org.apache.ignite.internal.cluster.management.ClusterTag;
@@ -50,6 +51,7 @@ import
org.apache.ignite.internal.cluster.management.topology.LogicalTopology;
import
org.apache.ignite.internal.cluster.management.topology.LogicalTopologyImpl;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
import org.apache.ignite.internal.manager.ComponentContext;
+import org.apache.ignite.internal.network.ConstantClusterIdSupplier;
import org.apache.ignite.internal.properties.IgniteProductVersion;
import org.apache.ignite.internal.raft.Command;
import org.apache.ignite.internal.raft.service.CommandClosure;
@@ -67,7 +69,7 @@ public class CmgRaftGroupListenerTest extends
BaseIgniteAbstractTest {
private final LongConsumer onLogicalTopologyChanged =
mock(LongConsumer.class);
- private final LogicalTopology logicalTopology = spy(new
LogicalTopologyImpl(storage));
+ private final LogicalTopology logicalTopology = spy(new
LogicalTopologyImpl(storage, new ConstantClusterIdSupplier(UUID.randomUUID())));
private CmgRaftGroupListener listener;
diff --git
a/modules/cluster-management/src/test/java/org/apache/ignite/internal/cluster/management/topology/LogicalTopologyImplTest.java
b/modules/cluster-management/src/test/java/org/apache/ignite/internal/cluster/management/topology/LogicalTopologyImplTest.java
index 51cb10363f..e39d6fcff8 100644
---
a/modules/cluster-management/src/test/java/org/apache/ignite/internal/cluster/management/topology/LogicalTopologyImplTest.java
+++
b/modules/cluster-management/src/test/java/org/apache/ignite/internal/cluster/management/topology/LogicalTopologyImplTest.java
@@ -46,6 +46,7 @@ import java.util.Collection;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.apache.ignite.internal.cluster.management.raft.ClusterStateStorage;
import
org.apache.ignite.internal.cluster.management.raft.TestClusterStateStorage;
@@ -53,6 +54,7 @@ import
org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
import
org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyEventListener;
import
org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologySnapshot;
import org.apache.ignite.internal.manager.ComponentContext;
+import org.apache.ignite.internal.network.ConstantClusterIdSupplier;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.apache.ignite.internal.testframework.WorkDirectory;
import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
@@ -97,7 +99,7 @@ class LogicalTopologyImplTest extends BaseIgniteAbstractTest {
void setUp() {
assertThat(storage.startAsync(new ComponentContext()),
willCompleteSuccessfully());
- topology = new LogicalTopologyImpl(storage);
+ topology = new LogicalTopologyImpl(storage, new
ConstantClusterIdSupplier(UUID.randomUUID()));
topology.addEventListener(listener);
}
diff --git
a/modules/cluster-management/src/testFixtures/java/org/apache/ignite/internal/cluster/management/MockNode.java
b/modules/cluster-management/src/testFixtures/java/org/apache/ignite/internal/cluster/management/MockNode.java
index 3a9ffd847a..1d50edb560 100644
---
a/modules/cluster-management/src/testFixtures/java/org/apache/ignite/internal/cluster/management/MockNode.java
+++
b/modules/cluster-management/src/testFixtures/java/org/apache/ignite/internal/cluster/management/MockNode.java
@@ -30,6 +30,7 @@ import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import
org.apache.ignite.internal.cluster.management.configuration.NodeAttributesConfiguration;
import
org.apache.ignite.internal.cluster.management.raft.RocksDbClusterStateStorage;
@@ -45,6 +46,7 @@ import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.manager.ComponentContext;
import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.network.ClusterService;
+import org.apache.ignite.internal.network.ConstantClusterIdSupplier;
import org.apache.ignite.internal.network.NodeFinder;
import org.apache.ignite.internal.network.utils.ClusterServiceTestUtils;
import org.apache.ignite.internal.raft.RaftGroupOptionsConfigurer;
@@ -132,7 +134,7 @@ public class MockNode {
new ClusterInitializer(clusterService, hocon -> hocon, new
TestConfigurationValidator()),
raftManager,
clusterStateStorage,
- new LogicalTopologyImpl(clusterStateStorage),
+ new LogicalTopologyImpl(clusterStateStorage, new
ConstantClusterIdSupplier(UUID.randomUUID())),
new NodeAttributesCollector(nodeAttributes,
storageProfilesConfiguration),
failureProcessor,
clusterIdHolder,
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/util/ByteUtils.java
b/modules/core/src/main/java/org/apache/ignite/internal/util/ByteUtils.java
index 86563608ed..1af345a207 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/ByteUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/ByteUtils.java
@@ -30,6 +30,7 @@ import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
+import java.util.UUID;
import org.apache.ignite.internal.lang.IgniteInternalException;
import org.jetbrains.annotations.Nullable;
@@ -344,4 +345,26 @@ public class ByteUtils {
return unmodifiableList(result);
}
+
+ /**
+ * Converts a UUID to bytes.
+ */
+ public static byte[] uuidToBytes(UUID uuid) {
+ byte[] bytes = new byte[2 * Long.BYTES];
+
+ putLongToBytes(uuid.getMostSignificantBits(), bytes, 0);
+ putLongToBytes(uuid.getLeastSignificantBits(), bytes, Long.BYTES);
+
+ return bytes;
+ }
+
+ /**
+ * Converts a UUID back from bytes.
+ */
+ public static UUID bytesToUuid(byte[] bytes) {
+ long higher = bytesToLong(bytes, 0);
+ long lower = bytesToLong(bytes, Long.BYTES);
+
+ return new UUID(higher, lower);
+ }
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/util/ByteUtilsTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/util/ByteUtilsTest.java
index 2141595d9d..f8fa207b40 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/util/ByteUtilsTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/util/ByteUtilsTest.java
@@ -20,13 +20,16 @@ package org.apache.ignite.internal.util;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.ignite.internal.util.ByteUtils.bytesToIntKeepingOrder;
import static
org.apache.ignite.internal.util.ByteUtils.bytesToLongKeepingOrder;
+import static org.apache.ignite.internal.util.ByteUtils.bytesToUuid;
import static org.apache.ignite.internal.util.ByteUtils.intToBytesKeepingOrder;
import static
org.apache.ignite.internal.util.ByteUtils.longToBytesKeepingOrder;
import static org.apache.ignite.internal.util.ByteUtils.stringFromBytes;
import static org.apache.ignite.internal.util.ByteUtils.stringToBytes;
import static org.apache.ignite.internal.util.ByteUtils.toByteArray;
import static org.apache.ignite.internal.util.ByteUtils.toByteArrayList;
+import static org.apache.ignite.internal.util.ByteUtils.uuidToBytes;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.lessThan;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -34,6 +37,7 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import java.nio.ByteBuffer;
import java.util.List;
+import java.util.UUID;
import org.junit.jupiter.api.Test;
/** For {@link ByteUtils} testing. */
@@ -126,4 +130,11 @@ public class ByteUtilsTest {
assertEquals(1, bytes.size());
assertArrayEquals(new byte[]{6, 8}, bytes.get(0));
}
+
+ @Test
+ void uuidToBytesFromBytesGivesOriginalUuid() {
+ UUID original = UUID.randomUUID();
+
+ assertThat(bytesToUuid(uuidToBytes(original)), is(original));
+ }
}
diff --git a/modules/distribution-zones/build.gradle
b/modules/distribution-zones/build.gradle
index 7c12418d6e..e7a915d884 100644
--- a/modules/distribution-zones/build.gradle
+++ b/modules/distribution-zones/build.gradle
@@ -64,6 +64,7 @@ dependencies {
testImplementation(testFixtures(project(':ignite-storage-api')))
testImplementation(testFixtures(project(':ignite-catalog')))
testImplementation(testFixtures(project(':ignite-table')))
+ testImplementation(testFixtures(project(':ignite-network')))
testFixturesImplementation libs.jetbrains.annotations
testFixturesImplementation libs.mockito.core
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 aa3d1f953c..3055118c9f 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
@@ -69,6 +69,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CountDownLatch;
@@ -188,6 +189,7 @@ public class ItIgniteDistributionZoneManagerNodeRestartTest
extends BaseIgniteRe
var clusterStateStorage = new TestClusterStateStorage();
var clusterIdService = new ClusterIdService(vault);
+ clusterIdService.clusterId(UUID.randomUUID());
ConfigurationModules modules = loadConfigurationModules(log,
Thread.currentThread().getContextClassLoader());
@@ -228,7 +230,7 @@ public class ItIgniteDistributionZoneManagerNodeRestartTest
extends BaseIgniteRe
mock(FailureProcessor.class)
);
- var logicalTopology = new LogicalTopologyImpl(clusterStateStorage);
+ var logicalTopology = new LogicalTopologyImpl(clusterStateStorage,
clusterIdService);
var cmgManager = mock(ClusterManagementGroupManager.class);
diff --git
a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
index 795cb2f190..92560a01b7 100644
---
a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
+++
b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.distributionzones;
+import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Collections.emptySet;
import static java.util.Collections.unmodifiableSet;
import static java.util.concurrent.CompletableFuture.allOf;
@@ -42,11 +43,13 @@ import static
org.apache.ignite.internal.distributionzones.DistributionZonesUtil
import static
org.apache.ignite.internal.distributionzones.DistributionZonesUtil.updateDataNodesAndScaleUpTriggerKey;
import static
org.apache.ignite.internal.distributionzones.DistributionZonesUtil.updateDataNodesAndTriggerKeys;
import static
org.apache.ignite.internal.distributionzones.DistributionZonesUtil.updateLogicalTopologyAndVersion;
+import static
org.apache.ignite.internal.distributionzones.DistributionZonesUtil.updateLogicalTopologyAndVersionAndClusterId;
import static
org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zoneDataNodesKey;
import static
org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zoneScaleDownChangeTriggerKey;
import static
org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zoneScaleUpChangeTriggerKey;
import static
org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zoneTopologyAugmentation;
import static
org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesLastHandledTopology;
+import static
org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesLogicalTopologyClusterIdKey;
import static
org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesLogicalTopologyKey;
import static
org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesLogicalTopologyPrefix;
import static
org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesLogicalTopologyVersionKey;
@@ -61,6 +64,7 @@ import static
org.apache.ignite.internal.util.ByteUtils.bytesToLongKeepingOrder;
import static org.apache.ignite.internal.util.ByteUtils.fromBytes;
import static
org.apache.ignite.internal.util.ByteUtils.longToBytesKeepingOrder;
import static org.apache.ignite.internal.util.ByteUtils.toBytes;
+import static org.apache.ignite.internal.util.ByteUtils.uuidToBytes;
import static
org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
@@ -612,20 +616,23 @@ public class DistributionZoneManager implements
IgniteComponent {
try {
Set<LogicalNode> logicalTopology = newTopology.nodes();
- Condition updateCondition;
-
- if (newTopology.version() == 1) {
- // Very first start of the cluster, so we just initialize
zonesLogicalTopologyVersionKey
- updateCondition = notExists(zonesLogicalTopologyVersionKey());
+ Condition condition;
+ Update update;
+
+ if (newTopology.version() ==
LogicalTopologySnapshot.FIRST_VERSION) {
+ // Very first start of the cluster, OR first topology version
after a cluster reset, so we just
+ // initialize zonesLogicalTopologyVersionKey.
+ // We don't need to check whether clusterId is 'newer' as it's
guaranteed that after a newer clusterId
+ // gets written to the Metastorage, we cannot send a
Metastorage update switching it back to older clusterId.
+ condition = notExists(zonesLogicalTopologyVersionKey())
+
.or(value(zonesLogicalTopologyClusterIdKey()).ne(uuidToBytes(newTopology.clusterId())));
+ update =
updateLogicalTopologyAndVersionAndClusterId(newTopology);
} else {
- updateCondition =
value(zonesLogicalTopologyVersionKey()).lt(longToBytesKeepingOrder(newTopology.version()));
+ condition =
value(zonesLogicalTopologyVersionKey()).lt(longToBytesKeepingOrder(newTopology.version()));
+ update = updateLogicalTopologyAndVersion(newTopology);
}
- Iif iff = iif(
- updateCondition,
- updateLogicalTopologyAndVersion(logicalTopology,
newTopology.version()),
- ops().yield(false)
- );
+ Iif iff = iif(condition, update, ops().yield(false));
metaStorageManager.invoke(iff).whenComplete((res, e) -> {
if (e != null) {
@@ -696,9 +703,9 @@ public class DistributionZoneManager implements
IgniteComponent {
}
try {
- assert evt.entryEvents().size() == 2 :
- "Expected an event with logical topology and
logical topology version entries but was events with keys: "
- + evt.entryEvents().stream().map(entry ->
entry.newEntry() == null ? "null" : entry.newEntry().key())
+ assert evt.entryEvents().size() == 2 ||
evt.entryEvents().size() == 3 :
+ "Expected an event with logical topology, its
version and maybe clusterId entries but was events with keys: "
+ +
evt.entryEvents().stream().map(DistributionZoneManager::entryKeyAsString)
.collect(toList());
byte[] newLogicalTopologyBytes;
@@ -738,6 +745,10 @@ public class DistributionZoneManager implements
IgniteComponent {
};
}
+ private static String entryKeyAsString(EntryEvent entry) {
+ return entry.newEntry() == null ? "null" : new
String(entry.newEntry().key(), UTF_8);
+ }
+
/**
* Reaction on an update of logical topology. In this method {@link
DistributionZoneManager#logicalTopology},
* {@link DistributionZoneManager#nodesAttributes}, {@link
ZoneState#topologyAugmentationMap} are updated.
diff --git
a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZonesUtil.java
b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZonesUtil.java
index e368491600..0a5f336535 100644
---
a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZonesUtil.java
+++
b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZonesUtil.java
@@ -34,10 +34,12 @@ import static
org.apache.ignite.internal.metastorage.dsl.Operations.remove;
import static
org.apache.ignite.internal.util.ByteUtils.bytesToLongKeepingOrder;
import static org.apache.ignite.internal.util.ByteUtils.fromBytes;
import static
org.apache.ignite.internal.util.ByteUtils.longToBytesKeepingOrder;
+import static org.apache.ignite.internal.util.ByteUtils.uuidToBytes;
import com.jayway.jsonpath.Configuration;
import com.jayway.jsonpath.JsonPath;
import com.jayway.jsonpath.Option;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
@@ -50,11 +52,12 @@ import
org.apache.ignite.internal.catalog.commands.StorageProfileParams;
import
org.apache.ignite.internal.catalog.descriptors.CatalogStorageProfileDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
-import org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
+import
org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologySnapshot;
import
org.apache.ignite.internal.distributionzones.DistributionZoneManager.ZoneState;
import org.apache.ignite.internal.lang.ByteArray;
import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.metastorage.dsl.CompoundCondition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
import org.apache.ignite.internal.metastorage.dsl.SimpleCondition;
import org.apache.ignite.internal.metastorage.dsl.Update;
import org.apache.ignite.internal.thread.NamedThreadFactory;
@@ -106,6 +109,9 @@ public class DistributionZonesUtil {
/** Key prefix for zones' logical topology version. */
private static final String DISTRIBUTION_ZONES_LOGICAL_TOPOLOGY_VERSION =
DISTRIBUTION_ZONES_LOGICAL_TOPOLOGY_PREFIX + "version";
+ /** Key prefix for zones' logical topology cluster ID. */
+ private static final String DISTRIBUTION_ZONES_LOGICAL_TOPOLOGY_CLUSTER_ID
= DISTRIBUTION_ZONES_LOGICAL_TOPOLOGY_PREFIX + "clusterId";
+
/** Key prefix that represents {@link
ZoneState#topologyAugmentationMap()}.*/
private static final String
DISTRIBUTION_ZONES_TOPOLOGY_AUGMENTATION_PREFIX =
"distributionZones.topologyAugmentation.";
@@ -127,6 +133,10 @@ public class DistributionZonesUtil {
private static final ByteArray
DISTRIBUTION_ZONES_LOGICAL_TOPOLOGY_VERSION_KEY =
new ByteArray(DISTRIBUTION_ZONES_LOGICAL_TOPOLOGY_VERSION);
+ /** ByteArray representation of {@link
DistributionZonesUtil#DISTRIBUTION_ZONES_LOGICAL_TOPOLOGY_CLUSTER_ID}. */
+ private static final ByteArray
DISTRIBUTION_ZONES_LOGICAL_TOPOLOGY_CLUSTER_ID_KEY =
+ new ByteArray(DISTRIBUTION_ZONES_LOGICAL_TOPOLOGY_CLUSTER_ID);
+
/**
* The initial value of trigger revision in case when it is not
initialized in the meta storage.
* It is possible because invoke to metastorage with the initialisation is
async, and scale up/down propagation could be
@@ -209,6 +219,16 @@ public class DistributionZonesUtil {
return DISTRIBUTION_ZONES_LOGICAL_TOPOLOGY_VERSION_KEY;
}
+ /**
+ * The key needed for processing the events about logical topology changes.
+ * Needed for the defencing against stale updates of logical topology
nodes ({@link #zonesLogicalTopologyVersionKey()}
+ * alone is not enough as version might be reset to 1 when a cluster reset
happens; this key allows to distinguish between
+ * two events about version=1).
+ */
+ public static ByteArray zonesLogicalTopologyClusterIdKey() {
+ return DISTRIBUTION_ZONES_LOGICAL_TOPOLOGY_CLUSTER_ID_KEY;
+ }
+
/**
* The key prefix needed for processing an event about zone's data nodes.
*/
@@ -377,21 +397,42 @@ public class DistributionZonesUtil {
}
/**
- * Updates logical topology and logical topology version values for zones.
+ * Updates logical topology and its version values for zones.
*
- * @param logicalTopology Logical topology.
- * @param topologyVersion Logical topology version.
+ * @param logicalTopology Logical topology snapshot.
* @return Update command for the meta storage.
*/
- static Update updateLogicalTopologyAndVersion(Set<LogicalNode>
logicalTopology, long topologyVersion) {
- Set<NodeWithAttributes> topologyFromCmg = logicalTopology.stream()
+ static Update updateLogicalTopologyAndVersion(LogicalTopologySnapshot
logicalTopology) {
+ return
updateLogicalTopologyAndVersionAndMaybeClusterId(logicalTopology, false);
+ }
+
+ /**
+ * Updates logical topology, its version and cluster ID values for zones.
+ *
+ * @param logicalTopology Logical topology snapshot.
+ * @return Update command for the meta storage.
+ */
+ static Update
updateLogicalTopologyAndVersionAndClusterId(LogicalTopologySnapshot
logicalTopology) {
+ return
updateLogicalTopologyAndVersionAndMaybeClusterId(logicalTopology, true);
+ }
+
+ private static Update updateLogicalTopologyAndVersionAndMaybeClusterId(
+ LogicalTopologySnapshot logicalTopology,
+ boolean updateClusterId
+ ) {
+ Set<NodeWithAttributes> topologyFromCmg =
logicalTopology.nodes().stream()
.map(n -> new NodeWithAttributes(n.name(), n.id(),
n.userAttributes(), n.storageProfiles()))
.collect(toSet());
- return ops(
- put(zonesLogicalTopologyVersionKey(),
longToBytesKeepingOrder(topologyVersion)),
- put(zonesLogicalTopologyKey(),
ByteUtils.toBytes(topologyFromCmg))
- ).yield(true);
+ List<Operation> operations = new ArrayList<>();
+
+ operations.add(put(zonesLogicalTopologyVersionKey(),
longToBytesKeepingOrder(logicalTopology.version())));
+ operations.add(put(zonesLogicalTopologyKey(),
ByteUtils.toBytes(topologyFromCmg)));
+ if (updateClusterId) {
+ operations.add(put(zonesLogicalTopologyClusterIdKey(),
uuidToBytes(logicalTopology.clusterId())));
+ }
+
+ return ops(operations.toArray(Operation[]::new)).yield(true);
}
/**
diff --git
a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/BaseDistributionZoneManagerTest.java
b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/BaseDistributionZoneManagerTest.java
index 417e299a2d..3474c6b365 100644
---
a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/BaseDistributionZoneManagerTest.java
+++
b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/BaseDistributionZoneManagerTest.java
@@ -32,6 +32,7 @@ import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.List;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -55,6 +56,7 @@ import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import
org.apache.ignite.internal.metastorage.impl.StandaloneMetaStorageManager;
import
org.apache.ignite.internal.metastorage.server.SimpleInMemoryKeyValueStorage;
+import org.apache.ignite.internal.network.ConstantClusterIdSupplier;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.apache.ignite.internal.thread.NamedThreadFactory;
import org.jetbrains.annotations.Nullable;
@@ -101,7 +103,7 @@ public abstract class BaseDistributionZoneManagerTest
extends BaseIgniteAbstract
components.add(clusterStateStorage);
- topology = new LogicalTopologyImpl(clusterStateStorage);
+ topology = new LogicalTopologyImpl(clusterStateStorage, new
ConstantClusterIdSupplier(UUID.randomUUID()));
ClusterManagementGroupManager cmgManager =
mock(ClusterManagementGroupManager.class);
diff --git
a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerLogicalTopologyEventsTest.java
b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerLogicalTopologyEventsTest.java
index c977690955..7865568468 100644
---
a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerLogicalTopologyEventsTest.java
+++
b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerLogicalTopologyEventsTest.java
@@ -26,6 +26,7 @@ import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import java.util.Set;
+import java.util.UUID;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
import
org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologySnapshot;
import org.apache.ignite.internal.hlc.HybridTimestamp;
@@ -42,6 +43,8 @@ public class DistributionZoneManagerLogicalTopologyEventsTest
extends BaseDistri
private static final LogicalNode NODE_2 = new LogicalNode("2", "name2",
new NetworkAddress("localhost", 123));
+ private final UUID clusterId = UUID.randomUUID();
+
@Test
void testMetaStorageKeysInitializedOnStartWhenTopVerEmpty() throws
Exception {
assertThat(distributionZoneManager.startAsync(new ComponentContext()),
willCompleteSuccessfully());
@@ -169,7 +172,7 @@ public class
DistributionZoneManagerLogicalTopologyEventsTest extends BaseDistri
var clusterNodes2 = Set.of(NODE_1, NODE_2);
- clusterStateStorage.put(LOGICAL_TOPOLOGY_KEY, ByteUtils.toBytes(new
LogicalTopologySnapshot(10L, clusterNodes2)));
+ clusterStateStorage.put(LOGICAL_TOPOLOGY_KEY, ByteUtils.toBytes(new
LogicalTopologySnapshot(10L, clusterNodes2, clusterId)));
topology.fireTopologyLeap();
@@ -190,7 +193,7 @@ public class
DistributionZoneManagerLogicalTopologyEventsTest extends BaseDistri
var clusterNodes2 = Set.of(NODE_1, NODE_2);
- clusterStateStorage.put(LOGICAL_TOPOLOGY_KEY, ByteUtils.toBytes(new
LogicalTopologySnapshot(10L, clusterNodes2)));
+ clusterStateStorage.put(LOGICAL_TOPOLOGY_KEY, ByteUtils.toBytes(new
LogicalTopologySnapshot(10L, clusterNodes2, clusterId)));
keyValueStorage.put(zonesLogicalTopologyVersionKey().bytes(),
ByteUtils.longToBytesKeepingOrder(11L), HybridTimestamp.MIN_VALUE);
diff --git
a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/causalitydatanodes/DistributionZoneCausalityDataNodesTest.java
b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/causalitydatanodes/DistributionZoneCausalityDataNodesTest.java
index 9aeaf1b6b7..901d5d94b8 100644
---
a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/causalitydatanodes/DistributionZoneCausalityDataNodesTest.java
+++
b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/causalitydatanodes/DistributionZoneCausalityDataNodesTest.java
@@ -420,7 +420,7 @@ public class DistributionZoneCausalityDataNodesTest extends
BaseDistributionZone
int zoneId = getZoneId(ZONE_NAME);
- // Check that data nodes value of the the zone is NODE_0.
+ // Check that data nodes value of the zone is NODE_0.
Set<String> dataNodes1 =
distributionZoneManager.dataNodes(topologyRevision1,
catalogManager.latestCatalogVersion(), zoneId)
.get(TIMEOUT, MILLISECONDS);
assertEquals(ONE_NODE_NAME, dataNodes1);
diff --git
a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageMultipleNodesAbstractTest.java
b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageMultipleNodesAbstractTest.java
index 9c109bbe2c..570dd58626 100644
---
a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageMultipleNodesAbstractTest.java
+++
b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageMultipleNodesAbstractTest.java
@@ -42,6 +42,7 @@ import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -80,6 +81,7 @@ import
org.apache.ignite.internal.metastorage.server.time.ClusterTime;
import org.apache.ignite.internal.metastorage.server.time.ClusterTimeImpl;
import org.apache.ignite.internal.metrics.NoOpMetricManager;
import org.apache.ignite.internal.network.ClusterService;
+import org.apache.ignite.internal.network.ConstantClusterIdSupplier;
import org.apache.ignite.internal.network.StaticNodeFinder;
import org.apache.ignite.internal.network.utils.ClusterServiceTestUtils;
import org.apache.ignite.internal.raft.Loza;
@@ -178,7 +180,7 @@ public abstract class
ItMetaStorageMultipleNodesAbstractTest extends IgniteAbstr
raftGroupEventsClientListener
);
- var logicalTopology = new LogicalTopologyImpl(clusterStateStorage);
+ var logicalTopology = new LogicalTopologyImpl(clusterStateStorage,
new ConstantClusterIdSupplier(UUID.randomUUID()));
var clusterInitializer = new ClusterInitializer(
clusterService,
diff --git
a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageWatchTest.java
b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageWatchTest.java
index 20e887d07c..2eea36e48b 100644
---
a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageWatchTest.java
+++
b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageWatchTest.java
@@ -42,6 +42,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
@@ -81,6 +82,7 @@ import org.apache.ignite.internal.metastorage.dsl.Operations;
import
org.apache.ignite.internal.metastorage.server.persistence.RocksDbKeyValueStorage;
import org.apache.ignite.internal.metrics.NoOpMetricManager;
import org.apache.ignite.internal.network.ClusterService;
+import org.apache.ignite.internal.network.ConstantClusterIdSupplier;
import org.apache.ignite.internal.network.StaticNodeFinder;
import org.apache.ignite.internal.network.utils.ClusterServiceTestUtils;
import org.apache.ignite.internal.raft.Loza;
@@ -164,7 +166,7 @@ public class ItMetaStorageWatchTest extends
IgniteAbstractTest {
components.add(clusterStateStorage);
- var logicalTopology = new LogicalTopologyImpl(clusterStateStorage);
+ var logicalTopology = new LogicalTopologyImpl(clusterStateStorage,
new ConstantClusterIdSupplier(UUID.randomUUID()));
var clusterInitializer = new ClusterInitializer(
clusterService,
diff --git
a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java
b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java
index 7f15fecdb6..050069ba45 100644
---
a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java
+++
b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java
@@ -56,6 +56,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
@@ -132,6 +133,7 @@ import
org.apache.ignite.internal.metastorage.server.KeyValueStorage;
import
org.apache.ignite.internal.metastorage.server.persistence.RocksDbKeyValueStorage;
import org.apache.ignite.internal.metrics.NoOpMetricManager;
import org.apache.ignite.internal.network.ClusterService;
+import org.apache.ignite.internal.network.ConstantClusterIdSupplier;
import org.apache.ignite.internal.network.StaticNodeFinder;
import
org.apache.ignite.internal.network.configuration.NetworkExtensionConfigurationSchema;
import org.apache.ignite.internal.network.recovery.InMemoryStaleIds;
@@ -987,7 +989,7 @@ public class ItReplicaLifecycleTest extends
BaseIgniteAbstractTest {
);
var clusterStateStorage = new TestClusterStateStorage();
- var logicalTopology = new LogicalTopologyImpl(clusterStateStorage);
+ var logicalTopology = new LogicalTopologyImpl(clusterStateStorage,
new ConstantClusterIdSupplier(UUID.randomUUID()));
var clusterInitializer = new ClusterInitializer(
clusterService,
diff --git
a/modules/replicator/src/integrationTest/java/org/apache/ignite/internal/replicator/ItPlacementDriverReplicaSideTest.java
b/modules/replicator/src/integrationTest/java/org/apache/ignite/internal/replicator/ItPlacementDriverReplicaSideTest.java
index 048abe85ca..edf130d85c 100644
---
a/modules/replicator/src/integrationTest/java/org/apache/ignite/internal/replicator/ItPlacementDriverReplicaSideTest.java
+++
b/modules/replicator/src/integrationTest/java/org/apache/ignite/internal/replicator/ItPlacementDriverReplicaSideTest.java
@@ -99,7 +99,7 @@ import
org.apache.ignite.internal.replicator.message.ReplicaRequest;
import
org.apache.ignite.internal.replicator.message.TestReplicaMessagesFactory;
import org.apache.ignite.internal.testframework.IgniteAbstractTest;
import org.apache.ignite.internal.thread.NamedThreadFactory;
-import org.apache.ignite.internal.topology.LogicalTopologyServiceTestImpl;
+import org.apache.ignite.internal.topology.TestLogicalTopologyService;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.apache.ignite.network.ClusterNode;
@@ -202,7 +202,7 @@ public class ItPlacementDriverReplicaSideTest extends
IgniteAbstractTest {
TopologyAwareRaftGroupServiceFactory
topologyAwareRaftGroupServiceFactory = new TopologyAwareRaftGroupServiceFactory(
clusterService,
- new LogicalTopologyServiceTestImpl(clusterService),
+ new TestLogicalTopologyService(clusterService),
Loza.FACTORY,
eventsClientListener
);
diff --git
a/modules/replicator/src/testFixtures/java/org/apache/ignite/internal/raft/client/AbstractTopologyAwareGroupServiceTest.java
b/modules/replicator/src/testFixtures/java/org/apache/ignite/internal/raft/client/AbstractTopologyAwareGroupServiceTest.java
index cc0bbcf0b5..423cdba8fa 100644
---
a/modules/replicator/src/testFixtures/java/org/apache/ignite/internal/raft/client/AbstractTopologyAwareGroupServiceTest.java
+++
b/modules/replicator/src/testFixtures/java/org/apache/ignite/internal/raft/client/AbstractTopologyAwareGroupServiceTest.java
@@ -62,7 +62,7 @@ import
org.apache.ignite.internal.raft.util.ThreadLocalOptimizedMarshaller;
import org.apache.ignite.internal.replicator.TestReplicationGroupId;
import org.apache.ignite.internal.testframework.IgniteAbstractTest;
import org.apache.ignite.internal.thread.NamedThreadFactory;
-import org.apache.ignite.internal.topology.LogicalTopologyServiceTestImpl;
+import org.apache.ignite.internal.topology.TestLogicalTopologyService;
import org.apache.ignite.internal.util.CollectionUtils;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.network.ClusterNode;
@@ -230,7 +230,7 @@ public abstract class AbstractTopologyAwareGroupServiceTest
extends IgniteAbstra
isServerAddress,
nodes,
null,
- new LogicalTopologyServiceTestImpl(clientClusterService),
+ new TestLogicalTopologyService(clientClusterService),
false
);
@@ -425,7 +425,7 @@ public abstract class AbstractTopologyAwareGroupServiceTest
extends IgniteAbstra
for (NetworkAddress addr : addresses) {
ClusterService cluster = clusterServices.get(addr);
- LogicalTopologyService logicalTopologyService = new
LogicalTopologyServiceTestImpl(cluster);
+ LogicalTopologyService logicalTopologyService = new
TestLogicalTopologyService(cluster);
RaftGroupEventsClientListener eventsClientListener = new
RaftGroupEventsClientListener();
diff --git
a/modules/replicator/src/testFixtures/java/org/apache/ignite/internal/topology/LogicalTopologyServiceTestImpl.java
b/modules/replicator/src/testFixtures/java/org/apache/ignite/internal/topology/TestLogicalTopologyService.java
similarity index 86%
rename from
modules/replicator/src/testFixtures/java/org/apache/ignite/internal/topology/LogicalTopologyServiceTestImpl.java
rename to
modules/replicator/src/testFixtures/java/org/apache/ignite/internal/topology/TestLogicalTopologyService.java
index 0a11a61cd2..1919d1eec0 100644
---
a/modules/replicator/src/testFixtures/java/org/apache/ignite/internal/topology/LogicalTopologyServiceTestImpl.java
+++
b/modules/replicator/src/testFixtures/java/org/apache/ignite/internal/topology/TestLogicalTopologyService.java
@@ -21,6 +21,7 @@ import static
java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.stream.Collectors.toSet;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
import
org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyEventListener;
@@ -32,10 +33,12 @@ import org.apache.ignite.network.ClusterNode;
/**
* Test implementation of {@link LogicalTopologyService}.
*/
-public class LogicalTopologyServiceTestImpl implements LogicalTopologyService {
+public class TestLogicalTopologyService implements LogicalTopologyService {
private final ClusterService clusterService;
- public LogicalTopologyServiceTestImpl(ClusterService clusterService) {
+ private final UUID clusterId = UUID.randomUUID();
+
+ public TestLogicalTopologyService(ClusterService clusterService) {
this.clusterService = clusterService;
}
@@ -53,15 +56,18 @@ public class LogicalTopologyServiceTestImpl implements
LogicalTopologyService {
public CompletableFuture<LogicalTopologySnapshot>
logicalTopologyOnLeader() {
return completedFuture(new LogicalTopologySnapshot(
1,
-
clusterService.topologyService().allMembers().stream().map(LogicalNode::new).collect(toSet()))
- );
+
clusterService.topologyService().allMembers().stream().map(LogicalNode::new).collect(toSet()),
+ clusterId
+ ));
}
@Override
public LogicalTopologySnapshot localLogicalTopology() {
return new LogicalTopologySnapshot(
1,
-
clusterService.topologyService().allMembers().stream().map(LogicalNode::new).collect(toSet()));
+
clusterService.topologyService().allMembers().stream().map(LogicalNode::new).collect(toSet()),
+ clusterId
+ );
}
@Override
diff --git
a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/ItDistributedConfigurationPropertiesTest.java
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/ItDistributedConfigurationPropertiesTest.java
index 239574d66e..704762f236 100644
---
a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/ItDistributedConfigurationPropertiesTest.java
+++
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/ItDistributedConfigurationPropertiesTest.java
@@ -34,6 +34,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import java.nio.file.Path;
import java.util.List;
import java.util.Objects;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Stream;
import org.apache.ignite.configuration.ConfigurationValue;
@@ -66,6 +67,7 @@ import
org.apache.ignite.internal.metastorage.impl.MetaStorageManagerImpl;
import
org.apache.ignite.internal.metastorage.server.SimpleInMemoryKeyValueStorage;
import org.apache.ignite.internal.metrics.NoOpMetricManager;
import org.apache.ignite.internal.network.ClusterService;
+import org.apache.ignite.internal.network.ConstantClusterIdSupplier;
import org.apache.ignite.internal.network.StaticNodeFinder;
import org.apache.ignite.internal.network.utils.ClusterServiceTestUtils;
import org.apache.ignite.internal.raft.Loza;
@@ -184,7 +186,7 @@ public class ItDistributedConfigurationPropertiesTest
extends BaseIgniteAbstract
);
var clusterStateStorage = new TestClusterStateStorage();
- var logicalTopology = new LogicalTopologyImpl(clusterStateStorage);
+ var logicalTopology = new LogicalTopologyImpl(clusterStateStorage,
new ConstantClusterIdSupplier(UUID.randomUUID()));
var clusterInitializer = new ClusterInitializer(
clusterService,
diff --git
a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItDistributedConfigurationStorageTest.java
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItDistributedConfigurationStorageTest.java
index f8cfddef2d..76a2d1ebfa 100644
---
a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItDistributedConfigurationStorageTest.java
+++
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItDistributedConfigurationStorageTest.java
@@ -31,6 +31,7 @@ import java.io.Serializable;
import java.nio.file.Path;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.cluster.management.ClusterIdHolder;
import org.apache.ignite.internal.cluster.management.ClusterInitializer;
@@ -58,6 +59,7 @@ import
org.apache.ignite.internal.metastorage.impl.MetaStorageManagerImpl;
import
org.apache.ignite.internal.metastorage.server.SimpleInMemoryKeyValueStorage;
import org.apache.ignite.internal.metrics.NoOpMetricManager;
import org.apache.ignite.internal.network.ClusterService;
+import org.apache.ignite.internal.network.ConstantClusterIdSupplier;
import org.apache.ignite.internal.network.StaticNodeFinder;
import org.apache.ignite.internal.network.utils.ClusterServiceTestUtils;
import org.apache.ignite.internal.raft.Loza;
@@ -157,7 +159,7 @@ public class ItDistributedConfigurationStorageTest extends
BaseIgniteAbstractTes
);
var clusterStateStorage = new TestClusterStateStorage();
- var logicalTopology = new LogicalTopologyImpl(clusterStateStorage);
+ var logicalTopology = new LogicalTopologyImpl(clusterStateStorage,
new ConstantClusterIdSupplier(UUID.randomUUID()));
var clusterInitializer = new ClusterInitializer(
clusterService,
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 86ea473c4c..2c2b4d8a8d 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
@@ -408,7 +408,7 @@ public class ItIgniteNodeRestartTest extends
BaseIgniteRestartTest {
raftGroupEventsClientListener
);
- var logicalTopology = new LogicalTopologyImpl(clusterStateStorage);
+ var logicalTopology = new LogicalTopologyImpl(clusterStateStorage,
clusterIdService);
var clusterInitializer = new ClusterInitializer(
clusterSvc,
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 f6967eac14..0fd3bb2afc 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
@@ -596,7 +596,7 @@ public class IgniteImpl implements Ignite {
RaftGroupOptionsConfigurer cmgRaftConfigurer =
RaftGroupOptionsConfigHelper.configureProperties(cmgLogStorageFactory,
cmgWorkDir.metaPath());
- var logicalTopology = new LogicalTopologyImpl(clusterStateStorage);
+ var logicalTopology = new LogicalTopologyImpl(clusterStateStorage,
clusterIdService);
ConfigurationTreeGenerator distributedConfigurationGenerator = new
ConfigurationTreeGenerator(
modules.distributed().rootKeys(),
diff --git
a/modules/runner/src/testFixtures/java/org/apache/ignite/internal/Cluster.java
b/modules/runner/src/testFixtures/java/org/apache/ignite/internal/Cluster.java
index 15f32cfa58..d3a9c43b03 100644
---
a/modules/runner/src/testFixtures/java/org/apache/ignite/internal/Cluster.java
+++
b/modules/runner/src/testFixtures/java/org/apache/ignite/internal/Cluster.java
@@ -126,6 +126,9 @@ public class Cluster {
/** Number of nodes in the cluster on first startAndInit() [if it was
invoked]. */
private volatile int initialClusterSize;
+ /** Number of seeds to use instead of initialClusterSize. */
+ private volatile int seedCountOverride = -1;
+
/** Indices of nodes that have been knocked out. */
private final Set<Integer> knockedOutNodesIndices = new
ConcurrentHashSet<>();
@@ -310,8 +313,10 @@ public class Cluster {
}
private String seedAddressesString() {
+ int localSeedCountOverride = seedCountOverride;
// We do this maxing because in some scenarios startAndInit() is not
invoked, instead startNode() is used directly.
- int seedsCount = Math.max(Math.max(initialClusterSize, nodes.size()),
1);
+ int seedsCount = localSeedCountOverride > 0 ? localSeedCountOverride
+ : Math.max(Math.max(initialClusterSize, nodes.size()), 1);
return IntStream.range(0, seedsCount)
.map(index -> BASE_PORT + index)
@@ -657,6 +662,15 @@ public class Cluster {
return tablePartitionIds.get(0);
}
+ /**
+ * Sets number of seeds to use instead of initialClusterSize.
+ *
+ * @param newOverride New override value.
+ */
+ public void overrideSeedsCount(int newOverride) {
+ seedCountOverride = newOverride;
+ }
+
private static class AddCensorshipByRecipientConsistentId implements
BiPredicate<String, NetworkMessage> {
private final String recipientName;
@Nullable
diff --git a/modules/system-disaster-recovery/build.gradle
b/modules/system-disaster-recovery/build.gradle
index dfd129e1f2..e4b9ec8d05 100644
--- a/modules/system-disaster-recovery/build.gradle
+++ b/modules/system-disaster-recovery/build.gradle
@@ -41,6 +41,9 @@ dependencies {
integrationTestImplementation project(':ignite-cluster-management')
integrationTestImplementation project(':ignite-vault')
integrationTestImplementation project(':ignite-network-api')
+ integrationTestImplementation project(':ignite-distribution-zones')
+ integrationTestImplementation project(':ignite-catalog')
+ integrationTestImplementation project(':ignite-metastorage-api')
integrationTestImplementation project(':ignite-client')
integrationTestImplementation testFixtures(project(':ignite-core'))
integrationTestImplementation testFixtures(project(':ignite-api'))
diff --git
a/modules/system-disaster-recovery/src/integrationTest/java/org/apache/ignite/internal/disaster/system/ItCmgDisasterRecoveryTest.java
b/modules/system-disaster-recovery/src/integrationTest/java/org/apache/ignite/internal/disaster/system/ItCmgDisasterRecoveryTest.java
index ab312bb929..b653539761 100644
---
a/modules/system-disaster-recovery/src/integrationTest/java/org/apache/ignite/internal/disaster/system/ItCmgDisasterRecoveryTest.java
+++
b/modules/system-disaster-recovery/src/integrationTest/java/org/apache/ignite/internal/disaster/system/ItCmgDisasterRecoveryTest.java
@@ -32,6 +32,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.List;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -83,6 +84,10 @@ class ItCmgDisasterRecoveryTest extends
ClusterPerTestIntegrationTest {
}
private void startAndInitCluster(int nodeCount, int[] cmgNodeIndexes,
int[] metastorageNodeIndexes) {
+ // Pre-allocate this to make sure that for each pair of nodes, if they
start almost at the same time, at least one is able to make
+ // an initial sync to another one.
+ cluster.overrideSeedsCount(10);
+
cluster.startAndInit(nodeCount, paramsBuilder -> {
paramsBuilder.cmgNodeNames(nodeNames(cmgNodeIndexes));
paramsBuilder.metaStorageNodeNames(nodeNames(metastorageNodeIndexes));
@@ -372,4 +377,76 @@ class ItCmgDisasterRecoveryTest extends
ClusterPerTestIntegrationTest {
.addresses(igniteImpl(0).clientAddress().host())
.build();
}
+
+ @Test
+ void dataNodesAreUpdatedCorrectlyAfterClusterReset() throws Exception {
+ startAndInitCluster(2, new int[]{0}, new int[]{1});
+ waitTillClusterStateIsSavedToVaultOnConductor(1);
+
+ final String zoneName = "TEST_ZONE";
+
+ cluster.node(1).sql().execute(
+ null,
+ "CREATE ZONE " + zoneName + " WITH STORAGE_PROFILES='default',
"
+ + "DATA_NODES_AUTO_ADJUST_SCALE_UP=0,
DATA_NODES_AUTO_ADJUST_SCALE_DOWN=0"
+ );
+
+ int zoneId = igniteImpl(1).catalogManager().zone("TEST_ZONE",
Long.MAX_VALUE).id();
+
+ waitTillDataNodesBecome(new int[]{0, 1}, zoneId, igniteImpl(1));
+
+ cluster.startNode(2);
+
+ waitTillDataNodesBecome(new int[]{0, 1, 2}, zoneId, igniteImpl(1));
+
+ // This makes the CMG majority go away.
+ cluster.stopNode(0);
+
+ // Now, dataNodes should have become [1, 2], but as there is no CMG
leader, no one is able to trigger data nodes update.
+
+ // Repair CMG with just node 1.
+ initiateCmgRepairVia(igniteImpl(1), 1);
+ IgniteImpl restartedIgniteImpl1 = waitTillNodeRestartsInternally(1);
+ waitTillCmgHasMajority(restartedIgniteImpl1);
+
+ waitTillDataNodesBecome(new int[]{1, 2}, zoneId, restartedIgniteImpl1);
+
+ // Starting the node that did not see the repair.
+ migrate(0, 1);
+
+ waitTillDataNodesBecome(new int[]{0, 1, 2}, zoneId,
restartedIgniteImpl1);
+
+ // Now let's make sure that normal additions/removals still work after
cluster reset.
+
+ cluster.stopNode(0);
+ waitTillDataNodesBecome(new int[]{1, 2}, zoneId, restartedIgniteImpl1);
+
+ cluster.startNode(3);
+ waitTillDataNodesBecome(new int[]{1, 2, 3}, zoneId,
restartedIgniteImpl1);
+
+ // TODO: IGNITE-23096 - remove after the hang is fixed.
+ waitTillNodesRestartInProcess(1, 2);
+ }
+
+ private void waitTillDataNodesBecome(int[] expectedDataNodeIndexes, int
zoneId, IgniteImpl ignite) throws InterruptedException {
+ int catalogVersion = ignite.catalogManager().latestCatalogVersion();
+
+ assertTrue(
+ waitForCondition(
+ () -> currentDataNodes(ignite, catalogVersion,
zoneId).equals(Set.of(nodeNames(expectedDataNodeIndexes))),
+ SECONDS.toMillis(10)
+ ),
+ "Did not see data nodes to become " +
IntStream.of(expectedDataNodeIndexes).boxed().collect(toList()) + " in time"
+ );
+ }
+
+ private static Set<String> currentDataNodes(IgniteImpl ignite, int
catalogVersion, int zoneId) {
+ long currentRevision = ignite.metaStorageManager().appliedRevision();
+
+ CompletableFuture<Set<String>> dataNodesFuture =
ignite.distributionZoneManager()
+ .dataNodes(currentRevision, catalogVersion, zoneId);
+
+ assertThat(dataNodesFuture, willCompleteSuccessfully());
+ return dataNodesFuture.join();
+ }
}
diff --git
a/modules/table/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceDistributedTest.java
b/modules/table/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceDistributedTest.java
index 5f547dac2d..8f4daf77f2 100644
---
a/modules/table/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceDistributedTest.java
+++
b/modules/table/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceDistributedTest.java
@@ -1175,7 +1175,7 @@ public class ItRebalanceDistributedTest extends
BaseIgniteAbstractTest {
));
var clusterStateStorage = new TestClusterStateStorage();
- var logicalTopology = new LogicalTopologyImpl(clusterStateStorage);
+ var logicalTopology = new LogicalTopologyImpl(clusterStateStorage,
clusterIdService);
var clusterInitializer = new ClusterInitializer(
clusterService,
diff --git
a/modules/table/src/testFixtures/java/org/apache/ignite/distributed/ItTxTestCluster.java
b/modules/table/src/testFixtures/java/org/apache/ignite/distributed/ItTxTestCluster.java
index ce61ad1b57..73028b0a03 100644
---
a/modules/table/src/testFixtures/java/org/apache/ignite/distributed/ItTxTestCluster.java
+++
b/modules/table/src/testFixtures/java/org/apache/ignite/distributed/ItTxTestCluster.java
@@ -48,6 +48,7 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
@@ -855,6 +856,8 @@ public class ItTxTestCluster {
private LogicalTopologyService logicalTopologyService(ClusterService
clusterService) {
return new LogicalTopologyService() {
+ private final UUID clusterId = UUID.randomUUID();
+
@Override
public void addEventListener(LogicalTopologyEventListener
listener) {
@@ -869,14 +872,18 @@ public class ItTxTestCluster {
public CompletableFuture<LogicalTopologySnapshot>
logicalTopologyOnLeader() {
return completedFuture(new LogicalTopologySnapshot(
1,
-
clusterService.topologyService().allMembers().stream().map(LogicalNode::new).collect(toSet())));
+
clusterService.topologyService().allMembers().stream().map(LogicalNode::new).collect(toSet()),
+ clusterId
+ ));
}
@Override
public LogicalTopologySnapshot localLogicalTopology() {
return new LogicalTopologySnapshot(
1,
-
clusterService.topologyService().allMembers().stream().map(LogicalNode::new).collect(toSet()));
+
clusterService.topologyService().allMembers().stream().map(LogicalNode::new).collect(toSet()),
+ clusterId
+ );
}
@Override
