Support zone based virtual topology assignment algorithm

helix-core/src/main/java/org/apache/helix/cloud/constants/VirtualTopologyGroupConstants.java

@@ -24,7 +24,12 @@ public class VirtualTopologyGroupConstants {
   public static final String GROUP_NAME = "virtualTopologyGroupName";
   public static final String GROUP_NUMBER = "virtualTopologyGroupNumber";
   public static final String AUTO_MAINTENANCE_MODE_DISABLED = "autoMaintenanceModeDisabled";
+  public static final String ASSIGNMENT_ALGORITHM = "virtualTopologyAssignmentAlgorithm";
   public static final String GROUP_NAME_SPLITTER = "_";
   public static final String PATH_NAME_SPLITTER = "/";
   public static final String VIRTUAL_FAULT_ZONE_TYPE = "virtualZone";
+
+  public enum VirtualGroupAssignmentAlgorithm {
+    ZONE_BASED, INSTANCE_BASED
+  }
 }

helix-core/src/main/java/org/apache/helix/cloud/topology/FaultZoneBasedVirtualGroupAssignmentAlgorithm.java

@@ -0,0 +1,73 @@
+package org.apache.helix.cloud.topology;
+
+/*
+ * 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.
+ */
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+import static org.apache.helix.util.VirtualTopologyUtil.computeVirtualGroupId;
+
+/**
+ * A strategy that densely assign virtual groups with input instance list, it doesn't move to the next one until
+ * the current one is filled.
+ * Given that instances.size = instancesPerGroup * numGroups + residuals,
+ * we break [residuals] into the first few groups, as a result each virtual group will have
+ * either [instancesPerGroup] or [instancesPerGroup + 1] instances.
+ */
+public class FaultZoneBasedVirtualGroupAssignmentAlgorithm implements VirtualGroupAssignmentAlgorithm {
+
+  private static final FaultZoneBasedVirtualGroupAssignmentAlgorithm _instance = new FaultZoneBasedVirtualGroupAssignmentAlgorithm();
+
+  private FaultZoneBasedVirtualGroupAssignmentAlgorithm() { }
+
+  public static FaultZoneBasedVirtualGroupAssignmentAlgorithm getInstance() {
+    return _instance;
+  }
+
+  @Override
+  public Map<String, Set<String>> computeAssignment(int numGroups, String virtualGroupName,
+      Map<String, Set<String>> zoneMapping) {
+    TreeSet<String> sortdFaultZones = new TreeSet<>(zoneMapping.keySet());
+    int faultZonesPerGroup = sortdFaultZones.size() / numGroups;
+    int residuals = sortdFaultZones.size() % numGroups;
+
+    Map<String, Set<String>> assignment = new TreeMap<>();
+    int virtualGroupIndex = 0;
+    int curFaultZoneNum = 0;
+
+    for (String faultZone : sortdFaultZones) {
+      String groupId = computeVirtualGroupId(virtualGroupIndex, virtualGroupName);
+      if (curFaultZoneNum == 0) {
+        assignment.put(groupId, new HashSet<>());
+      }
+      assignment.get(groupId).addAll(zoneMapping.get(faultZone));
+      curFaultZoneNum++;
+      if (curFaultZoneNum == faultZonesPerGroup + (virtualGroupIndex < residuals ? 1 : 0)) {
+        virtualGroupIndex++;
+        curFaultZoneNum = 0;
+      }
+    }
+
+    return assignment;
+  }
+}

helix-core/src/main/java/org/apache/helix/cloud/topology/FifoVirtualGroupAssignmentAlgorithm.java

@@ -30,6 +30,7 @@
 import org.apache.helix.cloud.constants.VirtualTopologyGroupConstants;
 import org.apache.helix.util.HelixUtil;
 
