You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sa...@apache.org on 2023/06/30 06:50:36 UTC
[ignite-3] branch main updated: IGNITE-19857 Default zone replica count is not listened by rebalance trigger (#2264)
This is an automated email from the ASF dual-hosted git repository.
sanpwc 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 1c968b377b IGNITE-19857 Default zone replica count is not listened by rebalance trigger (#2264)
1c968b377b is described below
commit 1c968b377bfecd87c86f27b8a23d87007c2c6afc
Author: Roman Puchkovskiy <ro...@gmail.com>
AuthorDate: Fri Jun 30 10:50:31 2023 +0400
IGNITE-19857 Default zone replica count is not listened by rebalance trigger (#2264)
---
.../rebalance/DistributionZoneRebalanceEngine.java | 9 +-
.../DistributionZoneRebalanceEngineTest.java | 139 +++++++++++++++------
2 files changed, 110 insertions(+), 38 deletions(-)
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngine.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngine.java
index c87cdd7c83..d2fed35132 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngine.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngine.java
@@ -35,6 +35,7 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.ignite.configuration.notifications.ConfigurationListener;
import org.apache.ignite.configuration.notifications.ConfigurationNotificationEvent;
import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
import org.apache.ignite.internal.distributionzones.Node;
@@ -81,6 +82,8 @@ public class DistributionZoneRebalanceEngine {
/** Meta storage listener for data nodes changes. */
private final WatchListener dataNodesListener;
+ private final ConfigurationListener<Integer> onUpdateReplicas = this::onUpdateReplicas;
+
/**
* The constructor.
*
@@ -118,7 +121,8 @@ public class DistributionZoneRebalanceEngine {
}
try {
- zonesConfiguration.distributionZones().any().replicas().listen(this::onUpdateReplicas);
+ zonesConfiguration.defaultDistributionZone().replicas().listen(onUpdateReplicas);
+ zonesConfiguration.distributionZones().any().replicas().listen(onUpdateReplicas);
// TODO: IGNITE-18694 - Recovery for the case when zones watch listener processed event but assignments were not updated.
metaStorageManager.registerPrefixWatch(zoneDataNodesKey(), dataNodesListener);
@@ -138,6 +142,9 @@ public class DistributionZoneRebalanceEngine {
busyLock.block();
metaStorageManager.unregisterWatch(dataNodesListener);
+
+ zonesConfiguration.defaultDistributionZone().replicas().stopListen(onUpdateReplicas);
+ zonesConfiguration.distributionZones().any().replicas().stopListen(onUpdateReplicas);
}
private WatchListener createDistributionZonesDataNodesListener() {
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngineTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngineTest.java
index bc3a7d973c..8b128c09aa 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngineTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngineTest.java
@@ -17,19 +17,25 @@
package org.apache.ignite.internal.distributionzones.rebalance;
+import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Collections.emptySet;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.affinity.AffinityUtils.calculateAssignmentForPartition;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.getZoneById;
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.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.util.ByteUtils.fromBytes;
import static org.apache.ignite.internal.util.ByteUtils.toBytes;
+import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
@@ -56,6 +62,7 @@ import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
import org.apache.ignite.internal.distributionzones.Node;
import org.apache.ignite.internal.distributionzones.configuration.DistributionZoneConfiguration;
import org.apache.ignite.internal.distributionzones.configuration.DistributionZonesConfiguration;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.metastorage.EntryEvent;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
@@ -66,6 +73,7 @@ import org.apache.ignite.internal.metastorage.command.MetaStorageWriteCommand;
import org.apache.ignite.internal.metastorage.command.MultiInvokeCommand;
import org.apache.ignite.internal.metastorage.dsl.Iif;
import org.apache.ignite.internal.metastorage.impl.EntryImpl;
+import org.apache.ignite.internal.metastorage.impl.StandaloneMetaStorageManager;
import org.apache.ignite.internal.metastorage.server.SimpleInMemoryKeyValueStorage;
import org.apache.ignite.internal.metastorage.server.raft.MetaStorageListener;
import org.apache.ignite.internal.metastorage.server.time.ClusterTimeImpl;
@@ -100,14 +108,14 @@ import org.mockito.quality.Strictness;
public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
private SimpleInMemoryKeyValueStorage keyValueStorage;
- @Mock()
+ @Mock
private ClusterService clusterService;
- private MetaStorageManager metaStorageManager = mock(MetaStorageManager.class);
+ private final MetaStorageManager metaStorageManager = mock(MetaStorageManager.class);
- private DistributionZoneManager distributionZoneManager = mock(DistributionZoneManager.class);
+ private final DistributionZoneManager distributionZoneManager = mock(DistributionZoneManager.class);
- private VaultManager vaultManager = mock(VaultManager.class);
+ private final VaultManager vaultManager = mock(VaultManager.class);
private DistributionZoneRebalanceEngine rebalanceEngine;
@@ -212,7 +220,7 @@ public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
}
@AfterEach
- public void tearDown() throws Exception {
+ public void tearDown() {
keyValueStorage.close();
rebalanceEngine.stop();
}
@@ -231,14 +239,7 @@ public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
) {
assignTableIds(tablesConfiguration);
- rebalanceEngine = new DistributionZoneRebalanceEngine(
- new AtomicBoolean(),
- new IgniteSpinBusyLock(),
- distributionZonesConfiguration,
- tablesConfiguration,
- metaStorageManager,
- distributionZoneManager
- );
+ createRebalanceEngine(tablesConfiguration);
rebalanceEngine.start();
@@ -273,14 +274,7 @@ public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
) {
assignTableIds(tablesConfiguration);
- rebalanceEngine = new DistributionZoneRebalanceEngine(
- new AtomicBoolean(),
- new IgniteSpinBusyLock(),
- distributionZonesConfiguration,
- tablesConfiguration,
- metaStorageManager,
- distributionZoneManager
- );
+ createRebalanceEngine(tablesConfiguration);
rebalanceEngine.start();
@@ -314,14 +308,7 @@ public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
) {
assignTableIds(tablesConfiguration);
- rebalanceEngine = new DistributionZoneRebalanceEngine(
- new AtomicBoolean(),
- new IgniteSpinBusyLock(),
- distributionZonesConfiguration,
- tablesConfiguration,
- metaStorageManager,
- distributionZoneManager
- );
+ createRebalanceEngine(tablesConfiguration);
rebalanceEngine.start();
@@ -357,14 +344,7 @@ public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
) {
assignTableIds(tablesConfiguration);
- rebalanceEngine = new DistributionZoneRebalanceEngine(
- new AtomicBoolean(),
- new IgniteSpinBusyLock(),
- distributionZonesConfiguration,
- tablesConfiguration,
- metaStorageManager,
- distributionZoneManager
- );
+ createRebalanceEngine(tablesConfiguration);
rebalanceEngine.start();
@@ -393,6 +373,91 @@ public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
verify(keyValueStorage, timeout(1000).times(2)).invoke(any(), any());
}
+ @Test
+ void replicasTriggersAssignmentsChangingOnNonDefaultZones(
+ @InjectConfiguration
+ ("mock.tables {"
+ + "table0 = { zoneId = 1, id = 1 }}")
+ TablesConfiguration tablesConfiguration
+ ) throws Exception {
+ when(distributionZoneManager.dataNodes(anyInt())).thenReturn(Set.of("node0"));
+
+ keyValueStorage.put(stablePartAssignmentsKey(new TablePartitionId(1, 0)).bytes(), toBytes(Set.of("node0")), someTimestamp());
+
+ MetaStorageManager realMetaStorageManager = StandaloneMetaStorageManager.create(vaultManager, keyValueStorage);
+
+ realMetaStorageManager.start();
+
+ try {
+ createRebalanceEngine(tablesConfiguration, realMetaStorageManager);
+
+ rebalanceEngine.start();
+
+ CompletableFuture<Void> changeFuture = distributionZonesConfiguration.change(zonesChange -> zonesChange.changeDistributionZones(
+ zoneListChange -> zoneListChange.update("zone0", zoneChange -> zoneChange.changeReplicas(2))
+ ));
+ assertThat(changeFuture, willCompleteSuccessfully());
+
+ assertTrue(waitForCondition(() -> keyValueStorage.get("assignments.pending.1_part_0".getBytes(UTF_8)) != null, 10_000));
+ } finally {
+ realMetaStorageManager.stop();
+ }
+ }
+
+ private static HybridTimestamp someTimestamp() {
+ return new HybridTimestamp(System.currentTimeMillis(), 0);
+ }
+
+ @Test
+ void replicasTriggersAssignmentsChangingOnDefaultZone(
+ @InjectConfiguration
+ ("mock.tables {"
+ + "table0 = { zoneId = 0, id = 1 }}")
+ TablesConfiguration tablesConfiguration
+ ) throws Exception {
+ when(distributionZoneManager.dataNodes(anyInt())).thenReturn(Set.of("node0"));
+
+ for (int i = 0; i < 25; i++) {
+ keyValueStorage.put(stablePartAssignmentsKey(new TablePartitionId(1, i)).bytes(), toBytes(Set.of("node0")), someTimestamp());
+ }
+
+ MetaStorageManager realMetaStorageManager = StandaloneMetaStorageManager.create(vaultManager, keyValueStorage);
+
+ realMetaStorageManager.start();
+
+ try {
+ createRebalanceEngine(tablesConfiguration, realMetaStorageManager);
+
+ rebalanceEngine.start();
+
+ CompletableFuture<Void> changeFuture = distributionZonesConfiguration.change(zonesChange ->
+ zonesChange.changeDefaultDistributionZone(zoneChange -> {
+ zoneChange.changeReplicas(2);
+ })
+ );
+ assertThat(changeFuture, willCompleteSuccessfully());
+
+ assertTrue(waitForCondition(() -> keyValueStorage.get("assignments.pending.1_part_0".getBytes(UTF_8)) != null, 10_000));
+ } finally {
+ realMetaStorageManager.stop();
+ }
+ }
+
+ private void createRebalanceEngine(TablesConfiguration tablesConfiguration) {
+ createRebalanceEngine(tablesConfiguration, metaStorageManager);
+ }
+
+ private void createRebalanceEngine(TablesConfiguration tablesConfiguration, MetaStorageManager metaStorageManager) {
+ rebalanceEngine = new DistributionZoneRebalanceEngine(
+ new AtomicBoolean(),
+ new IgniteSpinBusyLock(),
+ distributionZonesConfiguration,
+ tablesConfiguration,
+ metaStorageManager,
+ distributionZoneManager
+ );
+ }
+
private void checkAssignments(
TablesConfiguration tablesConfiguration,
Map<Integer, Set<String>> zoneNodes,