alievmirza commented on code in PR #1508:
URL: https://github.com/apache/ignite-3/pull/1508#discussion_r1068476869
##########
modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java:
##########
@@ -806,4 +866,230 @@ private static Throwable
unwrapDistributionZoneExceptionRecursively(Throwable e,
return null;
}
+
+ /**
+ * Method updates data nodes value for the specified zone after scale up
timer timeout,
+ * also sets {@code revision} to the {@link
DistributionZonesUtil#zoneScaleUpChangeTriggerKey(int)} if it passes the
condition.
+ *
+ * @param zoneId Unique id of a zone
+ * @param revision Revision of an event that has triggered this method.
+ */
+ private CompletableFuture<Boolean> saveDataNodesToMetaStorageOnScaleUp(int
zoneId, long revision) {
+ ZoneState zoneState = zonesTimers.get(zoneId);
+
+ if (zoneState == null) {
+ // Zone was deleted
+ return completedFuture(false);
+ }
+
+ ReentrantLock lockForTimers = zoneState.lockForTimers();
+
+ lockForTimers.lock();
+
+ Set<ByteArray> keysToGetFromMs = Set.of(
+ zoneDataNodesKey(zoneId),
+ zoneScaleUpChangeTriggerKey(zoneId),
+ zoneScaleDownChangeTriggerKey(zoneId)
+ );
+
+ return metaStorageManager.getAll(keysToGetFromMs).thenCompose(values
-> {
+ if (values.containsValue(null)) {
+ // Zone was deleted
+ return completedFuture(false);
+ }
+
+ Set<String> 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 <= scaleUpTriggerRevision) {
+ return completedFuture(false);
+ }
+
+ ReentrantLock lockForZone = zoneState.lock();
+
+ lockForZone.lock();
+
+ Set<String> deltaToAdd;
+
+ try {
+ deltaToAdd = new HashSet<>(zoneState.nodesToAdd());
+ } finally {
+ lockForZone.unlock();
+ }
+
+ Set<String> newDataNodes = new HashSet<>(dataNodesFromMetaStorage);
+
+ newDataNodes.addAll(deltaToAdd);
+
+ Update dataNodesAndTriggerKeyUpd =
updateDataNodesAndScaleUpTriggerKey(zoneId, revision, toBytes(newDataNodes));
+
+ If iif = If.iif(
+
triggerScaleUpScaleDownKeysCondition(scaleUpTriggerRevision,
scaleDownTriggerRevision, zoneId),
+ dataNodesAndTriggerKeyUpd,
+ ops().yield(false)
+ );
+
+ return
metaStorageManager.invoke(iif).thenApply(StatementResult::getAsBoolean).thenCompose(invokeResult
-> {
+ if (invokeResult) {
+ lockForZone.lock();
+ try {
+ zoneState.nodesToAdd().clear();
Review Comment:
yeah, thats obviously an inattention from my side, algorithm in pseudocode
has the logic that you've described
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
