sergeyuttsel commented on code in PR #1729: URL: https://github.com/apache/ignite-3/pull/1729#discussion_r1163051005
########## modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneAwaitDataNodesTest.java: ########## @@ -0,0 +1,910 @@ +/* + * 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.distributionzones; + +import static java.util.Collections.emptySet; +import static java.util.concurrent.CompletableFuture.completedFuture; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.apache.ignite.configuration.annotation.ConfigurationType.DISTRIBUTED; +import static org.apache.ignite.internal.distributionzones.DistributionZoneManager.DEFAULT_ZONE_ID; +import static org.apache.ignite.internal.distributionzones.DistributionZoneManager.DEFAULT_ZONE_NAME; +import static org.apache.ignite.internal.distributionzones.DistributionZoneManager.IMMEDIATE_TIMER_VALUE; +import static org.apache.ignite.internal.distributionzones.DistributionZoneManager.INFINITE_TIMER_VALUE; +import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.toDataNodesMap; +import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zoneDataNodesKey; +import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zoneLogicalTopologyPrefix; +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.zonesDataNodesPrefix; +import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesLogicalTopologyKey; +import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesLogicalTopologyVersionKey; +import static org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrowsWithCause; +import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition; +import static org.apache.ignite.internal.util.ByteUtils.longToBytes; +import static org.apache.ignite.internal.util.ByteUtils.toBytes; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.Arrays; +import java.util.List; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import org.apache.ignite.configuration.NamedConfigurationTree; +import org.apache.ignite.configuration.NamedListView; +import org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager; +import org.apache.ignite.internal.cluster.management.raft.ClusterStateStorage; +import org.apache.ignite.internal.cluster.management.raft.TestClusterStateStorage; +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.LogicalTopologyServiceImpl; +import org.apache.ignite.internal.configuration.ConfigurationManager; +import org.apache.ignite.internal.configuration.storage.TestConfigurationStorage; +import org.apache.ignite.internal.distributionzones.DistributionZoneManager.ZoneState; +import org.apache.ignite.internal.distributionzones.configuration.DistributionZoneConfigurationSchema; +import org.apache.ignite.internal.distributionzones.configuration.DistributionZonesConfiguration; +import org.apache.ignite.internal.distributionzones.exception.DistributionZoneWasRemovedException; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.metastorage.Entry; +import org.apache.ignite.internal.metastorage.EntryEvent; +import org.apache.ignite.internal.metastorage.MetaStorageManager; +import org.apache.ignite.internal.metastorage.WatchEvent; +import org.apache.ignite.internal.metastorage.WatchListener; +import org.apache.ignite.internal.metastorage.dsl.StatementResultImpl; +import org.apache.ignite.internal.metastorage.impl.EntryImpl; +import org.apache.ignite.internal.schema.configuration.TableChange; +import org.apache.ignite.internal.schema.configuration.TableConfiguration; +import org.apache.ignite.internal.schema.configuration.TableView; +import org.apache.ignite.internal.schema.configuration.TablesConfiguration; +import org.apache.ignite.internal.testframework.IgniteAbstractTest; +import org.apache.ignite.internal.vault.VaultEntry; +import org.apache.ignite.internal.vault.VaultManager; +import org.apache.ignite.lang.ByteArray; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * Tests awaiting data nodes algorithm in distribution zone manager in case when + * {@link DistributionZoneConfigurationSchema#dataNodesAutoAdjustScaleUp} + * or {@link DistributionZoneConfigurationSchema#dataNodesAutoAdjustScaleDown} are immediate. + */ +public class DistributionZoneAwaitDataNodesTest extends IgniteAbstractTest { + private static final IgniteLogger LOG = Loggers.forClass(DistributionZoneAwaitDataNodesTest.class); + + private MetaStorageManager metaStorageManager; + + private DistributionZoneManager distributionZoneManager; + + private LogicalTopology logicalTopology; + + private ClusterStateStorage clusterStateStorage; + + private DistributionZonesConfiguration zonesConfiguration; + + private ConfigurationManager clusterCfgMgr; + + private ClusterManagementGroupManager cmgManager; + + private VaultManager vaultManager; + + private WatchListener topologyWatchListener; + + private WatchListener dataNodesWatchListener; + + @BeforeEach + void setUp() { + vaultManager = mock(VaultManager.class); + + when(vaultManager.get(zonesLogicalTopologyKey())).thenReturn(completedFuture(new VaultEntry(zonesLogicalTopologyKey(), null))); + + when(vaultManager.get(zonesLogicalTopologyVersionKey())) + .thenReturn(completedFuture(new VaultEntry(zonesLogicalTopologyVersionKey(), longToBytes(0)))); + + cmgManager = mock(ClusterManagementGroupManager.class); + + metaStorageManager = mock(MetaStorageManager.class); + + doAnswer(invocation -> { + ByteArray key = invocation.getArgument(0); + + WatchListener watchListener = invocation.getArgument(1); + + if (Arrays.equals(key.bytes(), zoneLogicalTopologyPrefix().bytes())) { + topologyWatchListener = watchListener; + } else if (Arrays.equals(key.bytes(), zonesDataNodesPrefix().bytes())) { + dataNodesWatchListener = watchListener; + } + + return null; + }).when(metaStorageManager).registerPrefixWatch(any(), any()); + + when(metaStorageManager.invoke(any())) + .thenReturn(completedFuture(StatementResultImpl.builder().result(new byte[] {0}).build())); + + clusterStateStorage = new TestClusterStateStorage(); + + logicalTopology = new LogicalTopologyImpl(clusterStateStorage); + + LogicalTopologyServiceImpl logicalTopologyService = new LogicalTopologyServiceImpl(logicalTopology, cmgManager); + + TablesConfiguration tablesConfiguration = mock(TablesConfiguration.class); + + NamedConfigurationTree<TableConfiguration, TableView, TableChange> tables = mock(NamedConfigurationTree.class); + + when(tablesConfiguration.tables()).thenReturn(tables); + + NamedListView<TableView> tablesValue = mock(NamedListView.class); + + when(tables.value()).thenReturn(tablesValue); + + when(tablesValue.size()).thenReturn(0); + + clusterCfgMgr = new ConfigurationManager( + List.of(DistributionZonesConfiguration.KEY), + Set.of(), + new TestConfigurationStorage(DISTRIBUTED), + List.of(), + List.of() + ); + + clusterCfgMgr.start(); + + zonesConfiguration = clusterCfgMgr.configurationRegistry() + .getConfiguration(DistributionZonesConfiguration.KEY); + + distributionZoneManager = new DistributionZoneManager( + zonesConfiguration, + tablesConfiguration, + metaStorageManager, + logicalTopologyService, + vaultManager, + "node" + ); + + mockCmgLocalNodes(); + } + + @AfterEach + void tearDown() throws Exception { + distributionZoneManager.stop(); + clusterCfgMgr.stop(); + clusterStateStorage.stop(); + } + + /** + * Test checks that data nodes futures are completed on topology with added and removed nodes. + */ + @Test + void testSeveralScaleUpAndSeveralScaleDownThenScaleUpAndScaleDown() throws Exception { + startZoneManager(0); + + TestSeveralScaleUpAndSeveralScaleDownDataObject testData = testSeveralScaleUpAndSeveralScaleDownGeneral(); + + LOG.info("Topology with added and removed nodes."); + + Set<String> dataNodes = Set.of("node0", "node2"); + + topologyWatchListenerOnUpdate(dataNodes, testData.topVer2, testData.dataNodesRevision2); + + assertTrue(testData.topVerUpFut2.isDone()); + assertTrue(testData.topVerDownFut2.isDone()); + + CompletableFuture<Void> revisionUpFut = distributionZoneManager.zonesState().get(DEFAULT_ZONE_ID) + .scaleUpRevisionTracker().waitFor(testData.dataNodesRevision2); + + assertFalse(revisionUpFut.isDone()); + + dataNodesWatchListenerOnUpdate(DEFAULT_ZONE_ID, dataNodes, true, testData.dataNodesRevision2, + testData.dataNodesRevision2 + 1); + + assertTrue(waitForCondition(revisionUpFut::isDone, 3000)); + + CompletableFuture<Void> revisionDownFut = distributionZoneManager.zonesState().get(DEFAULT_ZONE_ID).scaleDownRevisionTracker() + .waitFor(testData.dataNodesRevision2); + + assertFalse(revisionDownFut.isDone()); + + assertFalse(testData.dataNodesUpFut3.isDone()); + assertFalse(testData.dataNodesDownFut3.isDone()); + + dataNodesWatchListenerOnUpdate(DEFAULT_ZONE_ID, dataNodes, false, testData.dataNodesRevision2, + testData.dataNodesRevision2 + 2); + + assertTrue(waitForCondition(revisionDownFut::isDone, 3000)); + + assertEquals(dataNodes, testData.dataNodesUpFut3.get(3, SECONDS)); + assertEquals(dataNodes, testData.dataNodesDownFut3.get(3, SECONDS)); + } + + /** + * Test checks that data nodes futures are completed on topology with added nodes. + */ + @Test + void testSeveralScaleUpAndSeveralScaleDownThenScaleUp() throws Exception { + startZoneManager(0); + + TestSeveralScaleUpAndSeveralScaleDownDataObject testData = testSeveralScaleUpAndSeveralScaleDownGeneral(); + + LOG.info("Topology with added nodes."); + + Set<String> dataNodes = Set.of("node0", "node1", "node2"); + + topologyWatchListenerOnUpdate(dataNodes, testData.topVer2, testData.dataNodesRevision2); + + assertTrue(testData.topVerUpFut2.isDone()); + assertTrue(testData.topVerDownFut2.isDone()); + + CompletableFuture<Void> revisionUpFut = distributionZoneManager.zonesState().get(DEFAULT_ZONE_ID) + .scaleUpRevisionTracker().waitFor(testData.dataNodesRevision2); + + assertFalse(revisionUpFut.isDone()); + + dataNodesWatchListenerOnUpdate(DEFAULT_ZONE_ID, dataNodes, true, testData.dataNodesRevision2, + testData.dataNodesRevision2 + 1); + + assertTrue(waitForCondition(revisionUpFut::isDone, 3000)); + assertEquals(dataNodes, testData.dataNodesUpFut3.get(3, SECONDS)); + assertEquals(dataNodes, testData.dataNodesDownFut3.get(3, SECONDS)); + } + + /** + * Test checks that data nodes futures are completed on topology with removed nodes. + */ + @Test + void testSeveralScaleUpAndSeveralScaleDownThenScaleDown() throws Exception { + startZoneManager(0); + + TestSeveralScaleUpAndSeveralScaleDownDataObject testData = testSeveralScaleUpAndSeveralScaleDownGeneral(); + + LOG.info("Topology with removed nodes."); + + Set<String> dataNodes = Set.of("node0"); + + topologyWatchListenerOnUpdate(dataNodes, testData.topVer2, testData.dataNodesRevision2); + + assertTrue(testData.topVerUpFut2.isDone()); + assertTrue(testData.topVerDownFut2.isDone()); + + CompletableFuture<Void> revisionDownFut = distributionZoneManager.zonesState().get(DEFAULT_ZONE_ID) + .scaleDownRevisionTracker().waitFor((long) testData.dataNodesRevision2); + + assertFalse(revisionDownFut.isDone()); + + dataNodesWatchListenerOnUpdate(DEFAULT_ZONE_ID, dataNodes, false, testData.dataNodesRevision2, + testData.dataNodesRevision2 + 1); + + assertTrue(waitForCondition(revisionDownFut::isDone, 3000)); + assertEquals(dataNodes, testData.dataNodesUpFut3.get(3, SECONDS)); + assertEquals(dataNodes, testData.dataNodesDownFut3.get(3, SECONDS)); + } + + private static class TestSeveralScaleUpAndSeveralScaleDownDataObject { + private final long topVer2; + private final long dataNodesRevision2; + private final CompletableFuture<Void> topVerUpFut2; + private final CompletableFuture<Void> topVerDownFut2; + private final CompletableFuture<Set<String>> dataNodesUpFut3; + private final CompletableFuture<Set<String>> dataNodesDownFut3; + + TestSeveralScaleUpAndSeveralScaleDownDataObject( + long topVer2, + long dataNodesRevision2, + CompletableFuture<Void> topVerUpFut2, + CompletableFuture<Void> topVerDownFut2, + CompletableFuture<Set<String>> dataNodesUpFut3, + CompletableFuture<Set<String>> dataNodesDownFut3) { + this.topVer2 = topVer2; + this.dataNodesRevision2 = dataNodesRevision2; + this.topVerUpFut2 = topVerUpFut2; + this.topVerDownFut2 = topVerDownFut2; + this.dataNodesUpFut3 = dataNodesUpFut3; + this.dataNodesDownFut3 = dataNodesDownFut3; + } + } + + /** + * This method invokes {@link DistributionZoneManager#topologyVersionedDataNodes(int, long)} with default and non-default zone id + * and different logical topology version. Collects data nodes futures. + * Simulates new logical topology with new nodes. Check that some of data nodes futures are completed. + * Simulates new logical topology with removed nodes. Check that some of data nodes futures are completed. + * + * @return Structure with data for continue testing. + */ + private TestSeveralScaleUpAndSeveralScaleDownDataObject testSeveralScaleUpAndSeveralScaleDownGeneral() Review Comment: I've removed TestSeveralScaleUpAndSeveralScaleDownDataObject -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
