[GitHub] [ignite-3] alievmirza commented on a diff in pull request #1508: IGNITE-18121 add scale up scheduler

[GitBox <gi...@apache.org>]

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: notifications-unsubscribe@ignite.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org