+import static org.apache.helix.util.VirtualTopologyUtil.computeVirtualGroupId;
 
 /**
  * A strategy that densely assign virtual groups with input instance list, it doesn't move to the next one until
@@ -72,8 +73,4 @@ public Map<String, Set<String>> computeAssignment(int numGroups, String virtualG
     Preconditions.checkState(numInstances.stream().mapToInt(Integer::intValue).sum() == sortedInstances.size());
     return ImmutableMap.copyOf(assignment);
   }
-
-  private static String computeVirtualGroupId(int groupIndex, String virtualGroupName) {
-    return virtualGroupName + VirtualTopologyGroupConstants.GROUP_NAME_SPLITTER + groupIndex;
-  }
 }

helix-core/src/main/java/org/apache/helix/util/VirtualTopologyUtil.java

@@ -0,0 +1,38 @@
+package org.apache.helix.util;
+
+import io.netty.util.internal.StringUtil;
+import org.apache.helix.cloud.constants.VirtualTopologyGroupConstants;
+
+public class VirtualTopologyUtil {
+  public static String computeVirtualGroupId(int groupIndex, String virtualGroupName) {
+    return virtualGroupName + VirtualTopologyGroupConstants.GROUP_NAME_SPLITTER + groupIndex;
+  }
+
+  /**
+   * Ensures the provided fault zone type string ends with
+   * the virtual fault zone type suffix.
+   *
+   * @param oldFaultZoneType The original fault zone type. Must not be null or empty.
+   * @return The fault zone type string with the virtual fault zone type appended if necessary.
+   * @throws IllegalArgumentException if {@code oldFaultZoneType} is null or empty
+   */
+  public static String computeVirtualFaultZoneTypeKey(String oldFaultZoneType) {
+    if (StringUtil.isNullOrEmpty(oldFaultZoneType)) {
+      throw new IllegalArgumentException("The old fault zone type is null or empty");
+    }
+
+    String suffix = VirtualTopologyGroupConstants.GROUP_NAME_SPLITTER
+        + VirtualTopologyGroupConstants.VIRTUAL_FAULT_ZONE_TYPE;
+
+    // If already ends with splitter + VIRTUAL_FAULT_ZONE_TYPE, return as-is
+    if (oldFaultZoneType.endsWith(suffix)) {
+      return oldFaultZoneType;
+    }
+
+    // Otherwise, remove any existing suffix parts beyond the first splitter, if needed
+     String[] segments = oldFaultZoneType.split(VirtualTopologyGroupConstants.GROUP_NAME_SPLITTER);
+     String baseName = segments[0];
+
+     return baseName + suffix;
+  }
+}

helix-core/src/test/java/org/apache/helix/cloud/virtualTopologyGroup/TestFaultZoneBasedVirtualGroupAssignment.java

