You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "sergeyuttsel (via GitHub)" <gi...@apache.org> on 2023/06/09 13:23:55 UTC
[GitHub] [ignite-3] sergeyuttsel commented on a diff in pull request #2172: IGNITE-19574 IGNITE-19574 restore global states of DistributionZoneManager after restart
sergeyuttsel commented on code in PR #2172:
URL: https://github.com/apache/ignite-3/pull/2172#discussion_r1224198380
##########
modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZonesUtil.java:
##########
@@ -73,6 +73,12 @@ public class DistributionZonesUtil {
/** Key prefix for zones' logical topology nodes and logical topology version. */
private static final String DISTRIBUTION_ZONES_LOGICAL_TOPOLOGY_PREFIX = "distributionZones.logicalTopology.";
+ /** Key value for zones' nodes' attributes in vault. */
+ private static final String DISTRIBUTION_ZONES_NODES_ATTRIBUTES_VAULT = "vault.distributionZones.nodesAttributes";
+
+ /** Key prefix for zones' logical topology nodes in vault. */
+ private static final String DISTRIBUTION_ZONES_LOGICAL_TOPOLOGY_VAULT = "vault.distributionZones.logicalTopology.nodes";
Review Comment:
Maybe it will be better to use `"vault." + DISTRIBUTION_ZONES_LOGICAL_TOPOLOGY`
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/distribution/zones/ItIgniteDistributionZoneManagerNodeRestartTest.java:
##########
@@ -0,0 +1,313 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.distribution.zones;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.internal.distributionzones.DistributionZoneManager.IMMEDIATE_TIMER_VALUE;
+import static org.apache.ignite.internal.recovery.ConfigurationCatchUpListener.CONFIGURATION_CATCH_UP_DIFFERENCE_PROPERTY;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.testNodeName;
+import static org.apache.ignite.utils.ClusterServiceTestUtils.defaultSerializationRegistry;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.BaseIgniteRestartTest;
+import org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
+import org.apache.ignite.internal.cluster.management.raft.TestClusterStateStorage;
+import org.apache.ignite.internal.cluster.management.topology.LogicalTopologyImpl;
+import org.apache.ignite.internal.cluster.management.topology.LogicalTopologyServiceImpl;
+import org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
+import org.apache.ignite.internal.configuration.ConfigurationManager;
+import org.apache.ignite.internal.configuration.ConfigurationModules;
+import org.apache.ignite.internal.configuration.ConfigurationTreeGenerator;
+import org.apache.ignite.internal.configuration.NodeConfigWriteException;
+import org.apache.ignite.internal.configuration.storage.DistributedConfigurationStorage;
+import org.apache.ignite.internal.configuration.storage.LocalFileConfigurationStorage;
+import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
+import org.apache.ignite.internal.configuration.validation.ConfigurationValidatorImpl;
+import org.apache.ignite.internal.distributionzones.DistributionZoneConfigurationParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.distributionzones.NodeWithAttributes;
+import org.apache.ignite.internal.distributionzones.configuration.DistributionZonesConfiguration;
+import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.metastorage.impl.StandaloneMetaStorageManager;
+import org.apache.ignite.internal.metastorage.server.TestRocksDbKeyValueStorage;
+import org.apache.ignite.internal.network.configuration.NetworkConfiguration;
+import org.apache.ignite.internal.network.recovery.VaultStateIds;
+import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
+import org.apache.ignite.internal.testframework.TestIgnitionManager;
+import org.apache.ignite.internal.testframework.WithSystemProperty;
+import org.apache.ignite.internal.vault.VaultManager;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.NettyBootstrapFactory;
+import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Tests for checking {@link DistributionZoneManager} behavior after node's restart.
+ */
+@WithSystemProperty(key = CONFIGURATION_CATCH_UP_DIFFERENCE_PROPERTY, value = "0")
+@ExtendWith(ConfigurationExtension.class)
+public class ItIgniteDistributionZoneManagerNodeRestartTest extends BaseIgniteRestartTest {
+ private static final LogicalNode A = new LogicalNode(
+ new ClusterNode("1", "A", new NetworkAddress("localhost", 123)),
+ Map.of("region", "US", "storage", "SSD", "dataRegionSize", "10")
+ );
+
+ private static final LogicalNode B = new LogicalNode(
+ new ClusterNode("2", "B", new NetworkAddress("localhost", 123)),
+ Map.of("region", "EU", "storage", "HHD", "dataRegionSize", "30")
+ );
+
+ private static final LogicalNode C = new LogicalNode(
+ new ClusterNode("3", "C", new NetworkAddress("localhost", 123)),
+ Map.of("region", "CN", "storage", "SSD", "dataRegionSize", "20")
+ );
+
+ private static final String ZONE_NAME = "zone1";
+
+ /**
+ * Start some of Ignite components that are able to serve as Ignite node for test purposes.
+ *
+ * @param idx Node index.
+ * @return Partial node.
+ */
+ private PartialNode startPartialNode(int idx) {
+ String name = testNodeName(testInfo, idx);
+
+ Path dir = workDir.resolve(name);
+
+ List<IgniteComponent> components = new ArrayList<>();
+
+ VaultManager vault = createVault(name, dir);
+
+ ConfigurationModules modules = loadConfigurationModules(log, Thread.currentThread().getContextClassLoader());
+
+ Path configFile = workDir.resolve(TestIgnitionManager.DEFAULT_CONFIG_NAME);
+ String configString = configurationString(idx);
+ try {
+ Files.writeString(configFile, configString);
+ } catch (IOException e) {
+ throw new NodeConfigWriteException("Failed to write config content to file.", e);
+ }
+
+ var localConfigurationGenerator = new ConfigurationTreeGenerator(
+ modules.local().rootKeys(),
+ modules.local().internalSchemaExtensions(),
+ modules.local().polymorphicSchemaExtensions()
+ );
+
+ var nodeCfgMgr = new ConfigurationManager(
+ modules.local().rootKeys(),
+ new LocalFileConfigurationStorage(configFile, localConfigurationGenerator),
+ localConfigurationGenerator,
+ ConfigurationValidatorImpl.withDefaultValidators(localConfigurationGenerator, modules.local().validators())
+ );
+
+ NetworkConfiguration networkConfiguration = nodeCfgMgr.configurationRegistry().getConfiguration(NetworkConfiguration.KEY);
+
+ var nettyBootstrapFactory = new NettyBootstrapFactory(networkConfiguration, name);
+
+ var clusterSvc = new TestScaleCubeClusterServiceFactory().createClusterService(
+ name,
+ networkConfiguration,
+ nettyBootstrapFactory,
+ defaultSerializationRegistry(),
+ new VaultStateIds(vault)
+ );
+
+ var clusterStateStorage = new TestClusterStateStorage();
+
+ var logicalTopology = new LogicalTopologyImpl(clusterStateStorage);
+
+ var cmgManager = mock(ClusterManagementGroupManager.class);
+
+ when(cmgManager.logicalTopology()).thenAnswer(invocation -> completedFuture(logicalTopology.getLogicalTopology()));
+
+ var metaStorageMgr = StandaloneMetaStorageManager.create(
+ vault,
+ new TestRocksDbKeyValueStorage("test", workDir.resolve("metastorage"))
+ );
+
+ var cfgStorage = new DistributedConfigurationStorage(metaStorageMgr, vault);
+
+ ConfigurationTreeGenerator distributedConfigurationGenerator = new ConfigurationTreeGenerator(
+ modules.distributed().rootKeys(),
+ modules.distributed().internalSchemaExtensions(),
+ modules.distributed().polymorphicSchemaExtensions()
+ );
+
+ var clusterCfgMgr = new ConfigurationManager(
+ modules.distributed().rootKeys(),
+ cfgStorage,
+ distributedConfigurationGenerator,
+ ConfigurationValidatorImpl.withDefaultValidators(distributedConfigurationGenerator, modules.distributed().validators())
+ );
+
+ DistributionZonesConfiguration zonesConfiguration = clusterCfgMgr.configurationRegistry()
+ .getConfiguration(DistributionZonesConfiguration.KEY);
+
+ TablesConfiguration tablesConfiguration = clusterCfgMgr.configurationRegistry().getConfiguration(TablesConfiguration.KEY);
+
+ LogicalTopologyServiceImpl logicalTopologyService = new LogicalTopologyServiceImpl(logicalTopology, cmgManager);
+
+ DistributionZoneManager distributionZoneManager = new DistributionZoneManager(
+ zonesConfiguration,
+ tablesConfiguration,
+ metaStorageMgr,
+ logicalTopologyService,
+ vault,
+ name
+ );
+
+ // Preparing the result map.
+
+ components.add(vault);
+ components.add(nodeCfgMgr);
+
+ // Start.
+
+ vault.start();
+ vault.putName(name).join();
+
+ nodeCfgMgr.start();
+
+ // Start the remaining components.
+ List<IgniteComponent> otherComponents = List.of(
+ nettyBootstrapFactory,
+ clusterSvc,
+ clusterStateStorage,
+ cmgManager,
+ metaStorageMgr,
+ clusterCfgMgr,
+ distributionZoneManager
+ );
+
+ for (IgniteComponent component : otherComponents) {
+ component.start();
+
+ components.add(component);
+ }
+
+ PartialNode partialNode = partialNode(
+ nodeCfgMgr,
+ clusterCfgMgr,
+ null,
+ components,
+ localConfigurationGenerator,
+ logicalTopology,
+ cfgStorage,
+ distributedConfigurationGenerator
+ );
+
+ partialNodes.add(partialNode);
+
+ return partialNode;
+ }
+
+ @Test
+ public void testNodeAttributesRestoredAfterRestart() throws Exception {
+ PartialNode partialNode = startPartialNode(0);
+
+ DistributionZoneManager distributionZoneManager = findComponent(partialNode.startedComponents(), DistributionZoneManager.class);
+
+ partialNode.logicalTopology().putNode(A);
+ partialNode.logicalTopology().putNode(B);
+ partialNode.logicalTopology().putNode(C);
+
+ distributionZoneManager.createZone(
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME)
+ .dataNodesAutoAdjustScaleUp(IMMEDIATE_TIMER_VALUE)
+ .dataNodesAutoAdjustScaleDown(IMMEDIATE_TIMER_VALUE)
+ .build()
+ ).get(10_000, TimeUnit.MILLISECONDS);
+
+ Set<String> nodes = distributionZoneManager.topologyVersionedDataNodes(
+ 1,
+ partialNode.logicalTopology().getLogicalTopology().version()
+ ).get(10_000, TimeUnit.MILLISECONDS);
+
+ assertEquals(Set.of(A, B, C).stream().map(ClusterNode::name).collect(Collectors.toSet()), nodes);
+
+ Map<String, Map<String, String>> nodeAttributesBeforeRestart = distributionZoneManager.nodesAttributes();
+
+ partialNode.stop();
+
+ partialNode = startPartialNode(0);
+
+ distributionZoneManager = findComponent(partialNode.startedComponents(), DistributionZoneManager.class);
+
+ Map<String, Map<String, String>> nodeAttributesAfterRestart = distributionZoneManager.nodesAttributes();
+
+ assertEquals(3, nodeAttributesAfterRestart.size());
+
+ assertEquals(nodeAttributesBeforeRestart, nodeAttributesAfterRestart);
+ }
+
+ @Test
+ public void testLogicalTopologyRestoredAfterRestart() throws Exception {
+ PartialNode partialNode = startPartialNode(0);
+
+ DistributionZoneManager distributionZoneManager = findComponent(partialNode.startedComponents(), DistributionZoneManager.class);
+
+ partialNode.logicalTopology().putNode(A);
+ partialNode.logicalTopology().putNode(B);
+ partialNode.logicalTopology().putNode(C);
+
+ distributionZoneManager.createZone(
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME)
+ .dataNodesAutoAdjustScaleUp(IMMEDIATE_TIMER_VALUE)
+ .dataNodesAutoAdjustScaleDown(IMMEDIATE_TIMER_VALUE)
+ .build()
+ ).get(10_000, TimeUnit.MILLISECONDS);
+
+ Set<String> nodes = distributionZoneManager.topologyVersionedDataNodes(
+ 1,
+ partialNode.logicalTopology().getLogicalTopology().version()
+ ).get(10_000, TimeUnit.MILLISECONDS);
+
+ assertEquals(Set.of(A, B, C).stream().map(ClusterNode::name).collect(Collectors.toSet()), nodes);
Review Comment:
`Set.of(A, B, C).stream().map(ClusterNode::name).collect(Collectors.toSet())`
Need to save it in the variable.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/distribution/zones/ItIgniteDistributionZoneManagerNodeRestartTest.java:
##########
@@ -0,0 +1,313 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.distribution.zones;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.internal.distributionzones.DistributionZoneManager.IMMEDIATE_TIMER_VALUE;
+import static org.apache.ignite.internal.recovery.ConfigurationCatchUpListener.CONFIGURATION_CATCH_UP_DIFFERENCE_PROPERTY;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.testNodeName;
+import static org.apache.ignite.utils.ClusterServiceTestUtils.defaultSerializationRegistry;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.BaseIgniteRestartTest;
+import org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
+import org.apache.ignite.internal.cluster.management.raft.TestClusterStateStorage;
+import org.apache.ignite.internal.cluster.management.topology.LogicalTopologyImpl;
+import org.apache.ignite.internal.cluster.management.topology.LogicalTopologyServiceImpl;
+import org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
+import org.apache.ignite.internal.configuration.ConfigurationManager;
+import org.apache.ignite.internal.configuration.ConfigurationModules;
+import org.apache.ignite.internal.configuration.ConfigurationTreeGenerator;
+import org.apache.ignite.internal.configuration.NodeConfigWriteException;
+import org.apache.ignite.internal.configuration.storage.DistributedConfigurationStorage;
+import org.apache.ignite.internal.configuration.storage.LocalFileConfigurationStorage;
+import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
+import org.apache.ignite.internal.configuration.validation.ConfigurationValidatorImpl;
+import org.apache.ignite.internal.distributionzones.DistributionZoneConfigurationParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.distributionzones.NodeWithAttributes;
+import org.apache.ignite.internal.distributionzones.configuration.DistributionZonesConfiguration;
+import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.metastorage.impl.StandaloneMetaStorageManager;
+import org.apache.ignite.internal.metastorage.server.TestRocksDbKeyValueStorage;
+import org.apache.ignite.internal.network.configuration.NetworkConfiguration;
+import org.apache.ignite.internal.network.recovery.VaultStateIds;
+import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
+import org.apache.ignite.internal.testframework.TestIgnitionManager;
+import org.apache.ignite.internal.testframework.WithSystemProperty;
+import org.apache.ignite.internal.vault.VaultManager;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.NettyBootstrapFactory;
+import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Tests for checking {@link DistributionZoneManager} behavior after node's restart.
+ */
+@WithSystemProperty(key = CONFIGURATION_CATCH_UP_DIFFERENCE_PROPERTY, value = "0")
+@ExtendWith(ConfigurationExtension.class)
+public class ItIgniteDistributionZoneManagerNodeRestartTest extends BaseIgniteRestartTest {
+ private static final LogicalNode A = new LogicalNode(
+ new ClusterNode("1", "A", new NetworkAddress("localhost", 123)),
+ Map.of("region", "US", "storage", "SSD", "dataRegionSize", "10")
+ );
+
+ private static final LogicalNode B = new LogicalNode(
+ new ClusterNode("2", "B", new NetworkAddress("localhost", 123)),
+ Map.of("region", "EU", "storage", "HHD", "dataRegionSize", "30")
+ );
+
+ private static final LogicalNode C = new LogicalNode(
+ new ClusterNode("3", "C", new NetworkAddress("localhost", 123)),
+ Map.of("region", "CN", "storage", "SSD", "dataRegionSize", "20")
+ );
+
+ private static final String ZONE_NAME = "zone1";
+
+ /**
+ * Start some of Ignite components that are able to serve as Ignite node for test purposes.
+ *
+ * @param idx Node index.
+ * @return Partial node.
+ */
+ private PartialNode startPartialNode(int idx) {
+ String name = testNodeName(testInfo, idx);
+
+ Path dir = workDir.resolve(name);
+
+ List<IgniteComponent> components = new ArrayList<>();
+
+ VaultManager vault = createVault(name, dir);
+
+ ConfigurationModules modules = loadConfigurationModules(log, Thread.currentThread().getContextClassLoader());
+
+ Path configFile = workDir.resolve(TestIgnitionManager.DEFAULT_CONFIG_NAME);
+ String configString = configurationString(idx);
+ try {
+ Files.writeString(configFile, configString);
+ } catch (IOException e) {
+ throw new NodeConfigWriteException("Failed to write config content to file.", e);
+ }
+
+ var localConfigurationGenerator = new ConfigurationTreeGenerator(
+ modules.local().rootKeys(),
+ modules.local().internalSchemaExtensions(),
+ modules.local().polymorphicSchemaExtensions()
+ );
+
+ var nodeCfgMgr = new ConfigurationManager(
+ modules.local().rootKeys(),
+ new LocalFileConfigurationStorage(configFile, localConfigurationGenerator),
+ localConfigurationGenerator,
+ ConfigurationValidatorImpl.withDefaultValidators(localConfigurationGenerator, modules.local().validators())
+ );
+
+ NetworkConfiguration networkConfiguration = nodeCfgMgr.configurationRegistry().getConfiguration(NetworkConfiguration.KEY);
+
+ var nettyBootstrapFactory = new NettyBootstrapFactory(networkConfiguration, name);
+
+ var clusterSvc = new TestScaleCubeClusterServiceFactory().createClusterService(
+ name,
+ networkConfiguration,
+ nettyBootstrapFactory,
+ defaultSerializationRegistry(),
+ new VaultStateIds(vault)
+ );
+
+ var clusterStateStorage = new TestClusterStateStorage();
+
+ var logicalTopology = new LogicalTopologyImpl(clusterStateStorage);
+
+ var cmgManager = mock(ClusterManagementGroupManager.class);
+
+ when(cmgManager.logicalTopology()).thenAnswer(invocation -> completedFuture(logicalTopology.getLogicalTopology()));
+
+ var metaStorageMgr = StandaloneMetaStorageManager.create(
+ vault,
+ new TestRocksDbKeyValueStorage("test", workDir.resolve("metastorage"))
+ );
+
+ var cfgStorage = new DistributedConfigurationStorage(metaStorageMgr, vault);
+
+ ConfigurationTreeGenerator distributedConfigurationGenerator = new ConfigurationTreeGenerator(
+ modules.distributed().rootKeys(),
+ modules.distributed().internalSchemaExtensions(),
+ modules.distributed().polymorphicSchemaExtensions()
+ );
+
+ var clusterCfgMgr = new ConfigurationManager(
+ modules.distributed().rootKeys(),
+ cfgStorage,
+ distributedConfigurationGenerator,
+ ConfigurationValidatorImpl.withDefaultValidators(distributedConfigurationGenerator, modules.distributed().validators())
+ );
+
+ DistributionZonesConfiguration zonesConfiguration = clusterCfgMgr.configurationRegistry()
+ .getConfiguration(DistributionZonesConfiguration.KEY);
+
+ TablesConfiguration tablesConfiguration = clusterCfgMgr.configurationRegistry().getConfiguration(TablesConfiguration.KEY);
+
+ LogicalTopologyServiceImpl logicalTopologyService = new LogicalTopologyServiceImpl(logicalTopology, cmgManager);
+
+ DistributionZoneManager distributionZoneManager = new DistributionZoneManager(
+ zonesConfiguration,
+ tablesConfiguration,
+ metaStorageMgr,
+ logicalTopologyService,
+ vault,
+ name
+ );
+
+ // Preparing the result map.
+
+ components.add(vault);
+ components.add(nodeCfgMgr);
+
+ // Start.
+
+ vault.start();
+ vault.putName(name).join();
+
+ nodeCfgMgr.start();
+
+ // Start the remaining components.
+ List<IgniteComponent> otherComponents = List.of(
+ nettyBootstrapFactory,
+ clusterSvc,
+ clusterStateStorage,
+ cmgManager,
+ metaStorageMgr,
+ clusterCfgMgr,
+ distributionZoneManager
+ );
+
+ for (IgniteComponent component : otherComponents) {
+ component.start();
+
+ components.add(component);
+ }
+
+ PartialNode partialNode = partialNode(
+ nodeCfgMgr,
+ clusterCfgMgr,
+ null,
+ components,
+ localConfigurationGenerator,
+ logicalTopology,
+ cfgStorage,
+ distributedConfigurationGenerator
+ );
+
+ partialNodes.add(partialNode);
+
+ return partialNode;
+ }
+
+ @Test
+ public void testNodeAttributesRestoredAfterRestart() throws Exception {
+ PartialNode partialNode = startPartialNode(0);
+
+ DistributionZoneManager distributionZoneManager = findComponent(partialNode.startedComponents(), DistributionZoneManager.class);
+
+ partialNode.logicalTopology().putNode(A);
+ partialNode.logicalTopology().putNode(B);
+ partialNode.logicalTopology().putNode(C);
+
+ distributionZoneManager.createZone(
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME)
+ .dataNodesAutoAdjustScaleUp(IMMEDIATE_TIMER_VALUE)
+ .dataNodesAutoAdjustScaleDown(IMMEDIATE_TIMER_VALUE)
+ .build()
+ ).get(10_000, TimeUnit.MILLISECONDS);
+
+ Set<String> nodes = distributionZoneManager.topologyVersionedDataNodes(
+ 1,
+ partialNode.logicalTopology().getLogicalTopology().version()
+ ).get(10_000, TimeUnit.MILLISECONDS);
+
+ assertEquals(Set.of(A, B, C).stream().map(ClusterNode::name).collect(Collectors.toSet()), nodes);
+
+ Map<String, Map<String, String>> nodeAttributesBeforeRestart = distributionZoneManager.nodesAttributes();
+
+ partialNode.stop();
+
+ partialNode = startPartialNode(0);
+
+ distributionZoneManager = findComponent(partialNode.startedComponents(), DistributionZoneManager.class);
+
+ Map<String, Map<String, String>> nodeAttributesAfterRestart = distributionZoneManager.nodesAttributes();
+
+ assertEquals(3, nodeAttributesAfterRestart.size());
+
+ assertEquals(nodeAttributesBeforeRestart, nodeAttributesAfterRestart);
+ }
+
+ @Test
+ public void testLogicalTopologyRestoredAfterRestart() throws Exception {
Review Comment:
Do we need two separate tests? They have many of the same lines of code. Perhaps need to move them to a separate method.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/distribution/zones/ItIgniteDistributionZoneManagerNodeRestartTest.java:
##########
@@ -0,0 +1,313 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.distribution.zones;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.internal.distributionzones.DistributionZoneManager.IMMEDIATE_TIMER_VALUE;
+import static org.apache.ignite.internal.recovery.ConfigurationCatchUpListener.CONFIGURATION_CATCH_UP_DIFFERENCE_PROPERTY;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.testNodeName;
+import static org.apache.ignite.utils.ClusterServiceTestUtils.defaultSerializationRegistry;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.BaseIgniteRestartTest;
+import org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
+import org.apache.ignite.internal.cluster.management.raft.TestClusterStateStorage;
+import org.apache.ignite.internal.cluster.management.topology.LogicalTopologyImpl;
+import org.apache.ignite.internal.cluster.management.topology.LogicalTopologyServiceImpl;
+import org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
+import org.apache.ignite.internal.configuration.ConfigurationManager;
+import org.apache.ignite.internal.configuration.ConfigurationModules;
+import org.apache.ignite.internal.configuration.ConfigurationTreeGenerator;
+import org.apache.ignite.internal.configuration.NodeConfigWriteException;
+import org.apache.ignite.internal.configuration.storage.DistributedConfigurationStorage;
+import org.apache.ignite.internal.configuration.storage.LocalFileConfigurationStorage;
+import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
+import org.apache.ignite.internal.configuration.validation.ConfigurationValidatorImpl;
+import org.apache.ignite.internal.distributionzones.DistributionZoneConfigurationParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.distributionzones.NodeWithAttributes;
+import org.apache.ignite.internal.distributionzones.configuration.DistributionZonesConfiguration;
+import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.metastorage.impl.StandaloneMetaStorageManager;
+import org.apache.ignite.internal.metastorage.server.TestRocksDbKeyValueStorage;
+import org.apache.ignite.internal.network.configuration.NetworkConfiguration;
+import org.apache.ignite.internal.network.recovery.VaultStateIds;
+import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
+import org.apache.ignite.internal.testframework.TestIgnitionManager;
+import org.apache.ignite.internal.testframework.WithSystemProperty;
+import org.apache.ignite.internal.vault.VaultManager;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.NettyBootstrapFactory;
+import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Tests for checking {@link DistributionZoneManager} behavior after node's restart.
+ */
+@WithSystemProperty(key = CONFIGURATION_CATCH_UP_DIFFERENCE_PROPERTY, value = "0")
+@ExtendWith(ConfigurationExtension.class)
+public class ItIgniteDistributionZoneManagerNodeRestartTest extends BaseIgniteRestartTest {
+ private static final LogicalNode A = new LogicalNode(
+ new ClusterNode("1", "A", new NetworkAddress("localhost", 123)),
+ Map.of("region", "US", "storage", "SSD", "dataRegionSize", "10")
+ );
+
+ private static final LogicalNode B = new LogicalNode(
+ new ClusterNode("2", "B", new NetworkAddress("localhost", 123)),
+ Map.of("region", "EU", "storage", "HHD", "dataRegionSize", "30")
+ );
+
+ private static final LogicalNode C = new LogicalNode(
+ new ClusterNode("3", "C", new NetworkAddress("localhost", 123)),
+ Map.of("region", "CN", "storage", "SSD", "dataRegionSize", "20")
+ );
+
+ private static final String ZONE_NAME = "zone1";
+
+ /**
+ * Start some of Ignite components that are able to serve as Ignite node for test purposes.
+ *
+ * @param idx Node index.
+ * @return Partial node.
+ */
+ private PartialNode startPartialNode(int idx) {
+ String name = testNodeName(testInfo, idx);
+
+ Path dir = workDir.resolve(name);
+
+ List<IgniteComponent> components = new ArrayList<>();
+
+ VaultManager vault = createVault(name, dir);
+
+ ConfigurationModules modules = loadConfigurationModules(log, Thread.currentThread().getContextClassLoader());
+
+ Path configFile = workDir.resolve(TestIgnitionManager.DEFAULT_CONFIG_NAME);
+ String configString = configurationString(idx);
+ try {
+ Files.writeString(configFile, configString);
+ } catch (IOException e) {
+ throw new NodeConfigWriteException("Failed to write config content to file.", e);
+ }
+
+ var localConfigurationGenerator = new ConfigurationTreeGenerator(
+ modules.local().rootKeys(),
+ modules.local().internalSchemaExtensions(),
+ modules.local().polymorphicSchemaExtensions()
+ );
+
+ var nodeCfgMgr = new ConfigurationManager(
+ modules.local().rootKeys(),
+ new LocalFileConfigurationStorage(configFile, localConfigurationGenerator),
+ localConfigurationGenerator,
+ ConfigurationValidatorImpl.withDefaultValidators(localConfigurationGenerator, modules.local().validators())
+ );
+
+ NetworkConfiguration networkConfiguration = nodeCfgMgr.configurationRegistry().getConfiguration(NetworkConfiguration.KEY);
+
+ var nettyBootstrapFactory = new NettyBootstrapFactory(networkConfiguration, name);
+
+ var clusterSvc = new TestScaleCubeClusterServiceFactory().createClusterService(
+ name,
+ networkConfiguration,
+ nettyBootstrapFactory,
+ defaultSerializationRegistry(),
+ new VaultStateIds(vault)
+ );
+
+ var clusterStateStorage = new TestClusterStateStorage();
+
+ var logicalTopology = new LogicalTopologyImpl(clusterStateStorage);
+
+ var cmgManager = mock(ClusterManagementGroupManager.class);
+
+ when(cmgManager.logicalTopology()).thenAnswer(invocation -> completedFuture(logicalTopology.getLogicalTopology()));
+
+ var metaStorageMgr = StandaloneMetaStorageManager.create(
+ vault,
+ new TestRocksDbKeyValueStorage("test", workDir.resolve("metastorage"))
+ );
+
+ var cfgStorage = new DistributedConfigurationStorage(metaStorageMgr, vault);
+
+ ConfigurationTreeGenerator distributedConfigurationGenerator = new ConfigurationTreeGenerator(
+ modules.distributed().rootKeys(),
+ modules.distributed().internalSchemaExtensions(),
+ modules.distributed().polymorphicSchemaExtensions()
+ );
+
+ var clusterCfgMgr = new ConfigurationManager(
+ modules.distributed().rootKeys(),
+ cfgStorage,
+ distributedConfigurationGenerator,
+ ConfigurationValidatorImpl.withDefaultValidators(distributedConfigurationGenerator, modules.distributed().validators())
+ );
+
+ DistributionZonesConfiguration zonesConfiguration = clusterCfgMgr.configurationRegistry()
+ .getConfiguration(DistributionZonesConfiguration.KEY);
+
+ TablesConfiguration tablesConfiguration = clusterCfgMgr.configurationRegistry().getConfiguration(TablesConfiguration.KEY);
+
+ LogicalTopologyServiceImpl logicalTopologyService = new LogicalTopologyServiceImpl(logicalTopology, cmgManager);
+
+ DistributionZoneManager distributionZoneManager = new DistributionZoneManager(
+ zonesConfiguration,
+ tablesConfiguration,
+ metaStorageMgr,
+ logicalTopologyService,
+ vault,
+ name
+ );
+
+ // Preparing the result map.
+
+ components.add(vault);
+ components.add(nodeCfgMgr);
+
+ // Start.
+
+ vault.start();
+ vault.putName(name).join();
+
+ nodeCfgMgr.start();
+
+ // Start the remaining components.
+ List<IgniteComponent> otherComponents = List.of(
+ nettyBootstrapFactory,
+ clusterSvc,
+ clusterStateStorage,
+ cmgManager,
+ metaStorageMgr,
+ clusterCfgMgr,
+ distributionZoneManager
+ );
+
+ for (IgniteComponent component : otherComponents) {
+ component.start();
+
+ components.add(component);
+ }
+
+ PartialNode partialNode = partialNode(
+ nodeCfgMgr,
+ clusterCfgMgr,
+ null,
+ components,
+ localConfigurationGenerator,
+ logicalTopology,
+ cfgStorage,
+ distributedConfigurationGenerator
+ );
+
+ partialNodes.add(partialNode);
+
+ return partialNode;
+ }
+
+ @Test
+ public void testNodeAttributesRestoredAfterRestart() throws Exception {
+ PartialNode partialNode = startPartialNode(0);
+
+ DistributionZoneManager distributionZoneManager = findComponent(partialNode.startedComponents(), DistributionZoneManager.class);
+
+ partialNode.logicalTopology().putNode(A);
+ partialNode.logicalTopology().putNode(B);
+ partialNode.logicalTopology().putNode(C);
+
+ distributionZoneManager.createZone(
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME)
+ .dataNodesAutoAdjustScaleUp(IMMEDIATE_TIMER_VALUE)
+ .dataNodesAutoAdjustScaleDown(IMMEDIATE_TIMER_VALUE)
+ .build()
+ ).get(10_000, TimeUnit.MILLISECONDS);
+
+ Set<String> nodes = distributionZoneManager.topologyVersionedDataNodes(
+ 1,
+ partialNode.logicalTopology().getLogicalTopology().version()
+ ).get(10_000, TimeUnit.MILLISECONDS);
+
+ assertEquals(Set.of(A, B, C).stream().map(ClusterNode::name).collect(Collectors.toSet()), nodes);
+
+ Map<String, Map<String, String>> nodeAttributesBeforeRestart = distributionZoneManager.nodesAttributes();
+
+ partialNode.stop();
+
+ partialNode = startPartialNode(0);
+
+ distributionZoneManager = findComponent(partialNode.startedComponents(), DistributionZoneManager.class);
+
+ Map<String, Map<String, String>> nodeAttributesAfterRestart = distributionZoneManager.nodesAttributes();
+
+ assertEquals(3, nodeAttributesAfterRestart.size());
+
+ assertEquals(nodeAttributesBeforeRestart, nodeAttributesAfterRestart);
+ }
+
+ @Test
+ public void testLogicalTopologyRestoredAfterRestart() throws Exception {
+ PartialNode partialNode = startPartialNode(0);
+
+ DistributionZoneManager distributionZoneManager = findComponent(partialNode.startedComponents(), DistributionZoneManager.class);
+
+ partialNode.logicalTopology().putNode(A);
+ partialNode.logicalTopology().putNode(B);
+ partialNode.logicalTopology().putNode(C);
+
+ distributionZoneManager.createZone(
Review Comment:
Why do you need to create a zone in this test?
##########
modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java:
##########
@@ -1340,6 +1349,26 @@ public void onError(Throwable e) {
};
}
+ /**
+ * Saves logical topology and nodes' attributes map to vault atomically in one batch.
+ * After restart it could be used to restore these fields.
+ *
+ * @param newLogicalTopology Logical topology.
+ */
+ private void saveLogicalTopologyNodeAttributesToVault(Set<NodeWithAttributes> newLogicalTopology) {
Review Comment:
It's not only save to vault but also update local state of `logicalTopology` and `nodesAttributes`. Maybe need to rename it or update these fields outside from this method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org