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/02/08 16:55:59 UTC
[ignite-3] branch main updated: IGNITE-18132 Scale down scheduler added (#1631)
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 397abf2d0c IGNITE-18132 Scale down scheduler added (#1631)
397abf2d0c is described below
commit 397abf2d0c85fd442077a19b0bfcb3f4fdf3494e
Author: Mirza Aliev <al...@gmail.com>
AuthorDate: Wed Feb 8 20:55:53 2023 +0400
IGNITE-18132 Scale down scheduler added (#1631)
---
.../distributionzones/DistributionZoneManager.java | 378 ++++--
.../distributionzones/DistributionZonesUtil.java | 48 +-
.../DistributionZoneConfigurationSchema.java | 7 +-
...ibutionZoneManagerConfigurationChangesTest.java | 54 +-
...butionZoneManagerLogicalTopologyEventsTest.java | 38 +-
.../DistributionZoneManagerScaleUpTest.java | 1341 ++++++++++++++++----
.../DistributionZoneManagerTest.java | 37 +-
.../DistributionZoneManagerWatchListenerTest.java | 390 ++----
.../util/DistributionZonesTestUtil.java | 313 +++++
.../internal/table/distributed/TableManager.java | 3 +-
.../TableManagerDistributionZonesTest.java | 3 +-
11 files changed, 1876 insertions(+), 736 deletions(-)
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 e592a34c55..e7a1d86789 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
@@ -22,8 +22,10 @@ import static java.util.concurrent.CompletableFuture.failedFuture;
import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toSet;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.deleteDataNodesAndUpdateTriggerKeys;
+import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.toDataNodesMap;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.triggerKeyConditionForZonesChanges;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.triggerScaleUpScaleDownKeysCondition;
+import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.updateDataNodesAndScaleDownTriggerKey;
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;
@@ -45,8 +47,10 @@ import static org.apache.ignite.lang.ErrorGroups.Common.NODE_STOPPING_ERR;
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashSet;
+import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
@@ -63,6 +67,7 @@ import java.util.function.Function;
import org.apache.ignite.configuration.ConfigurationChangeException;
import org.apache.ignite.configuration.NamedConfigurationTree;
import org.apache.ignite.configuration.NamedListChange;
+import org.apache.ignite.configuration.notifications.ConfigurationListener;
import org.apache.ignite.configuration.notifications.ConfigurationNamedListListener;
import org.apache.ignite.configuration.notifications.ConfigurationNotificationEvent;
import org.apache.ignite.configuration.validation.ConfigurationValidationException;
@@ -118,6 +123,9 @@ public class DistributionZoneManager implements IgniteComponent {
/** Id of the default distribution zone. */
public static final int DEFAULT_ZONE_ID = 0;
+ /** Default infinite value for the distribution zones' timers. */
+ public static final int INFINITE_TIMER_VALUE = Integer.MAX_VALUE;
+
/** The logger. */
private static final IgniteLogger LOG = Loggers.forClass(DistributionZoneManager.class);
@@ -252,20 +260,20 @@ public class DistributionZoneManager implements IgniteComponent {
try {
zonesListChange.create(distributionZoneCfg.name(), zoneChange -> {
if (distributionZoneCfg.dataNodesAutoAdjust() == null) {
- zoneChange.changeDataNodesAutoAdjust(Integer.MAX_VALUE);
+ zoneChange.changeDataNodesAutoAdjust(INFINITE_TIMER_VALUE);
} else {
zoneChange.changeDataNodesAutoAdjust(distributionZoneCfg.dataNodesAutoAdjust());
}
if (distributionZoneCfg.dataNodesAutoAdjustScaleUp() == null) {
- zoneChange.changeDataNodesAutoAdjustScaleUp(Integer.MAX_VALUE);
+ zoneChange.changeDataNodesAutoAdjustScaleUp(INFINITE_TIMER_VALUE);
} else {
zoneChange.changeDataNodesAutoAdjustScaleUp(
distributionZoneCfg.dataNodesAutoAdjustScaleUp());
}
if (distributionZoneCfg.dataNodesAutoAdjustScaleDown() == null) {
- zoneChange.changeDataNodesAutoAdjustScaleDown(Integer.MAX_VALUE);
+ zoneChange.changeDataNodesAutoAdjustScaleDown(INFINITE_TIMER_VALUE);
} else {
zoneChange.changeDataNodesAutoAdjustScaleDown(distributionZoneCfg.dataNodesAutoAdjustScaleDown());
}
@@ -483,8 +491,12 @@ public class DistributionZoneManager implements IgniteComponent {
ZonesConfigurationListener zonesConfigurationListener = new ZonesConfigurationListener();
zonesConfiguration.distributionZones().listenElements(zonesConfigurationListener);
+ zonesConfiguration.distributionZones().any().dataNodesAutoAdjustScaleUp().listen(onUpdateScaleUp());
+ zonesConfiguration.distributionZones().any().dataNodesAutoAdjustScaleDown().listen(onUpdateScaleDown());
zonesConfiguration.defaultDistributionZone().listen(zonesConfigurationListener);
+ zonesConfiguration.defaultDistributionZone().dataNodesAutoAdjustScaleUp().listen(onUpdateScaleUp());
+ zonesConfiguration.defaultDistributionZone().dataNodesAutoAdjustScaleDown().listen(onUpdateScaleDown());
// Init timers after restart.
zonesState.putIfAbsent(DEFAULT_ZONE_ID, new ZoneState(executor));
@@ -508,6 +520,82 @@ public class DistributionZoneManager implements IgniteComponent {
}
}
+ /**
+ * Creates configuration listener for updates of scale up value.
+ *
+ * @return Configuration listener for updates of scale up value.
+ */
+ private ConfigurationListener<Integer> onUpdateScaleUp() {
+ return ctx -> {
+ int zoneId = ctx.config(DistributionZoneConfiguration.class).zoneId().value();
+
+ if (ctx.oldValue() == null) {
+ // zone creation, already handled in a separate listener.
+ return completedFuture(null);
+ }
+
+ int newScaleUp = ctx.newValue().intValue();
+
+ // It is safe to zonesTimers.get(zoneId) in term of NPE because meta storage notifications are one-threaded
+ // and this map will be initialized on a manager start or with onCreate configuration notification
+ ZoneState zoneState = zonesState.get(zoneId);
+
+ if (newScaleUp != INFINITE_TIMER_VALUE) {
+ Optional<Long> highestRevision = zoneState.highestRevision(true);
+
+ assert highestRevision.isEmpty() || ctx.storageRevision() >= highestRevision.get() : "Expected revision that "
+ + "is greater or equal to already seen meta storage events.";
+
+ zoneState.rescheduleScaleUp(
+ newScaleUp,
+ () -> saveDataNodesToMetaStorageOnScaleUp(zoneId, ctx.storageRevision())
+ );
+ } else {
+ zoneState.stopScaleUp();
+ }
+
+ return completedFuture(null);
+ };
+ }
+
+ /**
+ * Creates configuration listener for updates of scale down value.
+ *
+ * @return Configuration listener for updates of scale down value.
+ */
+ private ConfigurationListener<Integer> onUpdateScaleDown() {
+ return ctx -> {
+ int zoneId = ctx.config(DistributionZoneConfiguration.class).zoneId().value();
+
+ if (ctx.oldValue() == null) {
+ // zone creation, already handled in a separate listener.
+ return completedFuture(null);
+ }
+
+ int newScaleDown = ctx.newValue().intValue();
+
+ // It is safe to zonesTimers.get(zoneId) in term of NPE because meta storage notifications are one-threaded
+ // and this map will be initialized on a manager start or with onCreate configuration notification
+ ZoneState zoneState = zonesState.get(zoneId);
+
+ if (newScaleDown != INFINITE_TIMER_VALUE) {
+ Optional<Long> highestRevision = zoneState.highestRevision(false);
+
+ assert highestRevision.isEmpty() || ctx.storageRevision() >= highestRevision.get() : "Expected revision that "
+ + "is greater or equal to already seen meta storage events.";
+
+ zoneState.rescheduleScaleDown(
+ newScaleDown,
+ () -> saveDataNodesToMetaStorageOnScaleDown(zoneId, ctx.storageRevision())
+ );
+ } else {
+ zoneState.stopScaleDown();
+ }
+
+ return completedFuture(null);
+ };
+ }
+
/** {@inheritDoc} */
@Override
public void stop() throws Exception {
@@ -533,7 +621,7 @@ public class DistributionZoneManager implements IgniteComponent {
zonesState.putIfAbsent(zoneId, zoneState);
- saveDataNodesAndUpdateTriggerKeysInMetaStorage(zoneId, ctx.storageRevision(), toBytes(logicalTopology));
+ saveDataNodesAndUpdateTriggerKeysInMetaStorage(zoneId, ctx.storageRevision(), logicalTopology);
return completedFuture(null);
}
@@ -550,35 +638,6 @@ public class DistributionZoneManager implements IgniteComponent {
return completedFuture(null);
}
-
- @Override
- public CompletableFuture<?> onUpdate(ConfigurationNotificationEvent<DistributionZoneView> ctx) {
- int zoneId = ctx.newValue().zoneId();
-
- int oldScaleUp;
-
- // ctx.oldValue() could be null for the default zone on a first start.
- if (ctx.oldValue() == null) {
- oldScaleUp = Integer.MAX_VALUE;
- } else {
- oldScaleUp = ctx.oldValue().dataNodesAutoAdjustScaleUp();
- }
-
- int newScaleUp = ctx.newValue().dataNodesAutoAdjustScaleUp();
-
- if (newScaleUp != Integer.MAX_VALUE && oldScaleUp != newScaleUp) {
- // It is safe to zonesTimers.get(zoneId) in term of NPE because meta storage notifications are one-threaded
- zonesState.get(zoneId).rescheduleScaleUp(
- newScaleUp,
- () -> CompletableFuture.supplyAsync(
- () -> saveDataNodesToMetaStorageOnScaleUp(zoneId, ctx.storageRevision()),
- Runnable::run
- )
- );
- }
-
- return completedFuture(null);
- }
}
/**
@@ -590,7 +649,7 @@ public class DistributionZoneManager implements IgniteComponent {
* @param revision Revision of an event that has triggered this method.
* @param dataNodes Data nodes.
*/
- private void saveDataNodesAndUpdateTriggerKeysInMetaStorage(int zoneId, long revision, byte[] dataNodes) {
+ private void saveDataNodesAndUpdateTriggerKeysInMetaStorage(int zoneId, long revision, Set<String> dataNodes) {
if (!busyLock.enterBusy()) {
throw new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException());
}
@@ -600,7 +659,7 @@ public class DistributionZoneManager implements IgniteComponent {
// so we do not react on a stale events
CompoundCondition triggerKeyCondition = triggerKeyConditionForZonesChanges(revision, zoneId);
- Update dataNodesAndTriggerKeyUpd = updateDataNodesAndTriggerKeys(zoneId, revision, dataNodes);
+ Update dataNodesAndTriggerKeyUpd = updateDataNodesAndTriggerKeys(zoneId, revision, toBytes(toDataNodesMap(dataNodes)));
Iif iif = iif(triggerKeyCondition, dataNodesAndTriggerKeyUpd, ops().yield(false));
@@ -655,20 +714,20 @@ public class DistributionZoneManager implements IgniteComponent {
private static void updateZoneChange(DistributionZoneChange zoneChange, DistributionZoneConfigurationParameters distributionZoneCfg) {
if (distributionZoneCfg.dataNodesAutoAdjust() != null) {
zoneChange.changeDataNodesAutoAdjust(distributionZoneCfg.dataNodesAutoAdjust());
- zoneChange.changeDataNodesAutoAdjustScaleUp(Integer.MAX_VALUE);
- zoneChange.changeDataNodesAutoAdjustScaleDown(Integer.MAX_VALUE);
+ zoneChange.changeDataNodesAutoAdjustScaleUp(INFINITE_TIMER_VALUE);
+ zoneChange.changeDataNodesAutoAdjustScaleDown(INFINITE_TIMER_VALUE);
}
if (distributionZoneCfg.dataNodesAutoAdjustScaleUp() != null) {
zoneChange.changeDataNodesAutoAdjustScaleUp(
distributionZoneCfg.dataNodesAutoAdjustScaleUp());
- zoneChange.changeDataNodesAutoAdjust(Integer.MAX_VALUE);
+ zoneChange.changeDataNodesAutoAdjust(INFINITE_TIMER_VALUE);
}
if (distributionZoneCfg.dataNodesAutoAdjustScaleDown() != null) {
zoneChange.changeDataNodesAutoAdjustScaleDown(
distributionZoneCfg.dataNodesAutoAdjustScaleDown());
- zoneChange.changeDataNodesAutoAdjust(Integer.MAX_VALUE);
+ zoneChange.changeDataNodesAutoAdjust(INFINITE_TIMER_VALUE);
}
}
@@ -819,7 +878,7 @@ public class DistributionZoneManager implements IgniteComponent {
saveDataNodesAndUpdateTriggerKeysInMetaStorage(
DEFAULT_ZONE_ID,
appliedRevision,
- vaultEntry.value()
+ logicalTopology
);
zonesConfiguration.distributionZones().value().namedListKeys()
@@ -829,7 +888,7 @@ public class DistributionZoneManager implements IgniteComponent {
saveDataNodesAndUpdateTriggerKeysInMetaStorage(
zoneId,
appliedRevision,
- vaultEntry.value()
+ logicalTopology
);
});
}
@@ -914,7 +973,8 @@ public class DistributionZoneManager implements IgniteComponent {
addedNodes,
removedNodes,
revision,
- this::saveDataNodesToMetaStorageOnScaleUp
+ this::saveDataNodesToMetaStorageOnScaleUp,
+ this::saveDataNodesToMetaStorageOnScaleDown
);
}
@@ -925,14 +985,16 @@ public class DistributionZoneManager implements IgniteComponent {
* @param addedNodes Nodes that was added to a topology and should be added to zones data nodes.
* @param removedNodes Nodes that was removed from a topology and should be removed from zones data nodes.
* @param revision Revision that triggered that event.
- * @param saveDataNodes Function that save nodes to a zone's data nodes.
+ * @param saveDataNodesOnScaleUp Function that saves nodes to a zone's data nodes in case of scale up was triggered.
+ * @param saveDataNodesOnScaleDown Function that saves nodes to a zone's data nodes in case of scale down was triggered.
*/
void scheduleTimers(
DistributionZoneView zoneCfg,
Set<String> addedNodes,
Set<String> removedNodes,
long revision,
- BiFunction<Integer, Long, CompletableFuture<Void>> saveDataNodes
+ BiFunction<Integer, Long, CompletableFuture<Void>> saveDataNodesOnScaleUp,
+ BiFunction<Integer, Long, CompletableFuture<Void>> saveDataNodesOnScaleDown
) {
int autoAdjust = zoneCfg.dataNodesAutoAdjust();
int autoAdjustScaleDown = zoneCfg.dataNodesAutoAdjustScaleDown();
@@ -940,26 +1002,26 @@ public class DistributionZoneManager implements IgniteComponent {
int zoneId = zoneCfg.zoneId();
- if ((!addedNodes.isEmpty() || !removedNodes.isEmpty()) && autoAdjust != Integer.MAX_VALUE) {
+ if ((!addedNodes.isEmpty() || !removedNodes.isEmpty()) && autoAdjust != INFINITE_TIMER_VALUE) {
//TODO: IGNITE-18134 Create scheduler with dataNodesAutoAdjust timer.
throw new UnsupportedOperationException("Data nodes auto adjust is not supported.");
} else {
- if (!addedNodes.isEmpty() && autoAdjustScaleUp != Integer.MAX_VALUE) {
- //TODO: IGNITE-18121 Create scale up scheduler with dataNodesAutoAdjustScaleUp timer.
- zonesState.get(zoneId).addNodesToDataNodes(addedNodes, revision);
+ if (!addedNodes.isEmpty() && autoAdjustScaleUp != INFINITE_TIMER_VALUE) {
+ zonesState.get(zoneId).nodesToAddToDataNodes(addedNodes, revision);
zonesState.get(zoneId).rescheduleScaleUp(
autoAdjustScaleUp,
- () -> CompletableFuture.supplyAsync(
- () -> saveDataNodes.apply(zoneId, revision),
- Runnable::run
- )
+ () -> saveDataNodesOnScaleUp.apply(zoneId, revision)
);
}
- if (!removedNodes.isEmpty() && autoAdjustScaleDown != Integer.MAX_VALUE) {
- //TODO: IGNITE-18132 Create scale down scheduler with dataNodesAutoAdjustScaleDown timer.
- throw new UnsupportedOperationException("Data nodes auto adjust scale down is not supported.");
+ if (!removedNodes.isEmpty() && autoAdjustScaleDown != INFINITE_TIMER_VALUE) {
+ zonesState.get(zoneId).nodesToRemoveFromDataNodes(removedNodes, revision);
+
+ zonesState.get(zoneId).rescheduleScaleDown(
+ autoAdjustScaleDown,
+ () -> saveDataNodesOnScaleDown.apply(zoneId, revision)
+ );
}
}
}
@@ -993,7 +1055,7 @@ public class DistributionZoneManager implements IgniteComponent {
/**
* Method updates data nodes value for the specified zone after scale up timer timeout, sets {@code revision} to the
- * {@link DistributionZonesUtil#zoneScaleUpChangeTriggerKey(int)}, {@link DistributionZonesUtil#zoneScaleDownChangeTriggerKey(int)} and
+ * {@link DistributionZonesUtil#zoneScaleUpChangeTriggerKey(int)} and
* {@link DistributionZonesUtil#zonesChangeTriggerKey(int)} if it passes the condition.
*
* @param zoneId Unique id of a zone
@@ -1024,7 +1086,7 @@ public class DistributionZoneManager implements IgniteComponent {
return completedFuture(null);
}
- Set<String> dataNodesFromMetaStorage = fromBytes(values.get(zoneDataNodesKey(zoneId)).value());
+ Map<String, Integer> dataNodesFromMetaStorage = fromBytes(values.get(zoneDataNodesKey(zoneId)).value());
long scaleUpTriggerRevision = bytesToLong(values.get(zoneScaleUpChangeTriggerKey(zoneId)).value());
@@ -1034,15 +1096,14 @@ public class DistributionZoneManager implements IgniteComponent {
return completedFuture(null);
}
- Set<String> deltaToAdd = zoneState.nodesToAddToDataNodes(scaleUpTriggerRevision, scaleDownTriggerRevision, revision);
+ List<String> deltaToAdd = zoneState.nodesToBeAddedToDataNodes(scaleUpTriggerRevision, revision);
- if (deltaToAdd.isEmpty()) {
- return completedFuture(null);
- }
+ Map<String, Integer> newDataNodes = new HashMap<>(dataNodesFromMetaStorage);
- Set<String> newDataNodes = new HashSet<>(dataNodesFromMetaStorage);
+ deltaToAdd.forEach(n -> newDataNodes.merge(n, 1, Integer::sum));
- newDataNodes.addAll(deltaToAdd);
+ // Remove redundant nodes that are not presented in the data nodes.
+ newDataNodes.entrySet().removeIf(e -> e.getValue() == 0);
Update dataNodesAndTriggerKeyUpd = updateDataNodesAndScaleUpTriggerKey(zoneId, revision, toBytes(newDataNodes));
@@ -1079,6 +1140,93 @@ public class DistributionZoneManager implements IgniteComponent {
}
}
+ /**
+ * Method updates data nodes value for the specified zone after scale down timer timeout, sets {@code revision} to the
+ * {@link DistributionZonesUtil#zoneScaleDownChangeTriggerKey(int)} and
+ * {@link DistributionZonesUtil#zonesChangeTriggerKey(int)} if it passes the condition.
+ *
+ * @param zoneId Unique id of a zone
+ * @param revision Revision of an event that has triggered this method.
+ */
+ CompletableFuture<Void> saveDataNodesToMetaStorageOnScaleDown(int zoneId, long revision) {
+ if (!busyLock.enterBusy()) {
+ throw new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException());
+ }
+
+ try {
+ ZoneState zoneState = zonesState.get(zoneId);
+
+ if (zoneState == null) {
+ // Zone was deleted
+ return completedFuture(null);
+ }
+
+ Set<ByteArray> keysToGetFromMs = Set.of(
+ zoneDataNodesKey(zoneId),
+ zoneScaleUpChangeTriggerKey(zoneId),
+ zoneScaleDownChangeTriggerKey(zoneId)
+ );
+
+ return metaStorageManager.getAll(keysToGetFromMs).thenCompose(values -> inBusyLock(busyLock, () -> {
+ if (values.containsValue(null)) {
+ // Zone was deleted
+ return completedFuture(null);
+ }
+
+ Map<String, Integer> dataNodesFromMetaStorage = fromBytes(values.get(zoneDataNodesKey(zoneId)).value());
+
+ long scaleUpTriggerRevision = bytesToLong(values.get(zoneScaleUpChangeTriggerKey(zoneId)).value());
+
+ long scaleDownTriggerRevision = bytesToLong(values.get(zoneScaleDownChangeTriggerKey(zoneId)).value());
+
+ if (revision <= scaleDownTriggerRevision) {
+ return completedFuture(null);
+ }
+
+ List<String> deltaToRemove = zoneState.nodesToBeRemovedFromDataNodes(scaleDownTriggerRevision, revision);
+
+ Map<String, Integer> newDataNodes = new HashMap<>(dataNodesFromMetaStorage);
+
+ deltaToRemove.forEach(n -> newDataNodes.merge(n, -1, Integer::sum));
+
+ // Remove redundant nodes that are not presented in the data nodes.
+ newDataNodes.entrySet().removeIf(e -> e.getValue() == 0);
+
+ Update dataNodesAndTriggerKeyUpd = updateDataNodesAndScaleDownTriggerKey(zoneId, revision, toBytes(newDataNodes));
+
+ Iif iif = iif(
+ triggerScaleUpScaleDownKeysCondition(scaleUpTriggerRevision, scaleDownTriggerRevision, zoneId),
+ dataNodesAndTriggerKeyUpd,
+ ops().yield(false)
+ );
+
+ return metaStorageManager.invoke(iif)
+ .thenApply(StatementResult::getAsBoolean)
+ .thenApply(invokeResult -> inBusyLock(busyLock, () -> {
+ if (invokeResult) {
+ zoneState.cleanUp(Math.min(scaleUpTriggerRevision, revision));
+ } else {
+ LOG.debug("Updating data nodes for a zone has not succeeded [zoneId = {}]", zoneId);
+
+ return saveDataNodesToMetaStorageOnScaleUp(zoneId, revision);
+ }
+
+ return completedFuture(null);
+ }));
+ })).handle((v, e) -> {
+ if (e != null) {
+ LOG.warn("Failed to update zones' dataNodes value [zoneId = {}]", e, zoneId);
+
+ return CompletableFuture.<Void>failedFuture(e);
+ }
+
+ return CompletableFuture.<Void>completedFuture(null);
+ }).thenCompose(Function.identity());
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
/**
* Class responsible for storing state for a distribution zone.
* States are needed to track nodes that we want to add or remove from the data nodes,
@@ -1141,7 +1289,6 @@ public class DistributionZoneManager implements IgniteComponent {
* @param runnable Custom logic to run.
*/
synchronized void rescheduleScaleDown(long delay, Runnable runnable) {
- //TODO: IGNITE-18132 Create scale down scheduler with dataNodesAutoAdjustScaleDown timer.
if (scaleDownTask != null) {
scaleDownTask.cancel(false);
}
@@ -1153,10 +1300,24 @@ public class DistributionZoneManager implements IgniteComponent {
* Cancels task for scale up and scale down.
*/
synchronized void stopTimers() {
+ stopScaleUp();
+
+ stopScaleDown();
+ }
+
+ /**
+ * Cancels task for scale up.
+ */
+ synchronized void stopScaleUp() {
if (scaleUpTask != null) {
scaleUpTask.cancel(false);
}
+ }
+ /**
+ * Cancels task for scale down.
+ */
+ synchronized void stopScaleDown() {
if (scaleDownTask != null) {
scaleDownTask.cancel(false);
}
@@ -1165,30 +1326,27 @@ public class DistributionZoneManager implements IgniteComponent {
/**
* Returns a set of nodes that should be added to zone's data nodes.
*
- * @param scaleUpRevision Last revision of the scale up event.
- * @param scaleDownRevision Last revision of the scale down event.
+ * @param scaleUpRevision Last revision of the scale up event. Nodes that were associated with this event
+ * or with the lower revisions, won't be included in the accumulation.
* @param revision Revision of the event for which this data nodes is needed.
- * @return Set of nodes that should be added to zone's data nodes.
+ * Nodes that were associated with this event will be included.
+ * @return List of nodes that should be added to zone's data nodes.
*/
- Set<String> nodesToAddToDataNodes(long scaleUpRevision, long scaleDownRevision, long revision) {
- Long toKey = topologyAugmentationMap.floorKey(revision);
-
- if (toKey == null) {
- return Set.of();
- }
-
- Set<String> nodesToAdd = accumulateNodes(scaleUpRevision, toKey, true);
-
- Set<String> nodesToRemove = accumulateNodes(scaleUpRevision, Math.max(toKey, scaleDownRevision), false);
-
- nodesToAdd.removeAll(nodesToRemove);
-
- return nodesToAdd;
+ List<String> nodesToBeAddedToDataNodes(long scaleUpRevision, long revision) {
+ return accumulateNodes(scaleUpRevision, revision, true);
}
- public Set<String> nodesToRemoveFromDataNodes() {
- //TODO: IGNITE-18132 Create scale down scheduler with dataNodesAutoAdjustScaleDown timer.
- return null;
+ /**
+ * Returns a set of nodes that should be removed from zone's data nodes.
+ *
+ * @param scaleDownRevision Last revision of the scale down event. Nodes that were associated with this event
+ * or with the lower revisions, won't be included in the accumulation.
+ * @param revision Revision of the event for which this data nodes is needed.
+ * Nodes that were associated with this event will be included.
+ * @return List of nodes that should be removed from zone's data nodes.
+ */
+ List<String> nodesToBeRemovedFromDataNodes(long scaleDownRevision, long revision) {
+ return accumulateNodes(scaleDownRevision, revision, false);
}
/**
@@ -1197,7 +1355,7 @@ public class DistributionZoneManager implements IgniteComponent {
* @param nodes Nodes to add to zone's data nodes.
* @param revision Revision of the event that triggered this addition.
*/
- void addNodesToDataNodes(Set<String> nodes, long revision) {
+ void nodesToAddToDataNodes(Set<String> nodes, long revision) {
topologyAugmentationMap.put(revision, new Augmentation(nodes, true));
}
@@ -1207,30 +1365,62 @@ public class DistributionZoneManager implements IgniteComponent {
* @param nodes Nodes to remove from zone's data nodes.
* @param revision Revision of the event that triggered this addition.
*/
- void removeNodesFromDataNodes(Set<String> nodes, long revision) {
- topologyAugmentationMap.put(revision, new Augmentation(nodes, true));
+ void nodesToRemoveFromDataNodes(Set<String> nodes, long revision) {
+ topologyAugmentationMap.put(revision, new Augmentation(nodes, false));
}
/**
- * Accumulate nodes from the {@link ZoneState#topologyAugmentationMap} starting from the {@code fromKey} (including)
+ * Accumulate nodes from the {@link ZoneState#topologyAugmentationMap} starting from the {@code fromKey} (excluding)
* to the {@code toKey} (including), where flag {@code addition} indicates whether we should accumulate nodes that should be
* added to data nodes, or removed.
*
- * @param fromKey Starting key (including).
+ * @param fromKey Starting key (excluding).
* @param toKey Ending key (including).
* @param addition Indicates whether we should accumulate nodes that should be added to data nodes, or removed.
* @return Accumulated nodes.
*/
- private Set<String> accumulateNodes(long fromKey, long toKey, boolean addition) {
- return topologyAugmentationMap.subMap(fromKey, true, toKey, true).values()
+ private List<String> accumulateNodes(long fromKey, long toKey, boolean addition) {
+ return topologyAugmentationMap.subMap(fromKey, false, toKey, true).values()
.stream()
.filter(a -> a.addition == addition)
.flatMap(a -> a.nodeNames.stream())
- .collect(toSet());
+ .collect(toList());
}
+ /**
+ * Cleans {@code topologyAugmentationMap} to the key, to which is safe to delete.
+ * Safe means that this key was handled both by scale up and scale down schedulers.
+ *
+ * @param toKey Key in map to which is safe to delete data from {@code topologyAugmentationMap}.
+ */
private void cleanUp(long toKey) {
- //TODO: IGNITE-18132 Create scale down scheduler with dataNodesAutoAdjustScaleDown timer.
+ topologyAugmentationMap.headMap(toKey, true).clear();
+ }
+
+ /**
+ * Returns the highest revision which is presented in the {@link ZoneState#topologyAugmentationMap()} taking into account
+ * the {@code addition} flag.
+ *
+ * @param addition Flag indicating the type of the nodes for which we want to find the highest revision.
+ * @return The highest revision which is presented in the {@link ZoneState#topologyAugmentationMap()} taking into account
+ * the {@code addition} flag.
+ */
+ Optional<Long> highestRevision(boolean addition) {
+ return topologyAugmentationMap().entrySet()
+ .stream()
+ .filter(e -> e.getValue().addition == addition)
+ .max(Map.Entry.comparingByKey())
+ .map(Map.Entry::getKey);
+ }
+
+ @TestOnly
+ synchronized ScheduledFuture<?> scaleUpTask() {
+ return scaleUpTask;
+ }
+
+ @TestOnly
+ synchronized ScheduledFuture<?> scaleDownTask() {
+ return scaleDownTask;
}
}
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 ac9cb9905d..b0e544e560 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
@@ -26,7 +26,11 @@ import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
import static org.apache.ignite.internal.metastorage.dsl.Operations.remove;
import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
import java.util.Set;
+import java.util.stream.Collectors;
import org.apache.ignite.internal.metastorage.dsl.CompoundCondition;
import org.apache.ignite.internal.metastorage.dsl.Update;
import org.apache.ignite.internal.util.ByteUtils;
@@ -96,7 +100,7 @@ public class DistributionZonesUtil {
* The key needed for processing an event about zone's creation and deletion.
* With this key we can be sure that event was triggered only once.
*/
- static ByteArray zonesChangeTriggerKey(int zoneId) {
+ public static ByteArray zonesChangeTriggerKey(int zoneId) {
return new ByteArray(DISTRIBUTION_ZONES_CHANGE_TRIGGER_KEY_PREFIX + zoneId);
}
@@ -104,7 +108,7 @@ public class DistributionZonesUtil {
* The key needed for processing an event about zone's data node propagation on scale up.
* With this key we can be sure that event was triggered only once.
*/
- static ByteArray zoneScaleUpChangeTriggerKey(int zoneId) {
+ public static ByteArray zoneScaleUpChangeTriggerKey(int zoneId) {
return new ByteArray(DISTRIBUTION_ZONE_SCALE_UP_CHANGE_TRIGGER_PREFIX + zoneId);
}
@@ -112,7 +116,7 @@ public class DistributionZonesUtil {
* The key needed for processing an event about zone's data node propagation on scale down.
* With this key we can be sure that event was triggered only once.
*/
- static ByteArray zoneScaleDownChangeTriggerKey(int zoneId) {
+ public static ByteArray zoneScaleDownChangeTriggerKey(int zoneId) {
return new ByteArray(DISTRIBUTION_ZONE_SCALE_DOWN_CHANGE_TRIGGER_PREFIX + zoneId);
}
@@ -120,7 +124,7 @@ public class DistributionZonesUtil {
* The key that represents logical topology nodes, needed for distribution zones. It is needed to store them in the metastore
* to serialize data nodes changes triggered by topology changes and changes of distribution zones configurations.
*/
- static ByteArray zonesLogicalTopologyKey() {
+ public static ByteArray zonesLogicalTopologyKey() {
return DISTRIBUTION_ZONE_LOGICAL_TOPOLOGY_KEY;
}
@@ -177,6 +181,14 @@ public class DistributionZonesUtil {
).yield(true);
}
+ static Update updateDataNodesAndScaleDownTriggerKey(int zoneId, long revision, byte[] nodes) {
+ return ops(
+ put(zoneDataNodesKey(zoneId), nodes),
+ put(zoneScaleDownChangeTriggerKey(zoneId), ByteUtils.longToBytes(revision))
+ ).yield(true);
+ }
+
+
/**
* Updates data nodes value for a zone and set {@code revision} to {@link DistributionZonesUtil#zoneScaleUpChangeTriggerKey(int)},
* {@link DistributionZonesUtil#zoneScaleDownChangeTriggerKey(int)} and {@link DistributionZonesUtil#zonesChangeTriggerKey(int)}.
@@ -226,4 +238,32 @@ public class DistributionZonesUtil {
put(zonesLogicalTopologyKey(), ByteUtils.toBytes(logicalTopology))
).yield(true);
}
+
+ /**
+ * Returns a set of data nodes retrieved from data nodes map, which value is more than 0.
+ *
+ * @param dataNodesMap This map has the following structure: node name is mapped to an integer,
+ * an integer represents counter for node joining or leaving the topology.
+ * Joining increases the counter, leaving decreases.
+ * @return Returns a set of data nodes retrieved from data nodes map, which value is more than 0.
+ */
+ public static Set<String> dataNodes(Map<String, Integer> dataNodesMap) {
+ return dataNodesMap.entrySet().stream().filter(e -> e.getValue() > 0).map(Entry::getKey).collect(Collectors.toSet());
+ }
+
+ /**
+ * Returns a map from a set of data nodes. This map has the following structure: node name is mapped to integer,
+ * integer represents how often node joined or leaved topology. In this case, set of nodes is interpreted as nodes
+ * that joined topology, so all mappings will be node -> 1.
+ *
+ * @param dataNodes Set of data nodes.
+ * @return Returns a map from a set of data nodes.
+ */
+ public static Map<String, Integer> toDataNodesMap(Set<String> dataNodes) {
+ Map<String, Integer> dataNodesMap = new HashMap<>();
+
+ dataNodes.forEach(n -> dataNodesMap.merge(n, 1, Integer::sum));
+
+ return dataNodesMap;
+ }
}
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/configuration/DistributionZoneConfigurationSchema.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/configuration/DistributionZoneConfigurationSchema.java
index cf45714db7..cf1b00a589 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/configuration/DistributionZoneConfigurationSchema.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/configuration/DistributionZoneConfigurationSchema.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.distributionzones.configuration;
import static org.apache.ignite.internal.distributionzones.DistributionZoneManager.DEFAULT_ZONE_ID;
+import static org.apache.ignite.internal.distributionzones.DistributionZoneManager.INFINITE_TIMER_VALUE;
import org.apache.ignite.configuration.annotation.Config;
import org.apache.ignite.configuration.annotation.InjectedName;
@@ -43,15 +44,15 @@ public class DistributionZoneConfigurationSchema {
/** Timeout in seconds between node added or node left topology event itself and data nodes switch. */
@Range(min = 0)
@Value(hasDefault = true)
- public int dataNodesAutoAdjust = Integer.MAX_VALUE;
+ public int dataNodesAutoAdjust = INFINITE_TIMER_VALUE;
/** Timeout in seconds between node added topology event itself and data nodes switch. */
@Range(min = 0)
@Value(hasDefault = true)
- public int dataNodesAutoAdjustScaleUp = Integer.MAX_VALUE;
+ public int dataNodesAutoAdjustScaleUp = INFINITE_TIMER_VALUE;
/** Timeout in seconds between node left topology event itself and data nodes switch. */
@Range(min = 0)
@Value(hasDefault = true)
- public int dataNodesAutoAdjustScaleDown = Integer.MAX_VALUE;
+ public int dataNodesAutoAdjustScaleDown = INFINITE_TIMER_VALUE;
}
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerConfigurationChangesTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerConfigurationChangesTest.java
index dde5995b89..521dab900f 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerConfigurationChangesTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerConfigurationChangesTest.java
@@ -23,6 +23,9 @@ import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zoneScaleUpChangeTriggerKey;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesChangeTriggerKey;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesLogicalTopologyKey;
+import static org.apache.ignite.internal.distributionzones.util.DistributionZonesTestUtil.assertDataNodesForZone;
+import static org.apache.ignite.internal.distributionzones.util.DistributionZonesTestUtil.assertZoneScaleUpChangeTriggerKey;
+import static org.apache.ignite.internal.distributionzones.util.DistributionZonesTestUtil.assertZonesChangeTriggerKey;
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;
@@ -39,7 +42,6 @@ import static org.mockito.Mockito.when;
import java.io.Serializable;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -73,7 +75,6 @@ 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.util.ByteUtils;
import org.apache.ignite.internal.vault.VaultEntry;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.lang.ByteArray;
@@ -292,22 +293,22 @@ public class DistributionZoneManagerConfigurationChangesTest extends IgniteAbstr
@Test
void testDataNodesPropagationAfterZoneCreation() throws Exception {
- assertDataNodesForZone(1, null);
+ assertDataNodesForZone(1, null, keyValueStorage);
distributionZoneManager.createZone(new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).build()).get();
- assertDataNodesForZone(1, nodes);
+ assertDataNodesForZone(1, nodes, keyValueStorage);
- assertZoneScaleUpChangeTriggerKey(1, 1);
+ assertZoneScaleUpChangeTriggerKey(1, 1, keyValueStorage);
- assertZonesChangeTriggerKey(1, 1);
+ assertZonesChangeTriggerKey(1, 1, keyValueStorage);
}
@Test
void testZoneDeleteRemovesMetaStorageKey() throws Exception {
distributionZoneManager.createZone(new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).build()).get();
- assertDataNodesForZone(1, nodes);
+ assertDataNodesForZone(1, nodes, keyValueStorage);
distributionZoneManager.dropZone(ZONE_NAME).get();
@@ -323,10 +324,10 @@ public class DistributionZoneManagerConfigurationChangesTest extends IgniteAbstr
distributionZoneManager.createZone(new DistributionZoneConfigurationParameters.Builder(NEW_ZONE_NAME).build()).get();
- assertZoneScaleUpChangeTriggerKey(1, 1);
- assertZoneScaleUpChangeTriggerKey(2, 2);
- assertZonesChangeTriggerKey(1, 1);
- assertZonesChangeTriggerKey(2, 2);
+ assertZoneScaleUpChangeTriggerKey(1, 1, keyValueStorage);
+ assertZoneScaleUpChangeTriggerKey(2, 2, keyValueStorage);
+ assertZonesChangeTriggerKey(1, 1, keyValueStorage);
+ assertZonesChangeTriggerKey(2, 2, keyValueStorage);
}
@Test
@@ -337,16 +338,16 @@ public class DistributionZoneManagerConfigurationChangesTest extends IgniteAbstr
verify(keyValueStorage, timeout(1000).times(1)).invoke(any());
- assertZonesChangeTriggerKey(100, 1);
+ assertZonesChangeTriggerKey(100, 1, keyValueStorage);
- assertDataNodesForZone(1, null);
+ assertDataNodesForZone(1, null, keyValueStorage);
}
@Test
void testZoneDeleteDoNotRemoveMetaStorageKey() throws Exception {
distributionZoneManager.createZone(new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).build()).get();
- assertDataNodesForZone(1, nodes);
+ assertDataNodesForZone(1, nodes, keyValueStorage);
keyValueStorage.put(zonesChangeTriggerKey(1).bytes(), longToBytes(100));
@@ -354,7 +355,7 @@ public class DistributionZoneManagerConfigurationChangesTest extends IgniteAbstr
verify(keyValueStorage, timeout(1000).times(2)).invoke(any());
- assertDataNodesForZone(1, nodes);
+ assertDataNodesForZone(1, nodes, keyValueStorage);
}
private void mockVaultZonesLogicalTopologyKey(Set<String> nodes) {
@@ -363,27 +364,4 @@ public class DistributionZoneManagerConfigurationChangesTest extends IgniteAbstr
when(vaultMgr.get(zonesLogicalTopologyKey()))
.thenReturn(completedFuture(new VaultEntry(zonesLogicalTopologyKey(), newLogicalTopology)));
}
-
- private void assertDataNodesForZone(int zoneId, @Nullable Set<String> clusterNodes) throws InterruptedException {
- byte[] nodes = clusterNodes == null ? null : toBytes(clusterNodes);
-
- assertTrue(waitForCondition(() -> Arrays.equals(keyValueStorage.get(zoneDataNodesKey(zoneId).bytes()).value(), nodes), 1000));
- }
-
- private void assertZoneScaleUpChangeTriggerKey(int revision, int zoneId) throws InterruptedException {
- assertTrue(
- waitForCondition(
- () -> ByteUtils.bytesToLong(keyValueStorage.get(zoneScaleUpChangeTriggerKey(zoneId).bytes()).value()) == revision,
- 2000
- )
- );
- }
-
- private void assertZonesChangeTriggerKey(int revision, int zoneId) throws InterruptedException {
- assertTrue(
- waitForCondition(
- () -> ByteUtils.bytesToLong(keyValueStorage.get(zonesChangeTriggerKey(zoneId).bytes()).value()) == revision, 1000
- )
- );
- }
}
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 99aaaf4202..d0b57d3cf3 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
@@ -20,8 +20,8 @@ package org.apache.ignite.internal.distributionzones;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.configuration.annotation.ConfigurationType.DISTRIBUTED;
import static org.apache.ignite.internal.cluster.management.topology.LogicalTopologyImpl.LOGICAL_TOPOLOGY_KEY;
-import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesLogicalTopologyKey;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesLogicalTopologyVersionKey;
+import static org.apache.ignite.internal.distributionzones.util.DistributionZonesTestUtil.assertLogicalTopology;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -36,12 +36,10 @@ import static org.mockito.Mockito.when;
import java.io.Serializable;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
-import java.util.stream.Collectors;
import org.apache.ignite.configuration.NamedConfigurationTree;
import org.apache.ignite.configuration.NamedListView;
import org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
@@ -286,7 +284,7 @@ public class DistributionZoneManagerLogicalTopologyEventsTest {
assertLogicalTopVer(1L);
- assertLogicalTopology(clusterNodes);
+ assertLogicalTopology(clusterNodes, keyValueStorage);
}
@Test
@@ -305,7 +303,7 @@ public class DistributionZoneManagerLogicalTopologyEventsTest {
assertLogicalTopVer(2L);
- assertLogicalTopology(clusterNodes);
+ assertLogicalTopology(clusterNodes, keyValueStorage);
}
@Test
@@ -324,7 +322,7 @@ public class DistributionZoneManagerLogicalTopologyEventsTest {
assertLogicalTopVer(2L);
- assertLogicalTopology(null);
+ assertLogicalTopology(null, keyValueStorage);
}
@Test
@@ -343,7 +341,7 @@ public class DistributionZoneManagerLogicalTopologyEventsTest {
assertLogicalTopVer(3L);
- assertLogicalTopology(null);
+ assertLogicalTopology(null, keyValueStorage);
}
@Test
@@ -366,7 +364,7 @@ public class DistributionZoneManagerLogicalTopologyEventsTest {
var clusterNodes2 = Set.of(node1, node2);
- assertLogicalTopology(clusterNodes2);
+ assertLogicalTopology(clusterNodes2, keyValueStorage);
assertLogicalTopVer(2L);
}
@@ -393,7 +391,7 @@ public class DistributionZoneManagerLogicalTopologyEventsTest {
assertEquals(2L, topology.getLogicalTopology().version());
- assertLogicalTopology(clusterNodes);
+ assertLogicalTopology(clusterNodes, keyValueStorage);
assertLogicalTopVer(4L);
}
@@ -416,13 +414,13 @@ public class DistributionZoneManagerLogicalTopologyEventsTest {
distributionZoneManager1.start();
- assertLogicalTopology(clusterNodes);
+ assertLogicalTopology(clusterNodes, keyValueStorage);
topology.removeNodes(Set.of(node2));
var clusterNodes2 = Set.of(node1);
- assertLogicalTopology(clusterNodes2);
+ assertLogicalTopology(clusterNodes2, keyValueStorage);
assertLogicalTopVer(3L);
}
@@ -451,7 +449,7 @@ public class DistributionZoneManagerLogicalTopologyEventsTest {
topology.removeNodes(Set.of(node2));
- assertLogicalTopology(clusterNodes);
+ assertLogicalTopology(clusterNodes, keyValueStorage);
assertLogicalTopVer(4L);
}
@@ -472,7 +470,7 @@ public class DistributionZoneManagerLogicalTopologyEventsTest {
distributionZoneManager1.start();
- assertLogicalTopology(clusterNodes);
+ assertLogicalTopology(clusterNodes, keyValueStorage);
var clusterNodes2 = Set.of(node1, node2);
@@ -480,7 +478,7 @@ public class DistributionZoneManagerLogicalTopologyEventsTest {
topology.fireTopologyLeap();
- assertLogicalTopology(clusterNodes2);
+ assertLogicalTopology(clusterNodes2, keyValueStorage);
assertLogicalTopVer(10L);
}
@@ -501,7 +499,7 @@ public class DistributionZoneManagerLogicalTopologyEventsTest {
distributionZoneManager1.start();
- assertLogicalTopology(clusterNodes);
+ assertLogicalTopology(clusterNodes, keyValueStorage);
var clusterNodes2 = Set.of(node1, node2);
@@ -511,7 +509,7 @@ public class DistributionZoneManagerLogicalTopologyEventsTest {
topology.fireTopologyLeap();
- assertLogicalTopology(clusterNodes);
+ assertLogicalTopology(clusterNodes, keyValueStorage);
assertLogicalTopVer(11L);
}
@@ -531,12 +529,4 @@ public class DistributionZoneManagerLogicalTopologyEventsTest {
)
);
}
-
- private void assertLogicalTopology(@Nullable Set<ClusterNode> clusterNodes) throws InterruptedException {
- byte[] nodes = clusterNodes == null
- ? null
- : ByteUtils.toBytes(clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
-
- assertTrue(waitForCondition(() -> Arrays.equals(keyValueStorage.get(zonesLogicalTopologyKey().bytes()).value(), nodes), 1000));
- }
}
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerScaleUpTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerScaleUpTest.java
index 4b9d52aeb7..624f9617e7 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerScaleUpTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerScaleUpTest.java
@@ -21,13 +21,22 @@ import static java.util.concurrent.CompletableFuture.completedFuture;
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.INFINITE_TIMER_VALUE;
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.zonesChangeTriggerKey;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesLogicalTopologyKey;
+import static org.apache.ignite.internal.distributionzones.util.DistributionZonesTestUtil.assertDataNodesForZone;
+import static org.apache.ignite.internal.distributionzones.util.DistributionZonesTestUtil.assertLogicalTopology;
+import static org.apache.ignite.internal.distributionzones.util.DistributionZonesTestUtil.assertZoneScaleDownChangeTriggerKey;
+import static org.apache.ignite.internal.distributionzones.util.DistributionZonesTestUtil.assertZoneScaleUpChangeTriggerKey;
+import static org.apache.ignite.internal.distributionzones.util.DistributionZonesTestUtil.mockMetaStorageListener;
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.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
@@ -37,15 +46,13 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
-import java.io.Serializable;
import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
+import java.util.Collections;
import java.util.List;
-import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import org.apache.ignite.configuration.NamedConfigurationTree;
@@ -56,9 +63,9 @@ import org.apache.ignite.internal.cluster.management.raft.TestClusterStateStorag
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.cluster.management.topology.api.LogicalTopologySnapshot;
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.DistributionZoneChange;
import org.apache.ignite.internal.distributionzones.configuration.DistributionZoneConfiguration;
import org.apache.ignite.internal.distributionzones.configuration.DistributionZoneView;
@@ -68,18 +75,12 @@ 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.command.GetAllCommand;
-import org.apache.ignite.internal.metastorage.command.GetCommand;
import org.apache.ignite.internal.metastorage.command.MetaStorageCommandsFactory;
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.server.SimpleInMemoryKeyValueStorage;
import org.apache.ignite.internal.metastorage.server.raft.MetaStorageListener;
-import org.apache.ignite.internal.raft.Command;
-import org.apache.ignite.internal.raft.ReadCommand;
-import org.apache.ignite.internal.raft.WriteCommand;
-import org.apache.ignite.internal.raft.service.CommandClosure;
import org.apache.ignite.internal.raft.service.RaftGroupService;
import org.apache.ignite.internal.schema.configuration.TableChange;
import org.apache.ignite.internal.schema.configuration.TableConfiguration;
@@ -88,8 +89,6 @@ import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
import org.apache.ignite.internal.util.ByteUtils;
import org.apache.ignite.internal.vault.VaultEntry;
import org.apache.ignite.internal.vault.VaultManager;
-import org.apache.ignite.lang.ByteArray;
-import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.NetworkAddress;
import org.jetbrains.annotations.Nullable;
@@ -99,7 +98,7 @@ import org.junit.jupiter.api.Test;
import org.mockito.Mock;
/**
- * Test scenarios for the distribution zone scale up.
+ * Test scenarios for the distribution zone scale up and scale down.
*/
public class DistributionZoneManagerScaleUpTest {
private static final String ZONE_NAME = "zone1";
@@ -128,6 +127,9 @@ public class DistributionZoneManagerScaleUpTest {
@Mock
private ClusterManagementGroupManager cmgManager;
+ @Mock
+ RaftGroupService metaStorageService;
+
@BeforeEach
public void setUp() {
clusterCfgMgr = new ConfigurationManager(
@@ -193,145 +195,100 @@ public class DistributionZoneManagerScaleUpTest {
MetaStorageListener metaStorageListener = new MetaStorageListener(keyValueStorage);
- RaftGroupService metaStorageService = mock(RaftGroupService.class);
+ metaStorageService = mock(RaftGroupService.class);
- // Delegate directly to listener.
- lenient().doAnswer(
- invocationClose -> {
- Command cmd = invocationClose.getArgument(0);
+ mockMetaStorageListener(raftIndex, metaStorageListener, metaStorageService, metaStorageManager);
+ }
- long commandIndex = raftIndex.incrementAndGet();
+ @AfterEach
+ public void tearDown() throws Exception {
+ distributionZoneManager.stop();
- CompletableFuture<Serializable> res = new CompletableFuture<>();
+ clusterCfgMgr.stop();
- CommandClosure<WriteCommand> clo = new CommandClosure<>() {
- /** {@inheritDoc} */
- @Override
- public long index() {
- return commandIndex;
- }
+ keyValueStorage.close();
- /** {@inheritDoc} */
- @Override
- public WriteCommand command() {
- return (WriteCommand) cmd;
- }
+ clusterStateStorage.destroy();
+ }
- /** {@inheritDoc} */
- @Override
- public void result(@Nullable Serializable r) {
- if (r instanceof Throwable) {
- res.completeExceptionally((Throwable) r);
- } else {
- res.complete(r);
- }
- }
- };
+ @Test
+ void testDataNodesPropagationAfterScaleUpTriggered() throws Exception {
+ ClusterNode node1 = new ClusterNode("1", "name1", new NetworkAddress("localhost", 123));
- try {
- metaStorageListener.onWrite(List.of(clo).iterator());
- } catch (Throwable e) {
- res.completeExceptionally(new IgniteInternalException(e));
- }
+ topology.putNode(node1);
- return res;
- }
- ).when(metaStorageService).run(any(WriteCommand.class));
+ Set<ClusterNode> clusterNodes = Set.of(node1);
- lenient().doAnswer(
- invocationClose -> {
- Command cmd = invocationClose.getArgument(0);
+ mockVaultZonesLogicalTopologyKey(clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
- long commandIndex = raftIndex.incrementAndGet();
+ mockCmgLocalNodes();
- CompletableFuture<Serializable> res = new CompletableFuture<>();
+ distributionZoneManager.start();
- CommandClosure<ReadCommand> clo = new CommandClosure<>() {
- /** {@inheritDoc} */
- @Override
- public long index() {
- return commandIndex;
- }
+ assertDataNodesForZone(DEFAULT_ZONE_ID, clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()), keyValueStorage);
- /** {@inheritDoc} */
- @Override
- public ReadCommand command() {
- return (ReadCommand) cmd;
- }
+ ClusterNode node2 = new ClusterNode("2", "name2", new NetworkAddress("localhost", 123));
- /** {@inheritDoc} */
- @Override
- public void result(@Nullable Serializable r) {
- if (r instanceof Throwable) {
- res.completeExceptionally((Throwable) r);
- } else {
- res.complete(r);
- }
- }
- };
+ topology.putNode(node2);
- try {
- metaStorageListener.onRead(List.of(clo).iterator());
- } catch (Throwable e) {
- res.completeExceptionally(new IgniteInternalException(e));
- }
+ Set<ClusterNode> clusterNodes2 = Set.of(node1, node2);
- return res;
- }
- ).when(metaStorageService).run(any(ReadCommand.class));
+ assertLogicalTopology(clusterNodes2, keyValueStorage);
- MetaStorageCommandsFactory commandsFactory = new MetaStorageCommandsFactory();
+ distributionZoneManager.createZone(
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleUp(0).build()
+ ).get();
- lenient().doAnswer(invocationClose -> {
- Iif iif = invocationClose.getArgument(0);
+ assertDataNodesForZone(1, clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()), keyValueStorage);
- MultiInvokeCommand multiInvokeCommand = commandsFactory.multiInvokeCommand().iif(iif).build();
+ assertZoneScaleUpChangeTriggerKey(1, 1, keyValueStorage);
- return metaStorageService.run(multiInvokeCommand);
- }).when(metaStorageManager).invoke(any());
+ watchListenerOnUpdate(topology.getLogicalTopology().nodes().stream().map(ClusterNode::name).collect(Collectors.toSet()), 2);
- lenient().doAnswer(invocationClose -> {
- Set<ByteArray> keysSet = invocationClose.getArgument(0);
+ assertDataNodesForZone(1, clusterNodes2.stream().map(ClusterNode::name).collect(Collectors.toSet()), keyValueStorage);
+ }
+
+ @Test
+ void testDataNodesPropagationAfterScaleDownTriggered() throws Exception {
+ ClusterNode node1 = new ClusterNode("1", "name1", new NetworkAddress("localhost", 123));
+
+ ClusterNode node2 = new ClusterNode("2", "name2", new NetworkAddress("localhost", 123));
- GetAllCommand getAllCommand = commandsFactory.getAllCommand().keys(
- keysSet.stream().map(ByteArray::bytes).collect(Collectors.toList())
- ).revision(0).build();
+ topology.putNode(node1);
- return metaStorageService.<List<Entry>>run(getAllCommand).thenApply(entries -> {
- Map<ByteArray, Entry> res = new HashMap<>();
+ topology.putNode(node2);
- for (Entry e : entries) {
- ByteArray key = new ByteArray(e.key());
+ Set<ClusterNode> clusterNodes = Set.of(node1, node2);
- res.put(key, new EntryImpl(key.bytes(), e.value(), e.revision(), e.updateCounter()));
- }
+ mockVaultZonesLogicalTopologyKey(clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
- return res;
- });
- }).when(metaStorageManager).getAll(any());
+ mockCmgLocalNodes();
- lenient().doAnswer(invocationClose -> {
- ByteArray key = invocationClose.getArgument(0);
+ distributionZoneManager.start();
- GetCommand getCommand = commandsFactory.getCommand().key(key.bytes()).build();
+ assertDataNodesForZone(DEFAULT_ZONE_ID, clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()), keyValueStorage);
- return metaStorageService.<Entry>run(getCommand);
- }).when(metaStorageManager).get(any());
- }
+ topology.removeNodes(Set.of(node2));
- @AfterEach
- public void tearDown() throws Exception {
- distributionZoneManager.stop();
+ Set<ClusterNode> clusterNodes2 = Set.of(node1);
- clusterCfgMgr.stop();
+ assertLogicalTopology(clusterNodes2, keyValueStorage);
- keyValueStorage.close();
+ distributionZoneManager.createZone(
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleDown(0).build()
+ ).get();
- clusterStateStorage.destroy();
+ assertDataNodesForZone(1, clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()), keyValueStorage);
+
+ assertZoneScaleDownChangeTriggerKey(1, 1, keyValueStorage);
+
+ watchListenerOnUpdate(topology.getLogicalTopology().nodes().stream().map(ClusterNode::name).collect(Collectors.toSet()), 2);
+
+ assertDataNodesForZone(1, clusterNodes2.stream().map(ClusterNode::name).collect(Collectors.toSet()), keyValueStorage);
}
@Test
- void testDataNodesPropagationAfterScaleUpTriggered() throws Exception {
+ void testDataNodesPropagationForDefaultZoneAfterScaleUpTriggered() throws Exception {
ClusterNode node1 = new ClusterNode("1", "name1", new NetworkAddress("localhost", 123));
topology.putNode(node1);
@@ -340,11 +297,11 @@ public class DistributionZoneManagerScaleUpTest {
mockVaultZonesLogicalTopologyKey(clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
- mockCmgLocalNodes(1L, clusterNodes);
+ mockCmgLocalNodes();
distributionZoneManager.start();
- assertDataNodesForZone(DEFAULT_ZONE_ID, clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
+ assertDataNodesForZone(DEFAULT_ZONE_ID, clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()), keyValueStorage);
ClusterNode node2 = new ClusterNode("2", "name2", new NetworkAddress("localhost", 123));
@@ -352,57 +309,52 @@ public class DistributionZoneManagerScaleUpTest {
Set<ClusterNode> clusterNodes2 = Set.of(node1, node2);
- mockCmgLocalNodes(1L, clusterNodes2);
+ assertLogicalTopology(clusterNodes2, keyValueStorage);
- assertLogicalTopology(clusterNodes2);
-
- distributionZoneManager.createZone(
- new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleUp(1).build()
+ distributionZoneManager.alterZone(
+ DEFAULT_ZONE_NAME,
+ new DistributionZoneConfigurationParameters.Builder(DEFAULT_ZONE_NAME).dataNodesAutoAdjustScaleUp(0).build()
).get();
- assertDataNodesForZone(1, clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
-
- assertZoneScaleUpChangeTriggerKey(1, 1);
-
watchListenerOnUpdate(topology.getLogicalTopology().nodes().stream().map(ClusterNode::name).collect(Collectors.toSet()), 2);
- assertDataNodesForZone(1, clusterNodes2.stream().map(ClusterNode::name).collect(Collectors.toSet()));
+ assertDataNodesForZone(DEFAULT_ZONE_ID, clusterNodes2.stream().map(ClusterNode::name).collect(Collectors.toSet()), keyValueStorage);
}
@Test
- void testDataNodesPropagationForDefaultZoneAfterScaleUpTriggered() throws Exception {
+ void testDataNodesPropagationForDefaultZoneAfterScaleDownTriggered() throws Exception {
ClusterNode node1 = new ClusterNode("1", "name1", new NetworkAddress("localhost", 123));
+ ClusterNode node2 = new ClusterNode("2", "name2", new NetworkAddress("localhost", 123));
+
topology.putNode(node1);
- Set<ClusterNode> clusterNodes = Set.of(node1);
+ topology.putNode(node2);
+
+ Set<ClusterNode> clusterNodes = Set.of(node1, node2);
mockVaultZonesLogicalTopologyKey(clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
- mockCmgLocalNodes(1L, clusterNodes);
+ mockCmgLocalNodes();
distributionZoneManager.start();
- assertDataNodesForZone(DEFAULT_ZONE_ID, clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
-
- ClusterNode node2 = new ClusterNode("2", "name2", new NetworkAddress("localhost", 123));
-
- topology.putNode(node2);
+ assertDataNodesForZone(DEFAULT_ZONE_ID, clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()), keyValueStorage);
- Set<ClusterNode> clusterNodes2 = Set.of(node1, node2);
+ topology.removeNodes(Set.of(node2));
- mockCmgLocalNodes(1L, clusterNodes2);
+ Set<ClusterNode> clusterNodes2 = Set.of(node1);
- assertLogicalTopology(clusterNodes2);
+ assertLogicalTopology(clusterNodes2, keyValueStorage);
distributionZoneManager.alterZone(
DEFAULT_ZONE_NAME,
- new DistributionZoneConfigurationParameters.Builder(DEFAULT_ZONE_NAME).dataNodesAutoAdjustScaleUp(0).build()
+ new DistributionZoneConfigurationParameters.Builder(DEFAULT_ZONE_NAME).dataNodesAutoAdjustScaleDown(0).build()
).get();
watchListenerOnUpdate(topology.getLogicalTopology().nodes().stream().map(ClusterNode::name).collect(Collectors.toSet()), 2);
- assertDataNodesForZone(DEFAULT_ZONE_ID, clusterNodes2.stream().map(ClusterNode::name).collect(Collectors.toSet()));
+ assertDataNodesForZone(DEFAULT_ZONE_ID, clusterNodes2.stream().map(ClusterNode::name).collect(Collectors.toSet()), keyValueStorage);
}
@Test
@@ -415,7 +367,7 @@ public class DistributionZoneManagerScaleUpTest {
mockVaultZonesLogicalTopologyKey(clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
- mockCmgLocalNodes(1L, clusterNodes);
+ mockCmgLocalNodes();
distributionZoneManager.start();
@@ -425,22 +377,59 @@ public class DistributionZoneManagerScaleUpTest {
Set<ClusterNode> clusterNodes2 = Set.of(node1, node2);
- mockCmgLocalNodes(1L, clusterNodes2);
-
- assertLogicalTopology(clusterNodes2);
+ assertLogicalTopology(clusterNodes2, keyValueStorage);
distributionZoneManager.createZone(
- new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleUp(1).build()
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleUp(0).build()
).get();
- assertDataNodesForZone(1, clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
+ assertDataNodesForZone(1, clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()), keyValueStorage);
+
+ assertZoneScaleUpChangeTriggerKey(1, 1, keyValueStorage);
- assertZoneScaleUpChangeTriggerKey(1, 1);
+ distributionZoneManager.dropZone(ZONE_NAME).get();
watchListenerOnUpdate(topology.getLogicalTopology().nodes().stream().map(ClusterNode::name).collect(Collectors.toSet()), 2);
+ assertNotEqualsDataNodesForZone(1, clusterNodes2.stream().map(ClusterNode::name).collect(Collectors.toSet()));
+ }
+
+ @Test
+ void testDropZoneDoNotPropagateDataNodesAfterScaleDown() throws Exception {
+ ClusterNode node1 = new ClusterNode("1", "name1", new NetworkAddress("localhost", 123));
+
+ ClusterNode node2 = new ClusterNode("2", "name2", new NetworkAddress("localhost", 123));
+
+ topology.putNode(node1);
+
+ topology.putNode(node2);
+
+ Set<ClusterNode> clusterNodes = Set.of(node1, node2);
+
+ mockVaultZonesLogicalTopologyKey(clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
+
+ mockCmgLocalNodes();
+
+ distributionZoneManager.start();
+
+ topology.removeNodes(Set.of(node2));
+
+ Set<ClusterNode> clusterNodes2 = Set.of(node1);
+
+ assertLogicalTopology(clusterNodes2, keyValueStorage);
+
+ distributionZoneManager.createZone(
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleDown(0).build()
+ ).get();
+
+ assertDataNodesForZone(1, clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()), keyValueStorage);
+
+ assertZoneScaleDownChangeTriggerKey(1, 1, keyValueStorage);
+
distributionZoneManager.dropZone(ZONE_NAME).get();
+ watchListenerOnUpdate(topology.getLogicalTopology().nodes().stream().map(ClusterNode::name).collect(Collectors.toSet()), 2);
+
assertNotEqualsDataNodesForZone(1, clusterNodes2.stream().map(ClusterNode::name).collect(Collectors.toSet()));
}
@@ -460,7 +449,7 @@ public class DistributionZoneManagerScaleUpTest {
distributionZoneManager.scheduleTimers(
zoneView,
- Set.of("B"),
+ Set.of("D"),
Set.of(),
2,
(zoneId, revision) -> {
@@ -471,10 +460,11 @@ public class DistributionZoneManagerScaleUpTest {
}
return testSaveDataNodesOnScaleUp(zoneId, revision).thenRun(out1::countDown);
- }
+ },
+ (t1, t2) -> null
);
- // Assert that first task was run and event about adding node "B" with revision 2 was added
+ // Assert that first task was run and event about adding node "D" with revision 2 was added
// to the topologyAugmentationMap of the zone.
assertTrue(
waitForCondition(
@@ -485,7 +475,7 @@ public class DistributionZoneManagerScaleUpTest {
distributionZoneManager.scheduleTimers(
zoneView,
- Set.of("C"),
+ Set.of("E"),
Set.of(),
3,
(zoneId, revision) -> {
@@ -502,10 +492,104 @@ public class DistributionZoneManagerScaleUpTest {
fail();
}
});
+ },
+ (t1, t2) -> null
+ );
+
+ // Assert that second task was run and event about adding node "E" with revision 3 was added
+ // to the topologyAugmentationMap of the zone.
+ assertTrue(
+ waitForCondition(
+ () -> distributionZoneManager.zonesTimers().get(1).topologyAugmentationMap().containsKey(3L),
+ 1000
+ )
+ );
+
+
+ //Second task is propagating data nodes first.
+ in2.countDown();
+
+ assertZoneScaleUpChangeTriggerKey(3, 1, keyValueStorage);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "C", "D", "E"), keyValueStorage);
+
+ out2.countDown();
+
+ in1.countDown();
+
+ //Waiting for the first scheduler ends it work.
+ out1.countDown();
+
+ // Assert that nothing has been changed.
+ assertZoneScaleUpChangeTriggerKey(3, 1, keyValueStorage);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "C", "D", "E"), keyValueStorage);
+ }
+
+ @Test
+ void testTwoScaleDownTimersSecondTimerRunFirst() throws Exception {
+ preparePrerequisites();
+
+ NamedConfigurationTree<DistributionZoneConfiguration, DistributionZoneView, DistributionZoneChange> zones =
+ zonesConfiguration.distributionZones();
+
+ DistributionZoneView zoneView = zones.value().get(0);
+
+ CountDownLatch in1 = new CountDownLatch(1);
+ CountDownLatch in2 = new CountDownLatch(1);
+ CountDownLatch out1 = new CountDownLatch(1);
+ CountDownLatch out2 = new CountDownLatch(1);
+
+ distributionZoneManager.scheduleTimers(
+ zoneView,
+ Set.of(),
+ Set.of("B"),
+ 2,
+ (t1, t2) -> null,
+ (zoneId, revision) -> {
+ try {
+ in1.await();
+ } catch (InterruptedException e) {
+ fail();
+ }
+
+ return testSaveDataNodesOnScaleDown(zoneId, revision).thenRun(out1::countDown);
+ }
+ );
+
+ // Assert that first task was run and event about removing node "B" with revision 2 was added
+ // to the topologyAugmentationMap of the zone.
+ assertTrue(
+ waitForCondition(
+ () -> distributionZoneManager.zonesTimers().get(1).topologyAugmentationMap().containsKey(2L),
+ 1000
+ )
+ );
+
+ distributionZoneManager.scheduleTimers(
+ zoneView,
+ Set.of(),
+ Set.of("C"),
+ 3,
+ (t1, t2) -> null,
+ (zoneId, revision) -> {
+ try {
+ in2.await();
+ } catch (InterruptedException e) {
+ fail();
+ }
+
+ return testSaveDataNodesOnScaleDown(zoneId, revision).thenRun(() -> {
+ try {
+ out2.await();
+ } catch (InterruptedException e) {
+ fail();
+ }
+ });
}
);
- // Assert that second task was run and event about adding node "C" with revision 3 was added
+ // Assert that second task was run and event about removing node "C" with revision 3 was added
// to the topologyAugmentationMap of the zone.
assertTrue(
waitForCondition(
@@ -518,9 +602,9 @@ public class DistributionZoneManagerScaleUpTest {
//Second task is propagating data nodes first.
in2.countDown();
- assertZoneScaleUpChangeTriggerKey(3, 1);
+ assertZoneScaleDownChangeTriggerKey(3, 1, keyValueStorage);
- assertDataNodesForZone(1, Set.of("A", "B", "C"));
+ assertDataNodesForZone(1, Set.of("A"), keyValueStorage);
out2.countDown();
@@ -530,9 +614,9 @@ public class DistributionZoneManagerScaleUpTest {
out1.countDown();
// Assert that nothing has been changed.
- assertZoneScaleUpChangeTriggerKey(3, 1);
+ assertZoneScaleDownChangeTriggerKey(3, 1, keyValueStorage);
- assertDataNodesForZone(1, Set.of("A", "B", "C"));
+ assertDataNodesForZone(1, Set.of("A"), keyValueStorage);
}
@Test
@@ -550,7 +634,7 @@ public class DistributionZoneManagerScaleUpTest {
distributionZoneManager.scheduleTimers(
zoneView,
- Set.of("B"),
+ Set.of("D"),
Set.of(),
2,
(zoneId, revision) -> {
@@ -563,7 +647,8 @@ public class DistributionZoneManagerScaleUpTest {
fail();
}
});
- }
+ },
+ (t1, t2) -> null
);
// Waiting for the first task to be run. We have to do that to be sure that watch events,
@@ -572,7 +657,7 @@ public class DistributionZoneManagerScaleUpTest {
distributionZoneManager.scheduleTimers(
zoneView,
- Set.of("C"),
+ Set.of("E"),
Set.of(),
3,
(zoneId, revision) -> {
@@ -583,10 +668,11 @@ public class DistributionZoneManagerScaleUpTest {
}
return testSaveDataNodesOnScaleUp(zoneId, revision);
- }
+ },
+ (t1, t2) -> null
);
- // Assert that second task was run and event about adding node "C" with revision 3 was added
+ // Assert that second task was run and event about adding node "E" with revision 3 was added
// to the topologyAugmentationMap of the zone.
assertTrue(
waitForCondition(
@@ -595,68 +681,878 @@ public class DistributionZoneManagerScaleUpTest {
)
);
- assertZoneScaleUpChangeTriggerKey(2, 1);
+ assertZoneScaleUpChangeTriggerKey(2, 1, keyValueStorage);
- assertDataNodesForZone(1, Set.of("A", "B"));
+ assertDataNodesForZone(1, Set.of("A", "B", "C", "D"), keyValueStorage);
- // Second task is run and we await that data nodes will be changed from ["A", "B"] to ["A", "B", "C"]
+ // Second task is run and we await that data nodes will be changed from ["A", "B", "C", "D"] to ["A", "B", "C", "D", "E"]
in2.countDown();
- assertZoneScaleUpChangeTriggerKey(3, 1);
+ assertZoneScaleUpChangeTriggerKey(3, 1, keyValueStorage);
- assertDataNodesForZone(1, Set.of("A", "B", "C"));
+ assertDataNodesForZone(1, Set.of("A", "B", "C", "D", "E"), keyValueStorage);
out1.countDown();
}
+ @Test
+ void testTwoScaleDownTimersFirstTimerRunFirst() throws Exception {
+ preparePrerequisites();
+ NamedConfigurationTree<DistributionZoneConfiguration, DistributionZoneView, DistributionZoneChange> zones =
+ zonesConfiguration.distributionZones();
- /**
- * Creates a zone with the auto adjust scale up trigger equals to 0 and the data nodes equals ["A"].
- *
- * @throws Exception when something goes wrong.
- */
- private void preparePrerequisites() throws Exception {
- ClusterNode node1 = new ClusterNode("1", "A", new NetworkAddress("localhost", 123));
-
- topology.putNode(node1);
-
- Set<ClusterNode> clusterNodes = Set.of(node1);
-
- mockVaultZonesLogicalTopologyKey(clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
+ DistributionZoneView zoneView = zones.value().get(0);
- mockCmgLocalNodes(1L, clusterNodes);
+ CountDownLatch in1 = new CountDownLatch(1);
+ CountDownLatch in2 = new CountDownLatch(1);
+ CountDownLatch out1 = new CountDownLatch(1);
- distributionZoneManager.start();
+ distributionZoneManager.scheduleTimers(
+ zoneView,
+ Set.of(),
+ Set.of("B"),
+ 2,
+ (t1, t2) -> null,
+ (zoneId, revision) -> {
+ in1.countDown();
- distributionZoneManager.createZone(
- new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleUp(0).build()
- ).get();
+ return testSaveDataNodesOnScaleDown(zoneId, revision).thenRun(() -> {
+ try {
+ out1.await();
+ } catch (InterruptedException e) {
+ fail();
+ }
+ });
+ }
+ );
- assertDataNodesForZone(1, clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
+ // Waiting for the first task to be run. We have to do that to be sure that watch events,
+ // which we try to emulate, are handled sequentially.
+ in1.await();
- assertZoneScaleUpChangeTriggerKey(1, 1);
- }
+ distributionZoneManager.scheduleTimers(
+ zoneView,
+ Set.of(),
+ Set.of("C"),
+ 3,
+ (t1, t2) -> null,
+ (zoneId, revision) -> {
+ try {
+ in2.await();
+ } catch (InterruptedException e) {
+ fail();
+ }
- private CompletableFuture<Void> testSaveDataNodesOnScaleUp(int zoneId, long revision) {
- return distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(zoneId, revision);
+ return testSaveDataNodesOnScaleDown(zoneId, revision);
+ }
+ );
+
+ // Assert that second task was run and event about removing node "C" with revision 3 was added
+ // to the topologyAugmentationMap of the zone.
+ assertTrue(
+ waitForCondition(
+ () -> distributionZoneManager.zonesTimers().get(1).topologyAugmentationMap().containsKey(3L),
+ 1000
+ )
+ );
+
+ assertZoneScaleDownChangeTriggerKey(2, 1, keyValueStorage);
+
+ assertDataNodesForZone(1, Set.of("A", "C"), keyValueStorage);
+
+ // Second task is run and we await that data nodes will be changed from ["A", "C"] to ["A"]
+ in2.countDown();
+
+ assertZoneScaleDownChangeTriggerKey(3, 1, keyValueStorage);
+
+ assertDataNodesForZone(1, Set.of("A"), keyValueStorage);
+
+ out1.countDown();
}
- private void assertDataNodesForZone(int zoneId, @Nullable Set<String> clusterNodes) throws InterruptedException {
- assertTrue(waitForCondition(
- () -> {
- byte[] dataNodes = keyValueStorage.get(zoneDataNodesKey(zoneId).bytes()).value();
+ @Test
+ void testEmptyDataNodesOnStart() throws Exception {
+ mockVaultZonesLogicalTopologyKey(Set.of());
- if (dataNodes == null) {
- return clusterNodes == null;
- }
+ mockCmgLocalNodes();
- Set<String> res = ByteUtils.fromBytes(dataNodes);
+ distributionZoneManager.start();
- return res.equals(clusterNodes);
- },
- 2000
- ));
+ assertLogicalTopology(Set.of(), keyValueStorage);
+
+ distributionZoneManager.createZone(
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleUp(0).build()
+ ).get();
+
+ assertDataNodesForZone(1, Set.of(), keyValueStorage);
+
+ assertZoneScaleDownChangeTriggerKey(1, 1, keyValueStorage);
+
+ ClusterNode node1 = new ClusterNode("1", "name1", new NetworkAddress("localhost", 123));
+
+ topology.putNode(node1);
+
+ assertLogicalTopology(Set.of(node1), keyValueStorage);
+
+ watchListenerOnUpdate(topology.getLogicalTopology().nodes().stream().map(ClusterNode::name).collect(Collectors.toSet()), 2);
+
+ assertDataNodesForZone(1, Set.of(node1.name()), keyValueStorage);
+ }
+
+ @Test
+ void testUpdateZoneScaleUpTriggersDataNodePropagation() throws Exception {
+ mockVaultZonesLogicalTopologyKey(Set.of());
+
+ mockCmgLocalNodes();
+
+ distributionZoneManager.start();
+
+ assertLogicalTopology(Set.of(), keyValueStorage);
+
+ distributionZoneManager.createZone(
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleUp(100).build()
+ ).get();
+
+ assertDataNodesForZone(1, Set.of(), keyValueStorage);
+
+ assertZoneScaleDownChangeTriggerKey(1, 1, keyValueStorage);
+
+ ClusterNode node1 = new ClusterNode("1", "name1", new NetworkAddress("localhost", 123));
+
+ topology.putNode(node1);
+
+ assertLogicalTopology(Set.of(node1), keyValueStorage);
+
+ watchListenerOnUpdate(topology.getLogicalTopology().nodes().stream().map(ClusterNode::name).collect(Collectors.toSet()), 2);
+
+ assertDataNodesForZone(1, Set.of(), keyValueStorage);
+
+ distributionZoneManager.alterZone(
+ ZONE_NAME,
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleUp(0).build()
+ ).get();
+
+ assertDataNodesForZone(1, Set.of(node1.name()), keyValueStorage);
+ }
+
+ @Test
+ void testUpdateZoneScaleDownTriggersDataNodePropagation() throws Exception {
+ ClusterNode node1 = new ClusterNode("1", "name1", new NetworkAddress("localhost", 123));
+
+ topology.putNode(node1);
+
+ mockVaultZonesLogicalTopologyKey(Set.of(node1.name()));
+
+ mockCmgLocalNodes();
+
+ distributionZoneManager.start();
+
+ assertLogicalTopology(Set.of(node1), keyValueStorage);
+
+ distributionZoneManager.createZone(
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleDown(100).build()
+ ).get();
+
+ assertDataNodesForZone(1, Set.of(node1.name()), keyValueStorage);
+
+ assertZoneScaleDownChangeTriggerKey(1, 1, keyValueStorage);
+
+ topology.removeNodes(Set.of(node1));
+
+ watchListenerOnUpdate(topology.getLogicalTopology().nodes().stream().map(ClusterNode::name).collect(Collectors.toSet()), 2);
+
+ assertDataNodesForZone(1, Set.of(node1.name()), keyValueStorage);
+
+ distributionZoneManager.alterZone(
+ ZONE_NAME,
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleDown(0).build()
+ ).get();
+
+ assertDataNodesForZone(1, Set.of(), keyValueStorage);
+ }
+
+ @Test
+ void testCleanUpAfterSchedulers() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 3);
+ zoneState.nodesToAddToDataNodes(Set.of("E"), 4);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 7);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 3);
+
+ assertTrue(zoneState.topologyAugmentationMap().containsKey(3L));
+ assertTrue(zoneState.topologyAugmentationMap().containsKey(4L));
+ assertTrue(zoneState.topologyAugmentationMap().containsKey(7L));
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 4);
+
+ assertTrue(zoneState.topologyAugmentationMap().containsKey(3L));
+ assertTrue(zoneState.topologyAugmentationMap().containsKey(4L));
+ assertTrue(zoneState.topologyAugmentationMap().containsKey(7L));
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 7);
+
+ assertFalse(zoneState.topologyAugmentationMap().containsKey(3L));
+ assertFalse(zoneState.topologyAugmentationMap().containsKey(4L));
+ assertTrue(zoneState.topologyAugmentationMap().containsKey(7L));
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 15);
+
+ assertTrue(zoneState.topologyAugmentationMap().isEmpty());
+ assertDataNodesForZone(1, Set.of("A", "B", "D", "E"), keyValueStorage);
+ }
+
+ @Test
+ void testScaleUpSetToMaxInt() throws Exception {
+ mockVaultZonesLogicalTopologyKey(Set.of());
+
+ mockCmgLocalNodes();
+
+ distributionZoneManager.start();
+
+ assertLogicalTopology(Set.of(), keyValueStorage);
+
+ distributionZoneManager.createZone(
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleUp(100).build()
+ ).get();
+
+ assertDataNodesForZone(1, Set.of(), keyValueStorage);
+
+ assertZoneScaleDownChangeTriggerKey(1, 1, keyValueStorage);
+
+ ClusterNode node1 = new ClusterNode("1", "name1", new NetworkAddress("localhost", 123));
+
+ topology.putNode(node1);
+
+ assertLogicalTopology(Set.of(node1), keyValueStorage);
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ assertNull(zoneState.scaleUpTask());
+
+ watchListenerOnUpdate(topology.getLogicalTopology().nodes().stream().map(ClusterNode::name).collect(Collectors.toSet()), 2);
+
+ assertTrue(waitForCondition(() -> zoneState.scaleUpTask() != null, 1000L));
+
+ distributionZoneManager.alterZone(
+ ZONE_NAME,
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleUp(INFINITE_TIMER_VALUE).build()
+ ).get();
+
+ assertTrue(waitForCondition(() -> zoneState.scaleUpTask().isCancelled(), 1000L));
+ }
+
+ @Test
+ void testScaleDownSetToMaxInt() throws Exception {
+ ClusterNode node1 = new ClusterNode("1", "name1", new NetworkAddress("localhost", 123));
+
+ topology.putNode(node1);
+
+ mockVaultZonesLogicalTopologyKey(Set.of(node1.name()));
+
+ mockCmgLocalNodes();
+
+ distributionZoneManager.start();
+
+ assertLogicalTopology(Set.of(node1), keyValueStorage);
+
+ distributionZoneManager.createZone(
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleDown(100).build()
+ ).get();
+
+ assertDataNodesForZone(1, Set.of(node1.name()), keyValueStorage);
+
+ assertZoneScaleDownChangeTriggerKey(1, 1, keyValueStorage);
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ assertNull(zoneState.scaleDownTask());
+
+ topology.removeNodes(Set.of(node1));
+
+ watchListenerOnUpdate(topology.getLogicalTopology().nodes().stream().map(ClusterNode::name).collect(Collectors.toSet()), 2);
+
+ assertTrue(waitForCondition(() -> zoneState.scaleDownTask() != null, 1000L));
+
+ distributionZoneManager.alterZone(
+ ZONE_NAME,
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleDown(INFINITE_TIMER_VALUE).build()
+ ).get();
+
+ assertTrue(waitForCondition(() -> zoneState.scaleDownTask().isCancelled(), 1000L));
+ }
+
+ @Test
+ void testScaleUpDidNotChangeDataNodesWhenTriggerKeyWasConcurrentlyChanged() throws Exception {
+ mockVaultZonesLogicalTopologyKey(Set.of());
+
+ mockCmgLocalNodes();
+
+ distributionZoneManager.start();
+
+ assertLogicalTopology(Set.of(), keyValueStorage);
+
+ distributionZoneManager.createZone(
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleUp(0).build()
+ ).get();
+
+ assertDataNodesForZone(1, Set.of(), keyValueStorage);
+
+ assertZoneScaleDownChangeTriggerKey(1, 1, keyValueStorage);
+
+ ClusterNode node1 = new ClusterNode("1", "name1", new NetworkAddress("localhost", 123));
+
+ topology.putNode(node1);
+
+ assertLogicalTopology(Set.of(node1), keyValueStorage);
+
+ MetaStorageCommandsFactory commandsFactory = new MetaStorageCommandsFactory();
+
+ lenient().doAnswer(invocationClose -> {
+ // First invoke of data nodes propagation on scale up do not pass because of trigger key revision condition.
+ // We had scaleUpChangeTriggerKey = 1 from ms, but after that it was changed to 100, so on the next call
+ // invoke won't be even triggered because event become stale.
+ keyValueStorage.put(zoneScaleUpChangeTriggerKey(1).bytes(), longToBytes(100));
+
+ Iif iif = invocationClose.getArgument(0);
+
+ MultiInvokeCommand multiInvokeCommand = commandsFactory.multiInvokeCommand().iif(iif).build();
+
+ return metaStorageService.run(multiInvokeCommand);
+ }).when(metaStorageManager).invoke(any());
+
+ watchListenerOnUpdate(topology.getLogicalTopology().nodes().stream().map(ClusterNode::name).collect(Collectors.toSet()), 2);
+
+ assertZoneScaleUpChangeTriggerKey(100, 1, keyValueStorage);
+
+ assertDataNodesForZone(1, Set.of(), keyValueStorage);
+ }
+
+ @Test
+ void testScaleDownDidNotChangeDataNodesWhenTriggerKeyWasConcurrentlyChanged() throws Exception {
+ ClusterNode node1 = new ClusterNode("1", "name1", new NetworkAddress("localhost", 123));
+
+ topology.putNode(node1);
+
+ mockVaultZonesLogicalTopologyKey(Set.of(node1.name()));
+
+ mockCmgLocalNodes();
+
+ distributionZoneManager.start();
+
+ assertLogicalTopology(Set.of(node1), keyValueStorage);
+
+ distributionZoneManager.createZone(
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME).dataNodesAutoAdjustScaleDown(0).build()
+ ).get();
+
+ assertDataNodesForZone(1, Set.of(node1.name()), keyValueStorage);
+
+ assertZoneScaleDownChangeTriggerKey(1, 1, keyValueStorage);
+
+ topology.removeNodes(Set.of(node1));
+
+ MetaStorageCommandsFactory commandsFactory = new MetaStorageCommandsFactory();
+
+ lenient().doAnswer(invocationClose -> {
+ // First invoke of data nodes propagation on scale down do not pass because of trigger key revision condition.
+ // We had scaleDownChangeTriggerKey = 1 from ms, but after that it was changed to 100, so on the next call
+ // invoke won't be even triggered because event become stale.
+ keyValueStorage.put(zoneScaleDownChangeTriggerKey(1).bytes(), longToBytes(100));
+
+ Iif iif = invocationClose.getArgument(0);
+
+ MultiInvokeCommand multiInvokeCommand = commandsFactory.multiInvokeCommand().iif(iif).build();
+
+ return metaStorageService.run(multiInvokeCommand);
+ }).when(metaStorageManager).invoke(any());
+
+ watchListenerOnUpdate(topology.getLogicalTopology().nodes().stream().map(ClusterNode::name).collect(Collectors.toSet()), 2);
+
+ assertDataNodesForZone(1, Set.of(node1.name()), keyValueStorage);
+
+ assertZoneScaleDownChangeTriggerKey(100, 1, keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios1_1() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 3);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 7);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 3);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 7);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "D"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios1_2() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 3);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 7);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 7);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 3);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "D"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios1_3() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 3);
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 7);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 3);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 7);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "D"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios1_4() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 3);
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 7);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 7);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 3);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "D"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios2_1() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 3);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("D"), 7);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 3);
+ assertDataNodesForZone(1, Set.of("A", "B", "C", "D"), keyValueStorage);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 7);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "C"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios2_2() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 3);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("D"), 7);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 7);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 3);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "C"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios2_3() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 3);
+ zoneState.nodesToAddToDataNodes(Set.of("C"), 7);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 3);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 7);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "C"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios2_4() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 3);
+ zoneState.nodesToAddToDataNodes(Set.of("C"), 7);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 7);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 3);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "C"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios3_1() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 3);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("D"), 7);
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 9);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 9);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 7);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "C", "D"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios3_2() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 3);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("D"), 7);
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 9);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 7);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 9);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "C", "D"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios3_3() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 3);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("D"), 7);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 7);
+
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 9);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 9);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "C", "D"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios3_4() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 3);
+ zoneState.nodesToAddToDataNodes(Set.of("C"), 7);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 9);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 9);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 7);
+
+ assertDataNodesForZone(1, Set.of("A", "B"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios3_5() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 3);
+ zoneState.nodesToAddToDataNodes(Set.of("C"), 7);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 9);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 7);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 9);
+
+ assertDataNodesForZone(1, Set.of("A", "B"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios3_6() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 3);
+ zoneState.nodesToAddToDataNodes(Set.of("C"), 7);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 7);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 9);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 9);
+
+ assertDataNodesForZone(1, Set.of("A", "B"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios4_1() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 3);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("D"), 7);
+ zoneState.nodesToAddToDataNodes(Set.of("E"), 9);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 9);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 7);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "C", "E"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios4_2() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 3);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("D"), 7);
+ zoneState.nodesToAddToDataNodes(Set.of("E"), 9);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 7);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 9);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "C", "E"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios4_3() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 3);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("D"), 7);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 7);
+
+ zoneState.nodesToAddToDataNodes(Set.of("E"), 9);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 9);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "C", "E"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios4_4() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 3);
+ zoneState.nodesToAddToDataNodes(Set.of("C"), 7);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("B"), 9);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 9);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 7);
+
+ assertDataNodesForZone(1, Set.of("A", "C"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios4_5() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 3);
+ zoneState.nodesToAddToDataNodes(Set.of("C"), 7);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("B"), 9);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 7);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 9);
+
+ assertDataNodesForZone(1, Set.of("A", "C"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios4_6() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 3);
+ zoneState.nodesToAddToDataNodes(Set.of("C"), 7);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 7);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("B"), 9);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 9);
+
+ assertDataNodesForZone(1, Set.of("A", "C"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios5_1() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 3);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 7);
+ zoneState.nodesToAddToDataNodes(Set.of("C"), 9);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 9);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 7);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "C", "D"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios5_2() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 3);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 7);
+ zoneState.nodesToAddToDataNodes(Set.of("C"), 9);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 7);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 9);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "C", "D"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios5_3() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 3);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 7);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 7);
+
+ zoneState.nodesToAddToDataNodes(Set.of("C"), 9);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 9);
+
+ assertDataNodesForZone(1, Set.of("A", "B", "C", "D"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios5_4() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 3);
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 7);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("D"), 9);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 9);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 7);
+
+ assertDataNodesForZone(1, Set.of("A", "B"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios5_5() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 3);
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 7);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("D"), 9);
+
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 7);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 9);
+
+ assertDataNodesForZone(1, Set.of("A", "B"), keyValueStorage);
+ }
+
+ @Test
+ void testVariousScaleUpScaleDownScenarios5_6() throws Exception {
+ preparePrerequisites();
+
+ ZoneState zoneState = distributionZoneManager.zonesTimers().get(1);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C"), 3);
+ zoneState.nodesToAddToDataNodes(Set.of("D"), 7);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(1, 7);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("D"), 9);
+ distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(1, 9);
+
+ assertDataNodesForZone(1, Set.of("A", "B"), keyValueStorage);
+ }
+
+ @Test
+ void testZoneStateAddRemoveNodesPreservesDuplicationsOfNodes() {
+ ZoneState zoneState = new ZoneState(new ScheduledThreadPoolExecutor(1));
+
+ zoneState.nodesToAddToDataNodes(Set.of("A", "B"), 1);
+ zoneState.nodesToAddToDataNodes(Set.of("A", "B"), 2);
+
+ List<String> nodes = zoneState.nodesToBeAddedToDataNodes(0, 2);
+
+ Collections.sort(nodes);
+ assertEquals(List.of("A", "A", "B", "B"), nodes);
+
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C", "D"), 3);
+ zoneState.nodesToRemoveFromDataNodes(Set.of("C", "D"), 4);
+
+ nodes = zoneState.nodesToBeRemovedFromDataNodes(2, 4);
+
+ Collections.sort(nodes);
+
+ assertEquals(List.of("C", "C", "D", "D"), nodes);
+ }
+
+ /**
+ * Creates a zone with the auto adjust scale up scale down trigger equals to 0 and the data nodes equals ["A", "B", "C"].
+ *
+ * @throws Exception when something goes wrong.
+ */
+ private void preparePrerequisites() throws Exception {
+ ClusterNode node1 = new ClusterNode("1", "A", new NetworkAddress("localhost", 123));
+ ClusterNode node2 = new ClusterNode("2", "B", new NetworkAddress("localhost", 123));
+ ClusterNode node3 = new ClusterNode("3", "C", new NetworkAddress("localhost", 123));
+
+ topology.putNode(node1);
+ topology.putNode(node2);
+ topology.putNode(node3);
+
+ Set<ClusterNode> clusterNodes = Set.of(node1, node2, node3);
+
+ mockVaultZonesLogicalTopologyKey(clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
+
+ mockCmgLocalNodes();
+
+ distributionZoneManager.start();
+
+ distributionZoneManager.createZone(
+ new DistributionZoneConfigurationParameters.Builder(ZONE_NAME)
+ .dataNodesAutoAdjustScaleUp(0)
+ .dataNodesAutoAdjustScaleDown(0)
+ .build()
+ ).get();
+
+ assertDataNodesForZone(1, clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()), keyValueStorage);
+
+ assertZoneScaleUpChangeTriggerKey(1, 1, keyValueStorage);
+ assertZoneScaleDownChangeTriggerKey(1, 1, keyValueStorage);
+ }
+
+ private CompletableFuture<Void> testSaveDataNodesOnScaleUp(int zoneId, long revision) {
+ return distributionZoneManager.saveDataNodesToMetaStorageOnScaleUp(zoneId, revision);
+ }
+
+ private CompletableFuture<Void> testSaveDataNodesOnScaleDown(int zoneId, long revision) {
+ return distributionZoneManager.saveDataNodesToMetaStorageOnScaleDown(zoneId, revision);
}
private void assertNotEqualsDataNodesForZone(int zoneId, @Nullable Set<String> clusterNodes) throws InterruptedException {
@@ -668,31 +1564,14 @@ public class DistributionZoneManagerScaleUpTest {
return clusterNodes == null;
}
- Set<String> res = ByteUtils.fromBytes(dataNodes);
+ Set<String> res = DistributionZonesUtil.dataNodes(ByteUtils.fromBytes(dataNodes));
return res.equals(clusterNodes);
},
- 2000
+ 1000
));
}
- private void assertZoneScaleUpChangeTriggerKey(int revision, int zoneId) throws InterruptedException {
- assertTrue(
- waitForCondition(
- () -> ByteUtils.bytesToLong(keyValueStorage.get(zoneScaleUpChangeTriggerKey(zoneId).bytes()).value()) == revision,
- 2000
- )
- );
- }
-
- private void assertZonesChangeTriggerKey(int revision, int zoneId) throws InterruptedException {
- assertTrue(
- waitForCondition(
- () -> ByteUtils.bytesToLong(keyValueStorage.get(zonesChangeTriggerKey(zoneId).bytes()).value()) == revision, 1000
- )
- );
- }
-
private void mockVaultAppliedRevision(long revision) {
when(metaStorageManager.appliedRevision()).thenReturn(revision);
}
@@ -709,20 +1588,8 @@ public class DistributionZoneManagerScaleUpTest {
watchListener.onUpdate(evt);
}
- private LogicalTopologySnapshot mockCmgLocalNodes(long version, Set<ClusterNode> clusterNodes) {
- LogicalTopologySnapshot logicalTopologySnapshot = new LogicalTopologySnapshot(version, clusterNodes);
-
- when(cmgManager.logicalTopology()).thenReturn(completedFuture(logicalTopologySnapshot));
-
- return logicalTopologySnapshot;
- }
-
- private void assertLogicalTopology(@Nullable Set<ClusterNode> clusterNodes) throws InterruptedException {
- byte[] nodes = clusterNodes == null
- ? null
- : toBytes(clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
-
- assertTrue(waitForCondition(() -> Arrays.equals(keyValueStorage.get(zonesLogicalTopologyKey().bytes()).value(), nodes), 1000));
+ private void mockCmgLocalNodes() {
+ when(cmgManager.logicalTopology()).thenReturn(completedFuture(topology.getLogicalTopology()));
}
private void mockVaultZonesLogicalTopologyKey(Set<String> nodes) {
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerTest.java
index 3e78a95010..fd51c2806d 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerTest.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.distributionzones;
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.INFINITE_TIMER_VALUE;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -124,8 +125,8 @@ class DistributionZoneManagerTest extends IgniteAbstractTest {
assertNotNull(zone1, "Zone was not created.");
assertEquals(ZONE_NAME, zone1.name().value(), "Zone name is wrong.");
- assertEquals(Integer.MAX_VALUE, zone1.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
- assertEquals(Integer.MAX_VALUE, zone1.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone1.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone1.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
assertEquals(100, zone1.dataNodesAutoAdjust().value(), "dataNodesAutoAdjust is wrong.");
}
@@ -143,8 +144,8 @@ class DistributionZoneManagerTest extends IgniteAbstractTest {
assertNotNull(zone1, "Zone was not created.");
assertEquals(ZONE_NAME, zone1.name().value(), "Zone name is wrong.");
assertEquals(100, zone1.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
- assertEquals(Integer.MAX_VALUE, zone1.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
- assertEquals(Integer.MAX_VALUE, zone1.dataNodesAutoAdjust().value(), "dataNodesAutoAdjust is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone1.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone1.dataNodesAutoAdjust().value(), "dataNodesAutoAdjust is wrong.");
distributionZoneManager.dropZone(ZONE_NAME).get(5, TimeUnit.SECONDS);
@@ -167,9 +168,9 @@ class DistributionZoneManagerTest extends IgniteAbstractTest {
assertNotNull(zone1, "Zone was not created.");
assertEquals(ZONE_NAME, zone1.name().value(), "Zone name is wrong.");
- assertEquals(Integer.MAX_VALUE, zone1.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone1.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
assertEquals(200, zone1.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
- assertEquals(Integer.MAX_VALUE, zone1.dataNodesAutoAdjust().value(), "dataNodesAutoAdjust is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone1.dataNodesAutoAdjust().value(), "dataNodesAutoAdjust is wrong.");
distributionZoneManager.dropZone(ZONE_NAME).get(5, TimeUnit.SECONDS);
@@ -253,14 +254,14 @@ class DistributionZoneManagerTest extends IgniteAbstractTest {
assertNotNull(zone, "Zone was not created.");
assertEquals(zoneName, zone.name().value(), "Zone name is wrong.");
- assertEquals(Integer.MAX_VALUE, zone.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
- assertEquals(Integer.MAX_VALUE, zone.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
assertEquals(100, zone.dataNodesAutoAdjust().value(), "dataNodesAutoAdjust is wrong.");
assertNotNull(zone, "Zone was not created.");
assertEquals(zoneName, zone.name().value(), "Zone name is wrong.");
- assertEquals(Integer.MAX_VALUE, zone.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
- assertEquals(Integer.MAX_VALUE, zone.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
assertEquals(100, zone.dataNodesAutoAdjust().value(), "dataNodesAutoAdjust is wrong.");
@@ -273,7 +274,7 @@ class DistributionZoneManagerTest extends IgniteAbstractTest {
assertNotNull(zone, "Zone was not created.");
assertEquals(200, zone.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
assertEquals(300, zone.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
- assertEquals(Integer.MAX_VALUE, zone.dataNodesAutoAdjust().value(), "dataNodesAutoAdjust is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone.dataNodesAutoAdjust().value(), "dataNodesAutoAdjust is wrong.");
distributionZoneManager.alterZone(zoneName, new DistributionZoneConfigurationParameters.Builder(zoneName)
@@ -285,7 +286,7 @@ class DistributionZoneManagerTest extends IgniteAbstractTest {
assertNotNull(zone, "Zone was not created.");
assertEquals(400, zone.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
assertEquals(300, zone.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
- assertEquals(Integer.MAX_VALUE, zone.dataNodesAutoAdjust().value(), "dataNodesAutoAdjust is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone.dataNodesAutoAdjust().value(), "dataNodesAutoAdjust is wrong.");
distributionZoneManager.alterZone(zoneName, new DistributionZoneConfigurationParameters.Builder(zoneName)
@@ -295,8 +296,8 @@ class DistributionZoneManagerTest extends IgniteAbstractTest {
zone = getZoneFromRegistry(zoneName);
assertNotNull(zone, "Zone was not created.");
- assertEquals(Integer.MAX_VALUE, zone.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
- assertEquals(Integer.MAX_VALUE, zone.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
assertEquals(500, zone.dataNodesAutoAdjust().value(), "dataNodesAutoAdjust is wrong.");
}
@@ -330,8 +331,8 @@ class DistributionZoneManagerTest extends IgniteAbstractTest {
assertNull(zone1, "Zone was not renamed.");
assertNotNull(zone2, "Zone was not renamed.");
assertEquals(NEW_ZONE_NAME, zone2.name().value(), "Zone was not renamed.");
- assertEquals(Integer.MAX_VALUE, zone2.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
- assertEquals(Integer.MAX_VALUE, zone2.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone2.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone2.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
assertEquals(100, zone2.dataNodesAutoAdjust().value(), "dataNodesAutoAdjust is wrong.");
}
@@ -356,8 +357,8 @@ class DistributionZoneManagerTest extends IgniteAbstractTest {
assertNull(zone1, "Zone was not renamed.");
assertNotNull(zone2, "Zone was not renamed.");
assertEquals(NEW_ZONE_NAME, zone2.name().value(), "Zone was not renamed.");
- assertEquals(Integer.MAX_VALUE, zone2.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
- assertEquals(Integer.MAX_VALUE, zone2.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone2.dataNodesAutoAdjustScaleUp().value(), "dataNodesAutoAdjustScaleUp is wrong.");
+ assertEquals(INFINITE_TIMER_VALUE, zone2.dataNodesAutoAdjustScaleDown().value(), "dataNodesAutoAdjustScaleDown is wrong.");
assertEquals(400, zone2.dataNodesAutoAdjust().value(), "dataNodesAutoAdjust is wrong.");
}
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerWatchListenerTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerWatchListenerTest.java
index c984406d5e..16413cf9ac 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerWatchListenerTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerWatchListenerTest.java
@@ -25,63 +25,43 @@ import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zoneScaleUpChangeTriggerKey;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesChangeTriggerKey;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesLogicalTopologyKey;
-import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
-import static org.apache.ignite.internal.util.ByteUtils.fromBytes;
+import static org.apache.ignite.internal.distributionzones.util.DistributionZonesTestUtil.assertDataNodesForZone;
+import static org.apache.ignite.internal.distributionzones.util.DistributionZonesTestUtil.mockMetaStorageListener;
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.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.Mockito.after;
import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-import java.io.Serializable;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.Set;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
-import java.util.stream.Collectors;
-import org.apache.ignite.configuration.ConfigurationValue;
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.cluster.management.topology.api.LogicalTopologySnapshot;
import org.apache.ignite.internal.configuration.ConfigurationManager;
import org.apache.ignite.internal.configuration.storage.TestConfigurationStorage;
-import org.apache.ignite.internal.distributionzones.configuration.DistributionZoneChange;
-import org.apache.ignite.internal.distributionzones.configuration.DistributionZoneConfiguration;
-import org.apache.ignite.internal.distributionzones.configuration.DistributionZoneView;
import org.apache.ignite.internal.distributionzones.configuration.DistributionZonesConfiguration;
-import org.apache.ignite.internal.distributionzones.configuration.DistributionZonesView;
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.command.GetAllCommand;
-import org.apache.ignite.internal.metastorage.command.MetaStorageCommandsFactory;
-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.server.SimpleInMemoryKeyValueStorage;
import org.apache.ignite.internal.metastorage.server.raft.MetaStorageListener;
-import org.apache.ignite.internal.raft.Command;
-import org.apache.ignite.internal.raft.ReadCommand;
-import org.apache.ignite.internal.raft.WriteCommand;
-import org.apache.ignite.internal.raft.service.CommandClosure;
import org.apache.ignite.internal.raft.service.RaftGroupService;
import org.apache.ignite.internal.schema.configuration.TableChange;
import org.apache.ignite.internal.schema.configuration.TableConfiguration;
@@ -90,33 +70,42 @@ 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.apache.ignite.lang.IgniteInternalException;
-import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
/**
* Tests distribution zones logical topology changes and reaction to that changes.
*/
//TODO: IGNITE-18564 Add tests with not default distribution zones, when distributionZones.change.trigger per zone will be created.
public class DistributionZoneManagerWatchListenerTest extends IgniteAbstractTest {
- private static final String ZONE_NAME_1 = "zone1";
-
- private VaultManager vaultMgr;
-
private DistributionZoneManager distributionZoneManager;
private SimpleInMemoryKeyValueStorage keyValueStorage;
private ConfigurationManager clusterCfgMgr;
+ @Mock
+ private LogicalTopologyServiceImpl logicalTopologyService;
+
+ private LogicalTopology topology;
+
+ private ClusterStateStorage clusterStateStorage;
+
+ private VaultManager vaultMgr;
+
+ private MetaStorageManager metaStorageManager;
+
private WatchListener watchListener;
private DistributionZonesConfiguration zonesConfiguration;
- private MetaStorageManager metaStorageManager;
+ @Mock
+ private ClusterManagementGroupManager cmgManager;
+
+ @Mock
+ RaftGroupService metaStorageService;
@BeforeEach
public void setUp() {
@@ -128,18 +117,18 @@ public class DistributionZoneManagerWatchListenerTest extends IgniteAbstractTest
List.of()
);
- zonesConfiguration = mock(DistributionZonesConfiguration.class);
+ zonesConfiguration = clusterCfgMgr.configurationRegistry()
+ .getConfiguration(DistributionZonesConfiguration.KEY);
metaStorageManager = mock(MetaStorageManager.class);
- LogicalTopologyServiceImpl logicalTopologyService = mock(LogicalTopologyServiceImpl.class);
+ cmgManager = mock(ClusterManagementGroupManager.class);
- LogicalTopologySnapshot topologySnapshot = mock(LogicalTopologySnapshot.class);
+ clusterStateStorage = new TestClusterStateStorage();
- when(topologySnapshot.version()).thenReturn(1L);
- when(topologySnapshot.nodes()).thenReturn(Collections.emptySet());
+ topology = new LogicalTopologyImpl(clusterStateStorage);
- when(logicalTopologyService.logicalTopologyOnLeader()).thenReturn(completedFuture(topologySnapshot));
+ LogicalTopologyServiceImpl logicalTopologyService = new LogicalTopologyServiceImpl(topology, cmgManager);
vaultMgr = mock(VaultManager.class);
@@ -177,132 +166,15 @@ public class DistributionZoneManagerWatchListenerTest extends IgniteAbstractTest
return null;
}).when(metaStorageManager).registerExactWatch(any(), any());
- mockDefaultZoneConfiguration();
- mockDefaultZoneView();
- mockEmptyZonesList();
-
AtomicLong raftIndex = new AtomicLong();
keyValueStorage = spy(new SimpleInMemoryKeyValueStorage("test"));
MetaStorageListener metaStorageListener = new MetaStorageListener(keyValueStorage);
- RaftGroupService metaStorageService = mock(RaftGroupService.class);
-
- // Delegate directly to listener.
- lenient().doAnswer(
- invocationClose -> {
- Command cmd = invocationClose.getArgument(0);
-
- long commandIndex = raftIndex.incrementAndGet();
-
- CompletableFuture<Serializable> res = new CompletableFuture<>();
-
- CommandClosure<WriteCommand> clo = new CommandClosure<>() {
- /** {@inheritDoc} */
- @Override
- public long index() {
- return commandIndex;
- }
-
- /** {@inheritDoc} */
- @Override
- public WriteCommand command() {
- return (WriteCommand) cmd;
- }
-
- /** {@inheritDoc} */
- @Override
- public void result(@Nullable Serializable r) {
- if (r instanceof Throwable) {
- res.completeExceptionally((Throwable) r);
- } else {
- res.complete(r);
- }
- }
- };
-
- try {
- metaStorageListener.onWrite(List.of(clo).iterator());
- } catch (Throwable e) {
- res.completeExceptionally(new IgniteInternalException(e));
- }
-
- return res;
- }
- ).when(metaStorageService).run(any(WriteCommand.class));
-
- lenient().doAnswer(
- invocationClose -> {
- Command cmd = invocationClose.getArgument(0);
-
- long commandIndex = raftIndex.incrementAndGet();
-
- CompletableFuture<Serializable> res = new CompletableFuture<>();
-
- CommandClosure<ReadCommand> clo = new CommandClosure<>() {
- /** {@inheritDoc} */
- @Override
- public long index() {
- return commandIndex;
- }
-
- /** {@inheritDoc} */
- @Override
- public ReadCommand command() {
- return (ReadCommand) cmd;
- }
-
- /** {@inheritDoc} */
- @Override
- public void result(@Nullable Serializable r) {
- if (r instanceof Throwable) {
- res.completeExceptionally((Throwable) r);
- } else {
- res.complete(r);
- }
- }
- };
-
- try {
- metaStorageListener.onRead(List.of(clo).iterator());
- } catch (Throwable e) {
- res.completeExceptionally(new IgniteInternalException(e));
- }
-
- return res;
- }
- ).when(metaStorageService).run(any(ReadCommand.class));
-
- MetaStorageCommandsFactory commandsFactory = new MetaStorageCommandsFactory();
-
- lenient().doAnswer(invocationClose -> {
- Iif iif = invocationClose.getArgument(0);
-
- MultiInvokeCommand multiInvokeCommand = commandsFactory.multiInvokeCommand().iif(iif).build();
-
- return metaStorageService.run(multiInvokeCommand);
- }).when(metaStorageManager).invoke(any());
-
- lenient().doAnswer(invocationClose -> {
- Set<ByteArray> keysSet = invocationClose.getArgument(0);
-
- GetAllCommand getAllCommand = commandsFactory.getAllCommand().keys(
- keysSet.stream().map(ByteArray::bytes).collect(Collectors.toList())
- ).revision(0).build();
-
- return metaStorageService.<List<Entry>>run(getAllCommand).thenApply(entries -> {
- Map<ByteArray, Entry> res = new HashMap<>();
-
- for (Entry e : entries) {
- ByteArray key = new ByteArray(e.key());
-
- res.put(key, new EntryImpl(key.bytes(), e.value(), e.revision(), e.updateCounter()));
- }
-
- return res;
- });
- }).when(metaStorageManager).getAll(any());
+ metaStorageService = mock(RaftGroupService.class);
+
+ mockMetaStorageListener(raftIndex, metaStorageListener, metaStorageService, metaStorageManager);
}
@AfterEach
@@ -317,20 +189,27 @@ public class DistributionZoneManagerWatchListenerTest extends IgniteAbstractTest
}
@Test
- void testDataNodesOfDefaultZoneUpdatedOnWatchListenerEvent() {
+ void testDataNodesOfDefaultZoneUpdatedOnWatchListenerEvent() throws Exception {
mockVaultZonesLogicalTopologyKey(Set.of());
+ mockCmgLocalNodes();
+
distributionZoneManager.start();
+ distributionZoneManager.alterZone(
+ DEFAULT_ZONE_NAME,
+ new DistributionZoneConfigurationParameters.Builder(DEFAULT_ZONE_NAME).dataNodesAutoAdjustScaleUp(0).build()
+ ).get();
+
//first event
Set<String> nodes = Set.of("node1", "node2");
watchListenerOnUpdate(nodes, 2);
- verify(keyValueStorage, timeout(1000).times(2)).invoke(any());
+ verify(keyValueStorage, timeout(1000).times(3)).invoke(any());
- checkDataNodesOfZone(DEFAULT_ZONE_ID, nodes);
+ assertDataNodesForZone(DEFAULT_ZONE_ID, nodes, keyValueStorage);
//second event
@@ -338,12 +217,12 @@ public class DistributionZoneManagerWatchListenerTest extends IgniteAbstractTest
watchListenerOnUpdate(nodes, 3);
- verify(keyValueStorage, timeout(1000).times(3)).invoke(any());
+ verify(keyValueStorage, timeout(1000).times(4)).invoke(any());
nodes = Set.of("node1", "node2", "node3");
// Scale up just adds node to data nodes
- checkDataNodesOfZone(DEFAULT_ZONE_ID, nodes);
+ assertDataNodesForZone(DEFAULT_ZONE_ID, nodes, keyValueStorage);
//third event
@@ -351,48 +230,46 @@ public class DistributionZoneManagerWatchListenerTest extends IgniteAbstractTest
watchListenerOnUpdate(nodes, 4);
- verify(keyValueStorage, timeout(1000).times(3)).invoke(any());
+ verify(keyValueStorage, timeout(1000).times(4)).invoke(any());
nodes = Set.of("node1", "node2", "node3");
// Scale up wasn't triggered
- checkDataNodesOfZone(DEFAULT_ZONE_ID, nodes);
- }
-
- private void assertDataNodesForZone(int zoneId, @Nullable Set<String> clusterNodes) throws InterruptedException {
- byte[] nodes = clusterNodes == null ? null : toBytes(clusterNodes);
-
- assertTrue(waitForCondition(() -> Arrays.equals(keyValueStorage.get(zoneDataNodesKey(zoneId).bytes()).value(), nodes), 1000));
+ assertDataNodesForZone(DEFAULT_ZONE_ID, nodes, keyValueStorage);
}
@Test
- void testStaleWatchEvent() {
+ void testStaleWatchEvent() throws Exception {
mockVaultZonesLogicalTopologyKey(Set.of());
- mockZones(mockZoneWithAutoAdjustScaleUp(100));
+ mockCmgLocalNodes();
distributionZoneManager.start();
- mockVaultAppliedRevision(1);
+ distributionZoneManager.alterZone(
+ DEFAULT_ZONE_NAME,
+ new DistributionZoneConfigurationParameters.Builder(DEFAULT_ZONE_NAME).dataNodesAutoAdjustScaleUp(0).build()
+ ).get();
long revision = 100;
- keyValueStorage.put(zoneScaleUpChangeTriggerKey(1).bytes(), longToBytes(revision));
+ keyValueStorage.put(zoneScaleUpChangeTriggerKey(DEFAULT_ZONE_ID).bytes(), longToBytes(revision));
Set<String> nodes = Set.of("node1", "node2");
watchListenerOnUpdate(nodes, revision);
+ // two invokes on start, and invoke for update scale up won't be triggered, because revision == zoneScaleUpChangeTriggerKey
verify(keyValueStorage, timeout(1000).times(2)).invoke(any());
- checkDataNodesOfZone(DEFAULT_ZONE_ID, Collections.emptySet());
+ assertDataNodesForZone(DEFAULT_ZONE_ID, Set.of(), keyValueStorage);
}
@Test
- void testStaleVaultRevisionOnZoneManagerStart() {
+ void testStaleVaultRevisionOnZoneManagerStart() throws InterruptedException {
long revision = 100;
- keyValueStorage.put(zonesChangeTriggerKey(0).bytes(), longToBytes(revision));
+ keyValueStorage.put(zonesChangeTriggerKey(DEFAULT_ZONE_ID).bytes(), longToBytes(revision));
Set<String> nodes = Set.of("node1", "node2");
@@ -400,171 +277,48 @@ public class DistributionZoneManagerWatchListenerTest extends IgniteAbstractTest
mockVaultAppliedRevision(revision);
+ mockCmgLocalNodes();
+
distributionZoneManager.start();
- verify(metaStorageManager, timeout(1000).times(1)).invoke(any());
+ verify(metaStorageManager, timeout(1000).times(2)).invoke(any());
- checkDataNodesOfZone(1, null);
+ assertDataNodesForZone(DEFAULT_ZONE_ID, null, keyValueStorage);
}
@Test
- void testDataNodesUpdatedOnZoneManagerStart() {
+ void testDataNodesUpdatedOnZoneManagerStart() throws InterruptedException {
mockVaultAppliedRevision(2);
Set<String> nodes = Set.of("node1", "node2");
mockVaultZonesLogicalTopologyKey(nodes);
+ mockCmgLocalNodes();
+
distributionZoneManager.start();
- verify(keyValueStorage, timeout(1000).times(1)).invoke(any());
+ verify(keyValueStorage, timeout(1000).times(2)).invoke(any());
- checkDataNodesOfZone(DEFAULT_ZONE_ID, nodes);
+ assertDataNodesForZone(DEFAULT_ZONE_ID, nodes, keyValueStorage);
}
@Test
void testLogicalTopologyIsNullOnZoneManagerStart1() {
- mockZones(mockZoneWithAutoAdjust());
-
- mockVaultAppliedRevision(2);
+ mockCmgLocalNodes();
when(vaultMgr.get(zonesLogicalTopologyKey()))
.thenReturn(completedFuture(new VaultEntry(zonesLogicalTopologyKey(), null)));
distributionZoneManager.start();
- verify(keyValueStorage, after(500).never()).invoke(any());
+ // 1 invoke because only invoke to zones logical topology happens
+ verify(keyValueStorage, timeout(1000).times(1)).invoke(any());
assertNull(keyValueStorage.get(zoneDataNodesKey(DEFAULT_ZONE_ID).bytes()).value());
assertNull(keyValueStorage.get(zoneDataNodesKey(1).bytes()).value());
}
- private void checkDataNodesOfZone(int zoneId, Set<String> nodes) {
- Entry entry = keyValueStorage.get(zoneDataNodesKey(zoneId).bytes());
-
- if (nodes == null) {
- assertNull(entry.value(), () -> "Actual node list: " + fromBytes(entry.value()));
- } else {
- assertNotNull(entry);
- assertNotNull(entry.value());
-
- Set<String> newDataNodes = fromBytes(entry.value());
-
- assertTrue(newDataNodes.containsAll(nodes));
- assertEquals(nodes.size(), newDataNodes.size());
- }
- }
-
- private void mockEmptyZonesList() {
- NamedConfigurationTree<DistributionZoneConfiguration, DistributionZoneView, DistributionZoneChange> namedConfigurationTree =
- mock(NamedConfigurationTree.class);
- when(zonesConfiguration.distributionZones()).thenReturn(namedConfigurationTree);
-
- NamedListView<DistributionZoneView> namedListView = mock(NamedListView.class);
- when(namedConfigurationTree.value()).thenReturn(namedListView);
-
- when(zonesConfiguration.distributionZones().value().namedListKeys()).thenReturn(Collections.emptyList());
- }
-
- private void mockZones(DistributionZoneConfiguration... zones) {
- List<String> names = new ArrayList<>();
-
- NamedConfigurationTree<DistributionZoneConfiguration, DistributionZoneView, DistributionZoneChange> namedConfigurationTree =
- mock(NamedConfigurationTree.class);
- when(zonesConfiguration.distributionZones()).thenReturn(namedConfigurationTree);
-
- NamedListView<DistributionZoneView> namedListView = mock(NamedListView.class);
- when(namedConfigurationTree.value()).thenReturn(namedListView);
-
- for (DistributionZoneConfiguration zone : zones) {
- names.add(zone.name().value());
-
- when(namedConfigurationTree.get(zone.name().value())).thenReturn(zone);
- }
-
- when(namedListView.namedListKeys()).thenReturn(names);
- }
-
- private DistributionZoneConfiguration mockZoneConfiguration(
- Integer zoneId,
- String name,
- Integer dataNodesAutoAdjustTime,
- Integer dataNodesAutoAdjustScaleUpTime,
- Integer dataNodesAutoAdjustScaleDownTime
- ) {
- DistributionZoneConfiguration distributionZoneConfiguration = mock(DistributionZoneConfiguration.class);
-
- ConfigurationValue<String> nameValue = mock(ConfigurationValue.class);
- when(distributionZoneConfiguration.name()).thenReturn(nameValue);
- when(nameValue.value()).thenReturn(name);
-
- ConfigurationValue<Integer> zoneIdValue = mock(ConfigurationValue.class);
- when(distributionZoneConfiguration.zoneId()).thenReturn(zoneIdValue);
- when(zoneIdValue.value()).thenReturn(zoneId);
-
- ConfigurationValue<Integer> dataNodesAutoAdjust = mock(ConfigurationValue.class);
- when(distributionZoneConfiguration.dataNodesAutoAdjust()).thenReturn(dataNodesAutoAdjust);
- when(dataNodesAutoAdjust.value()).thenReturn(dataNodesAutoAdjustTime);
-
- ConfigurationValue<Integer> dataNodesAutoAdjustScaleUp = mock(ConfigurationValue.class);
- when(distributionZoneConfiguration.dataNodesAutoAdjustScaleUp()).thenReturn(dataNodesAutoAdjustScaleUp);
- when(dataNodesAutoAdjustScaleUp.value()).thenReturn(dataNodesAutoAdjustScaleUpTime);
-
- ConfigurationValue<Integer> dataNodesAutoAdjustScaleDown = mock(ConfigurationValue.class);
- when(distributionZoneConfiguration.dataNodesAutoAdjustScaleDown()).thenReturn(dataNodesAutoAdjustScaleDown);
- when(dataNodesAutoAdjustScaleDown.value()).thenReturn(dataNodesAutoAdjustScaleDownTime);
-
- return distributionZoneConfiguration;
- }
-
- private DistributionZoneView mockZoneView(
- Integer zoneId,
- String name,
- Integer dataNodesAutoAdjustTime,
- Integer dataNodesAutoAdjustScaleUpTime,
- Integer dataNodesAutoAdjustScaleDownTime
- ) {
- DistributionZoneView distributionZoneView = mock(DistributionZoneView.class);
-
- when(distributionZoneView.name()).thenReturn(name);
- when(distributionZoneView.zoneId()).thenReturn(zoneId);
- when(distributionZoneView.dataNodesAutoAdjust()).thenReturn(dataNodesAutoAdjustTime);
- when(distributionZoneView.dataNodesAutoAdjustScaleUp()).thenReturn(dataNodesAutoAdjustScaleUpTime);
- when(distributionZoneView.dataNodesAutoAdjustScaleDown()).thenReturn(dataNodesAutoAdjustScaleDownTime);
-
- return distributionZoneView;
- }
-
- private DistributionZoneConfiguration mockZoneWithAutoAdjust() {
- return mockZoneConfiguration(1, ZONE_NAME_1, 100, Integer.MAX_VALUE, Integer.MAX_VALUE);
- }
-
- private DistributionZoneConfiguration mockDefaultZoneConfiguration() {
- DistributionZoneConfiguration defaultZone = mockZoneConfiguration(DEFAULT_ZONE_ID, DEFAULT_ZONE_NAME, 100, Integer.MAX_VALUE,
- Integer.MAX_VALUE);
-
- when(zonesConfiguration.defaultDistributionZone()).thenReturn(defaultZone);
-
- return defaultZone;
- }
-
- private DistributionZoneView mockDefaultZoneView() {
- DistributionZoneView defaultZone = mockZoneView(DEFAULT_ZONE_ID, DEFAULT_ZONE_NAME, Integer.MAX_VALUE, 0,
- Integer.MAX_VALUE);
-
- DistributionZonesView zonesView = mock(DistributionZonesView.class);
-
- when(zonesView.defaultDistributionZone()).thenReturn(defaultZone);
- when(zonesConfiguration.value()).thenReturn(zonesView);
-
- return defaultZone;
- }
-
- private DistributionZoneConfiguration mockZoneWithAutoAdjustScaleUp(int scaleUp) {
- return mockZoneConfiguration(1, ZONE_NAME_1, Integer.MAX_VALUE, scaleUp, Integer.MAX_VALUE);
- }
-
-
private void mockVaultZonesLogicalTopologyKey(Set<String> nodes) {
byte[] newLogicalTopology = toBytes(nodes);
@@ -587,4 +341,8 @@ public class DistributionZoneManagerWatchListenerTest extends IgniteAbstractTest
private void mockVaultAppliedRevision(long revision) {
when(metaStorageManager.appliedRevision()).thenReturn(revision);
}
+
+ private void mockCmgLocalNodes() {
+ when(cmgManager.logicalTopology()).thenReturn(completedFuture(topology.getLogicalTopology()));
+ }
}
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/util/DistributionZonesTestUtil.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/util/DistributionZonesTestUtil.java
new file mode 100644
index 0000000000..8286d98401
--- /dev/null
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/util/DistributionZonesTestUtil.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.distributionzones.util;
+
+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.zonesChangeTriggerKey;
+import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zonesLogicalTopologyKey;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.internal.util.ByteUtils.toBytes;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.lenient;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.distributionzones.DistributionZonesUtil;
+import org.apache.ignite.internal.metastorage.Entry;
+import org.apache.ignite.internal.metastorage.MetaStorageManager;
+import org.apache.ignite.internal.metastorage.command.GetAllCommand;
+import org.apache.ignite.internal.metastorage.command.GetCommand;
+import org.apache.ignite.internal.metastorage.command.MetaStorageCommandsFactory;
+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.server.KeyValueStorage;
+import org.apache.ignite.internal.metastorage.server.raft.MetaStorageListener;
+import org.apache.ignite.internal.raft.Command;
+import org.apache.ignite.internal.raft.ReadCommand;
+import org.apache.ignite.internal.raft.WriteCommand;
+import org.apache.ignite.internal.raft.service.CommandClosure;
+import org.apache.ignite.internal.raft.service.RaftGroupService;
+import org.apache.ignite.internal.util.ByteUtils;
+import org.apache.ignite.lang.ByteArray;
+import org.apache.ignite.lang.IgniteInternalException;
+import org.apache.ignite.network.ClusterNode;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Util class for methods for Distribution zones tests.
+ */
+public class DistributionZonesTestUtil {
+ /**
+ * Asserts data nodes from {@link DistributionZonesUtil#zoneDataNodesKey(int)}.
+ *
+ * @param zoneId Zone id.
+ * @param clusterNodes Data nodes.
+ * @param keyValueStorage Key-value storage.
+ * @throws InterruptedException If thread was interrupted.
+ */
+ public static void assertDataNodesForZone(
+ int zoneId,
+ @Nullable Set<String> clusterNodes,
+ KeyValueStorage keyValueStorage
+ ) throws InterruptedException {
+ assertTrue(waitForCondition(
+ () -> {
+ byte[] dataNodes = keyValueStorage.get(zoneDataNodesKey(zoneId).bytes()).value();
+
+ if (dataNodes == null) {
+ return clusterNodes == null;
+ }
+
+ Set<String> res = DistributionZonesUtil.dataNodes(ByteUtils.fromBytes(dataNodes));
+
+ return res.equals(clusterNodes);
+ },
+ 2000
+ ));
+ }
+
+ /**
+ * Asserts {@link DistributionZonesUtil#zoneScaleUpChangeTriggerKey(int)} revision.
+ *
+ * @param revision Revision.
+ * @param zoneId Zone id.
+ * @param keyValueStorage Key-value storage.
+ * @throws InterruptedException If thread was interrupted.
+ */
+ public static void assertZoneScaleUpChangeTriggerKey(
+ int revision,
+ int zoneId,
+ KeyValueStorage keyValueStorage
+ ) throws InterruptedException {
+ assertTrue(
+ waitForCondition(
+ () -> ByteUtils.bytesToLong(keyValueStorage.get(zoneScaleUpChangeTriggerKey(zoneId).bytes()).value()) == revision,
+ 2000
+ )
+ );
+ }
+
+ /**
+ * Asserts {@link DistributionZonesUtil#zoneScaleDownChangeTriggerKey(int)} revision.
+ *
+ * @param revision Revision.
+ * @param zoneId Zone id.
+ * @param keyValueStorage Key-value storage.
+ * @throws InterruptedException If thread was interrupted.
+ */
+ public static void assertZoneScaleDownChangeTriggerKey(
+ int revision,
+ int zoneId,
+ KeyValueStorage keyValueStorage
+ ) throws InterruptedException {
+ assertTrue(
+ waitForCondition(
+ () -> ByteUtils.bytesToLong(keyValueStorage.get(zoneScaleDownChangeTriggerKey(zoneId).bytes()).value()) == revision,
+ 2000
+ )
+ );
+ }
+
+ /**
+ * Asserts {@link DistributionZonesUtil#zonesChangeTriggerKey(int)} revision.
+ *
+ * @param revision Revision.
+ * @param zoneId Zone id.
+ * @param keyValueStorage Key-value storage.
+ * @throws InterruptedException If thread was interrupted.
+ */
+ public static void assertZonesChangeTriggerKey(
+ int revision,
+ int zoneId,
+ KeyValueStorage keyValueStorage
+ ) throws InterruptedException {
+ assertTrue(
+ waitForCondition(
+ () -> ByteUtils.bytesToLong(keyValueStorage.get(zonesChangeTriggerKey(zoneId).bytes()).value()) == revision, 1000
+ )
+ );
+ }
+
+ /**
+ * Asserts {@link DistributionZonesUtil#zonesLogicalTopologyKey()} value.
+ *
+ * @param clusterNodes Expected cluster nodes.
+ * @param keyValueStorage Key-value storage.
+ * @throws InterruptedException If thread was interrupted.
+ */
+ public static void assertLogicalTopology(
+ @Nullable Set<ClusterNode> clusterNodes,
+ KeyValueStorage keyValueStorage
+ ) throws InterruptedException {
+ byte[] nodes = clusterNodes == null
+ ? null
+ : toBytes(clusterNodes.stream().map(ClusterNode::name).collect(Collectors.toSet()));
+
+ assertTrue(waitForCondition(() -> Arrays.equals(keyValueStorage.get(zonesLogicalTopologyKey().bytes()).value(), nodes), 1000));
+ }
+
+ /**
+ * Delegates meta storage manager commands directly to meta storage listener.
+ *
+ * @param raftIndex Raft index.
+ * @param metaStorageListener Meta storage listener.
+ * @param metaStorageService Meta storage service.
+ * @param metaStorageManager Meta storage manager.
+ */
+ public static void mockMetaStorageListener(
+ AtomicLong raftIndex,
+ MetaStorageListener metaStorageListener,
+ RaftGroupService metaStorageService,
+ MetaStorageManager metaStorageManager) {
+ // Delegate directly to listener.
+ lenient().doAnswer(
+ invocationClose -> {
+ Command cmd = invocationClose.getArgument(0);
+
+ long commandIndex = raftIndex.incrementAndGet();
+
+ CompletableFuture<Serializable> res = new CompletableFuture<>();
+
+ CommandClosure<WriteCommand> clo = new CommandClosure<>() {
+ /** {@inheritDoc} */
+ @Override
+ public long index() {
+ return commandIndex;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public WriteCommand command() {
+ return (WriteCommand) cmd;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void result(@Nullable Serializable r) {
+ if (r instanceof Throwable) {
+ res.completeExceptionally((Throwable) r);
+ } else {
+ res.complete(r);
+ }
+ }
+ };
+
+ try {
+ metaStorageListener.onWrite(List.of(clo).iterator());
+ } catch (Throwable e) {
+ res.completeExceptionally(new IgniteInternalException(e));
+ }
+
+ return res;
+ }
+ ).when(metaStorageService).run(any(WriteCommand.class));
+
+ lenient().doAnswer(
+ invocationClose -> {
+ Command cmd = invocationClose.getArgument(0);
+
+ long commandIndex = raftIndex.incrementAndGet();
+
+ CompletableFuture<Serializable> res = new CompletableFuture<>();
+
+ CommandClosure<ReadCommand> clo = new CommandClosure<>() {
+ /** {@inheritDoc} */
+ @Override
+ public long index() {
+ return commandIndex;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public ReadCommand command() {
+ return (ReadCommand) cmd;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void result(@Nullable Serializable r) {
+ if (r instanceof Throwable) {
+ res.completeExceptionally((Throwable) r);
+ } else {
+ res.complete(r);
+ }
+ }
+ };
+
+ try {
+ metaStorageListener.onRead(List.of(clo).iterator());
+ } catch (Throwable e) {
+ res.completeExceptionally(new IgniteInternalException(e));
+ }
+
+ return res;
+ }
+ ).when(metaStorageService).run(any(ReadCommand.class));
+
+ MetaStorageCommandsFactory commandsFactory = new MetaStorageCommandsFactory();
+
+ lenient().doAnswer(invocationClose -> {
+ Iif iif = invocationClose.getArgument(0);
+
+ MultiInvokeCommand multiInvokeCommand = commandsFactory.multiInvokeCommand().iif(iif).build();
+
+ return metaStorageService.run(multiInvokeCommand);
+ }).when(metaStorageManager).invoke(any());
+
+ lenient().doAnswer(invocationClose -> {
+ Set<ByteArray> keysSet = invocationClose.getArgument(0);
+
+ GetAllCommand getAllCommand = commandsFactory.getAllCommand().keys(
+ keysSet.stream().map(ByteArray::bytes).collect(Collectors.toList())
+ ).revision(0).build();
+
+ return metaStorageService.<List<Entry>>run(getAllCommand).thenApply(entries -> {
+ Map<ByteArray, Entry> res = new HashMap<>();
+
+ for (Entry e : entries) {
+ ByteArray key = new ByteArray(e.key());
+
+ res.put(key, new EntryImpl(key.bytes(), e.value(), e.revision(), e.updateCounter()));
+ }
+
+ return res;
+ });
+ }).when(metaStorageManager).getAll(any());
+
+ lenient().doAnswer(invocationClose -> {
+ ByteArray key = invocationClose.getArgument(0);
+
+ GetCommand getCommand = commandsFactory.getCommand().key(key.bytes()).build();
+
+ return metaStorageService.<Entry>run(getCommand).thenApply(
+ entry -> new EntryImpl(entry.key(), entry.value(), entry.revision(), entry.updateCounter())
+ );
+ }).when(metaStorageManager).get(any());
+ }
+}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index 6bfa418150..c2cd5f503a 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -24,6 +24,7 @@ import static java.util.concurrent.CompletableFuture.failedFuture;
import static java.util.concurrent.CompletableFuture.runAsync;
import static java.util.stream.Collectors.toList;
import static org.apache.ignite.internal.configuration.util.ConfigurationUtil.getByInternalId;
+import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.dataNodes;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.extractZoneId;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zoneDataNodesPrefix;
import static org.apache.ignite.internal.schema.SchemaManager.INITIAL_SCHEMA_VERSION;
@@ -1852,7 +1853,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
int zoneId = extractZoneId(evt.entryEvent().newEntry().key());
- Set<String> dataNodes = ByteUtils.fromBytes(dataNodesBytes);
+ Set<String> dataNodes = dataNodes(ByteUtils.fromBytes(dataNodesBytes));
for (int i = 0; i < tables.value().size(); i++) {
TableView tableView = tables.value().get(i);
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerDistributionZonesTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerDistributionZonesTest.java
index 1645d99ebe..fba4182dce 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerDistributionZonesTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerDistributionZonesTest.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.table.distributed;
import static java.util.Collections.emptySet;
import static org.apache.ignite.configuration.annotation.ConfigurationType.DISTRIBUTED;
import static org.apache.ignite.internal.affinity.AffinityUtils.calculateAssignmentForPartition;
+import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.toDataNodesMap;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zoneDataNodesKey;
import static org.apache.ignite.internal.util.ByteUtils.fromBytes;
import static org.apache.ignite.internal.util.ByteUtils.toBytes;
@@ -402,7 +403,7 @@ public class TableManagerDistributionZonesTest extends IgniteAbstractTest {
byte[] newLogicalTopology;
if (nodes != null) {
- newLogicalTopology = toBytes(nodes);
+ newLogicalTopology = toBytes(toDataNodesMap(nodes));
} else {
newLogicalTopology = null;
}