@@ -0,0 +1,132 @@
+package org.apache.helix.cloud.virtualTopologyGroup;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.helix.cloud.topology.FaultZoneBasedVirtualGroupAssignmentAlgorithm;
+import org.apache.helix.cloud.topology.VirtualGroupAssignmentAlgorithm;
+import org.testng.Assert;
+import org.testng.annotations.BeforeTest;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.apache.helix.util.VirtualTopologyUtil.computeVirtualGroupId;
+
+public class TestFaultZoneBasedVirtualGroupAssignment {
+
+  private static final String GROUP_NAME = "test_virtual_group";
+  private static final int ZONE_NUMBER = 20;
+  private static final int INSTANCES_PER_ZONE = 5;
+  private Map<String, Set<String>> _zoneMapping = new HashMap<>();
+
+  @BeforeTest
+  public void prepare() {
+    _zoneMapping = new HashMap<>();
+    int instanceIdx = 0;
+    for (int i = 0; i < ZONE_NUMBER; i++) {
+      String zone = "zone_" + i;
+      _zoneMapping.computeIfAbsent(zone, k -> new HashSet<>());
+      for (int j = 0; j < INSTANCES_PER_ZONE; j++) {
+        String instance = "instance_" + instanceIdx++;
+        _zoneMapping.get(zone).add(instance);
+      }
+    }
+  }
+
+  @Test(dataProvider = "getMappingTests")
+  public void testAssignmentScheme(int numGroups, Map<String, Set<String>> expected,
+      VirtualGroupAssignmentAlgorithm algorithm) {
+    Assert.assertEquals(algorithm.computeAssignment(numGroups, GROUP_NAME, _zoneMapping), expected);
+  }
+
+  @DataProvider
+  public Object[][] getMappingTests() {
+    VirtualGroupAssignmentAlgorithm algorithm = FaultZoneBasedVirtualGroupAssignmentAlgorithm.getInstance();
+
+    // Ordered zones:
+    // zone_0, zone_1, zone_10, zone_11, zone_12, zone_13, zone_14, zone_15, zone_16, zone_17,
+    // zone_18, zone_19, zone_2, zone_3, zone_4, zone_5, zone_6, zone_7, zone_8, zone_9
+
+    // Expected mapping with 4 virtual groups:
+    // 0 -> zone_0, zone_1, zone_10, zone_11, zone_12
+    // 1 -> zone_13, zone_14, zone_15, zone_16, zone_17
+    // 2 -> zone_18, zone_19, zone_2, zone_3, zone_4
+    // 3 -> zone_5, zone_6, zone_7, zone_8, zone_9
+    Map<String, Set<String>> virtualMapping = new HashMap<>();
+
+    virtualMapping.put(computeVirtualGroupId(0, GROUP_NAME), new HashSet<>());
+    virtualMapping.get(computeVirtualGroupId(0, GROUP_NAME)).addAll(_zoneMapping.get("zone_0"));
+    virtualMapping.get(computeVirtualGroupId(0, GROUP_NAME)).addAll(_zoneMapping.get("zone_1"));
+    virtualMapping.get(computeVirtualGroupId(0, GROUP_NAME)).addAll(_zoneMapping.get("zone_10"));
+    virtualMapping.get(computeVirtualGroupId(0, GROUP_NAME)).addAll(_zoneMapping.get("zone_11"));
+    virtualMapping.get(computeVirtualGroupId(0, GROUP_NAME)).addAll(_zoneMapping.get("zone_12"));
+
+    virtualMapping.put(computeVirtualGroupId(1, GROUP_NAME), new HashSet<>());
+    virtualMapping.get(computeVirtualGroupId(1, GROUP_NAME)).addAll(_zoneMapping.get("zone_13"));
+    virtualMapping.get(computeVirtualGroupId(1, GROUP_NAME)).addAll(_zoneMapping.get("zone_14"));
+    virtualMapping.get(computeVirtualGroupId(1, GROUP_NAME)).addAll(_zoneMapping.get("zone_15"));
+    virtualMapping.get(computeVirtualGroupId(1, GROUP_NAME)).addAll(_zoneMapping.get("zone_16"));
+    virtualMapping.get(computeVirtualGroupId(1, GROUP_NAME)).addAll(_zoneMapping.get("zone_17"));
+
+    virtualMapping.put(computeVirtualGroupId(2, GROUP_NAME), new HashSet<>());
+    virtualMapping.get(computeVirtualGroupId(2, GROUP_NAME)).addAll(_zoneMapping.get("zone_18"));
+    virtualMapping.get(computeVirtualGroupId(2, GROUP_NAME)).addAll(_zoneMapping.get("zone_19"));
+    virtualMapping.get(computeVirtualGroupId(2, GROUP_NAME)).addAll(_zoneMapping.get("zone_2"));
+    virtualMapping.get(computeVirtualGroupId(2, GROUP_NAME)).addAll(_zoneMapping.get("zone_3"));
+    virtualMapping.get(computeVirtualGroupId(2, GROUP_NAME)).addAll(_zoneMapping.get("zone_4"));
+
+    virtualMapping.put(computeVirtualGroupId(3, GROUP_NAME), new HashSet<>());
+    virtualMapping.get(computeVirtualGroupId(3, GROUP_NAME)).addAll(_zoneMapping.get("zone_5"));
+    virtualMapping.get(computeVirtualGroupId(3, GROUP_NAME)).addAll(_zoneMapping.get("zone_6"));
+    virtualMapping.get(computeVirtualGroupId(3, GROUP_NAME)).addAll(_zoneMapping.get("zone_7"));
+    virtualMapping.get(computeVirtualGroupId(3, GROUP_NAME)).addAll(_zoneMapping.get("zone_8"));
+    virtualMapping.get(computeVirtualGroupId(3, GROUP_NAME)).addAll(_zoneMapping.get("zone_9"));
+
+    // Expected mapping with 7 virtual groups:
+    // 0 -> zone_0, zone_1, zone_10
+    // 1 -> zone_11, zone_12, zone_13
+    // 2 -> zone_14, zone_15, zone_16
+    // 3 -> zone_17, zone_18, zone_19
+    // 4 -> zone_2, zone_3, zone_4
+    // 5 -> zone_5, zone_6, zone_7
+    // 6 -> zone_8, zone_9
+    Map<String, Set<String>> virtualMapping2 = new HashMap<>();
+    virtualMapping2.put(computeVirtualGroupId(0, GROUP_NAME), new HashSet<>());
+    virtualMapping2.get(computeVirtualGroupId(0, GROUP_NAME)).addAll(_zoneMapping.get("zone_0"));
+    virtualMapping2.get(computeVirtualGroupId(0, GROUP_NAME)).addAll(_zoneMapping.get("zone_1"));
+    virtualMapping2.get(computeVirtualGroupId(0, GROUP_NAME)).addAll(_zoneMapping.get("zone_10"));
+
+    virtualMapping2.put(computeVirtualGroupId(1, GROUP_NAME), new HashSet<>());
+    virtualMapping2.get(computeVirtualGroupId(1, GROUP_NAME)).addAll(_zoneMapping.get("zone_11"));
+    virtualMapping2.get(computeVirtualGroupId(1, GROUP_NAME)).addAll(_zoneMapping.get("zone_12"));
+    virtualMapping2.get(computeVirtualGroupId(1, GROUP_NAME)).addAll(_zoneMapping.get("zone_13"));
+
+    virtualMapping2.put(computeVirtualGroupId(2, GROUP_NAME), new HashSet<>());
+    virtualMapping2.get(computeVirtualGroupId(2, GROUP_NAME)).addAll(_zoneMapping.get("zone_14"));
+    virtualMapping2.get(computeVirtualGroupId(2, GROUP_NAME)).addAll(_zoneMapping.get("zone_15"));
+    virtualMapping2.get(computeVirtualGroupId(2, GROUP_NAME)).addAll(_zoneMapping.get("zone_16"));
+
+    virtualMapping2.put(computeVirtualGroupId(3, GROUP_NAME), new HashSet<>());
+    virtualMapping2.get(computeVirtualGroupId(3, GROUP_NAME)).addAll(_zoneMapping.get("zone_17"));
+    virtualMapping2.get(computeVirtualGroupId(3, GROUP_NAME)).addAll(_zoneMapping.get("zone_18"));
+    virtualMapping2.get(computeVirtualGroupId(3, GROUP_NAME)).addAll(_zoneMapping.get("zone_19"));
+
+    virtualMapping2.put(computeVirtualGroupId(4, GROUP_NAME), new HashSet<>());
+    virtualMapping2.get(computeVirtualGroupId(4, GROUP_NAME)).addAll(_zoneMapping.get("zone_2"));
+    virtualMapping2.get(computeVirtualGroupId(4, GROUP_NAME)).addAll(_zoneMapping.get("zone_3"));
+    virtualMapping2.get(computeVirtualGroupId(4, GROUP_NAME)).addAll(_zoneMapping.get("zone_4"));
+
+    virtualMapping2.put(computeVirtualGroupId(5, GROUP_NAME), new HashSet<>());
+    virtualMapping2.get(computeVirtualGroupId(5, GROUP_NAME)).addAll(_zoneMapping.get("zone_5"));
+    virtualMapping2.get(computeVirtualGroupId(5, GROUP_NAME)).addAll(_zoneMapping.get("zone_6"));
+    virtualMapping2.get(computeVirtualGroupId(5, GROUP_NAME)).addAll(_zoneMapping.get("zone_7"));
+
+    virtualMapping2.put(computeVirtualGroupId(6, GROUP_NAME), new HashSet<>());
+    virtualMapping2.get(computeVirtualGroupId(6, GROUP_NAME)).addAll(_zoneMapping.get("zone_8"));
+    virtualMapping2.get(computeVirtualGroupId(6, GROUP_NAME)).addAll(_zoneMapping.get("zone_9"));
+
+    return new Object[][]{{4, virtualMapping, algorithm}, {7, virtualMapping2, algorithm}};
+  }
+}

helix-core/src/test/java/org/apache/helix/cloud/virtualTopologyGroup/TestVirtualTopologyGroupAssignment.java

@@ -25,7 +25,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import org.apache.helix.cloud.constants.VirtualTopologyGroupConstants;
+
 import org.apache.helix.cloud.topology.FifoVirtualGroupAssignmentAlgorithm;
 import org.apache.helix.cloud.topology.VirtualGroupAssignmentAlgorithm;
 import org.apache.helix.util.HelixUtil;
@@ -34,6 +34,8 @@
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
+import static org.apache.helix.util.VirtualTopologyUtil.computeVirtualGroupId;
+
 public class TestVirtualTopologyGroupAssignment {
 
   private static final String GROUP_NAME = "test_virtual_group";
@@ -68,27 +70,24 @@ public void testAssignmentScheme(int numGroups, Map<String, Set<String>> expecte
   public Object[][] getMappingTests() {
     Map<String, Set<String>> virtualMapping = new HashMap<>();
     VirtualGroupAssignmentAlgorithm algorithm = FifoVirtualGroupAssignmentAlgorithm.getInstance();
-    virtualMapping.put(computeVirtualGroupId(0), Sets.newHashSet("1", "2", "3", "4", "5"));
-    virtualMapping.put(computeVirtualGroupId(1), Sets.newHashSet("6", "7", "8", "9"));
-    virtualMapping.put(computeVirtualGroupId(2), Sets.newHashSet("a", "b", "c", "d"));
+    virtualMapping.put(computeVirtualGroupId(0, GROUP_NAME), Sets.newHashSet("1", "2", "3", "4", "5"));
+    virtualMapping.put(computeVirtualGroupId(1, GROUP_NAME), Sets.newHashSet("6", "7", "8", "9"));
+    virtualMapping.put(computeVirtualGroupId(2, GROUP_NAME), Sets.newHashSet("a", "b", "c", "d"));
     Assert.assertEquals(algorithm.computeAssignment(3, GROUP_NAME, _zoneMapping),
         virtualMapping);
     Map<String, Set<String>> virtualMapping2 = new HashMap<>();
-    virtualMapping2.put(computeVirtualGroupId(0), Sets.newHashSet("1", "2"));
-    virtualMapping2.put(computeVirtualGroupId(1), Sets.newHashSet("3", "4"));
-    virtualMapping2.put(computeVirtualGroupId(2), Sets.newHashSet("5", "6"));
-    virtualMapping2.put(computeVirtualGroupId(3), Sets.newHashSet("7", "8"));
-    virtualMapping2.put(computeVirtualGroupId(4), Sets.newHashSet("9", "a"));
-    virtualMapping2.put(computeVirtualGroupId(5), Sets.newHashSet("b"));
-    virtualMapping2.put(computeVirtualGroupId(6), Sets.newHashSet("c"));
-    virtualMapping2.put(computeVirtualGroupId(7), Sets.newHashSet("d"));
+    virtualMapping2.put(computeVirtualGroupId(0, GROUP_NAME), Sets.newHashSet("1", "2"));
+    virtualMapping2.put(computeVirtualGroupId(1, GROUP_NAME), Sets.newHashSet("3", "4"));
+    virtualMapping2.put(computeVirtualGroupId(2, GROUP_NAME), Sets.newHashSet("5", "6"));
+    virtualMapping2.put(computeVirtualGroupId(3, GROUP_NAME), Sets.newHashSet("7", "8"));
+    virtualMapping2.put(computeVirtualGroupId(4, GROUP_NAME), Sets.newHashSet("9", "a"));
+    virtualMapping2.put(computeVirtualGroupId(5, GROUP_NAME), Sets.newHashSet("b"));
+    virtualMapping2.put(computeVirtualGroupId(6, GROUP_NAME), Sets.newHashSet("c"));
+    virtualMapping2.put(computeVirtualGroupId(7, GROUP_NAME), Sets.newHashSet("d"));
+
     return new Object[][] {
         {3, virtualMapping, algorithm},
         {8, virtualMapping2, algorithm}
     };
   }
-
-  private static String computeVirtualGroupId(int groupIndex) {
-    return GROUP_NAME + VirtualTopologyGroupConstants.GROUP_NAME_SPLITTER + groupIndex;
-  }
 }

helix-rest/src/main/java/org/apache/helix/rest/server/service/ClusterService.java

@@ -35,6 +35,13 @@ public interface ClusterService {
    */
   ClusterTopology getClusterTopology(String cluster);
 
+  /**
+   * Get virtual cluster topology of the given cluster
+   * @param cluster
+   * @return virtual cluster topology
+   */
+  ClusterTopology getVirtualClusterTopology(String cluster);
+
   /**
    * Get cluster basic information
    * @param clusterId

helix-rest/src/main/java/org/apache/helix/rest/server/service/ClusterServiceImpl.java

@@ -30,6 +30,7 @@
 import org.apache.helix.ConfigAccessor;
 import org.apache.helix.HelixDataAccessor;
 import org.apache.helix.PropertyKey;
+import org.apache.helix.cloud.constants.VirtualTopologyGroupConstants;
 import org.apache.helix.model.ClusterConfig;
 import org.apache.helix.model.InstanceConfig;
 import org.apache.helix.model.LiveInstance;
@@ -48,18 +49,29 @@ public ClusterServiceImpl(HelixDataAccessor dataAccessor, ConfigAccessor configA
   @Override
   public ClusterTopology getClusterTopology(String cluster) {
     String zoneField = _configAccessor.getClusterConfig(cluster).getFaultZoneType();
+    return getTopologyUnderDomainType(zoneField, cluster);
+  }
+
+  @Override
+  public ClusterTopology getVirtualClusterTopology(String cluster) {
+    String concatenatedKey = _configAccessor.getClusterConfig(cluster).getFaultZoneType();
+    String faultZone = concatenatedKey.split(VirtualTopologyGroupConstants.GROUP_NAME_SPLITTER)[0];
+    return getTopologyUnderDomainType(faultZone, cluster);
+  }
+
+  private ClusterTopology getTopologyUnderDomainType(String faultZone, String clusterId) {
     PropertyKey.Builder keyBuilder = _dataAccessor.keyBuilder();
     List<InstanceConfig> instanceConfigs =
         _dataAccessor.getChildValues(keyBuilder.instanceConfigs(), true);
     Map<String, List<ClusterTopology.Instance>> instanceMapByZone = new HashMap<>();
     if (instanceConfigs != null && !instanceConfigs.isEmpty()) {
       for (InstanceConfig instanceConfig : instanceConfigs) {
-        if (!instanceConfig.getDomainAsMap().containsKey(zoneField)) {
+        if (!instanceConfig.getDomainAsMap().containsKey(faultZone)) {
           continue;
         }
         final String instanceName = instanceConfig.getInstanceName();
         final ClusterTopology.Instance instance = new ClusterTopology.Instance(instanceName);
-        final String zoneId = instanceConfig.getDomainAsMap().get(zoneField);
+        final String zoneId = instanceConfig.getDomainAsMap().get(faultZone);
         if (instanceMapByZone.containsKey(zoneId)) {
           instanceMapByZone.get(zoneId).add(instance);
         } else {
@@ -79,7 +91,7 @@ public ClusterTopology getClusterTopology(String cluster) {
     }
 
     // Get all the instances names
-    return new ClusterTopology(cluster, zones,
+    return new ClusterTopology(clusterId, zones,
         instanceConfigs.stream().map(InstanceConfig::getInstanceName).collect(Collectors.toSet()));
   }