jasperjiaguo commented on code in PR #8434: URL: https://github.com/apache/pinot/pull/8434#discussion_r892895430
########## pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/FDAwareInstancePartitionSelector.java: ########## @@ -0,0 +1,455 @@ +/** + * 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.pinot.controller.helix.core.assignment.instance; + +import com.google.common.base.Preconditions; +import java.util.ArrayList; +import java.util.Deque; +import java.util.HashMap; +import java.util.LinkedHashSet; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.NavigableMap; +import java.util.Optional; +import java.util.TreeMap; +import java.util.stream.Collectors; +import javax.annotation.Nullable; +import org.apache.commons.lang3.tuple.ImmutablePair; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.helix.model.InstanceConfig; +import org.apache.pinot.common.assignment.InstancePartitions; +import org.apache.pinot.spi.config.table.assignment.InstanceReplicaGroupPartitionConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * The instance replica-group/partition selector is responsible for selecting the instances for every replica-group/ + * partition, with each fault-domain mapping 1:1 to a server pool + * Algorithm details and proof see + * https://docs.google.com/document/d/1KmJ1DsYXVdzrojj_JYBHRJ2gRMQ5y-o63YqPs7ei7nI/edit#heading=h.nrxxvujq7ael + */ +public class FDAwareInstancePartitionSelector extends InstancePartitionSelector { + private static final Logger LOGGER = LoggerFactory.getLogger(FDAwareInstancePartitionSelector.class); + + public FDAwareInstancePartitionSelector(InstanceReplicaGroupPartitionConfig replicaGroupPartitionConfig, + String tableNameWithType, @Nullable InstancePartitions existingInstancePartitions) { + super(replicaGroupPartitionConfig, tableNameWithType, existingInstancePartitions); + } + + /** + * @return pair of (numFaultDomains, numTotalInstances) + */ + private Pair<Integer, Integer> processFaultDomainPreconditions( + Map<Integer, List<InstanceConfig>> faultDomainToInstanceConfigsMap) { + int numFDs = faultDomainToInstanceConfigsMap.size(); + Preconditions.checkState(numFDs != 0, "No pool (fault-domain) qualified for selection"); + + // Collect the number of instances in each FD into a list and sort + List<Integer> numInstancesPerFD = + faultDomainToInstanceConfigsMap.values().stream().map(List::size).sorted().collect(Collectors.toList()); + // Should have non-zero total instances + Optional<Integer> totalInstancesOptional = numInstancesPerFD.stream().reduce(Integer::sum); + Preconditions.checkState(totalInstancesOptional.orElse(0) > 0, "The number of total instances is zero"); + int numTotalInstances = totalInstancesOptional.get(); + // Assume best-effort round-robin assignment of instances to FDs, the assignment should be balanced + // i.e., the difference between max num instances and min num instances should be at most 1 + Preconditions.checkState(numInstancesPerFD.get(numInstancesPerFD.size() - 1) - numInstancesPerFD.get(0) <= 1, + "The instances are not balanced for each pool (fault-domain)"); + return new ImmutablePair<>(numFDs, numTotalInstances); + } + + /** + * @return (numReplicaGroups, numInstancesPerReplicaGroup) + */ + private Pair<Integer, Integer> processReplicaGroupAssignmentPreconditions(int numFaultDomains, int numTotalInstances, + InstanceReplicaGroupPartitionConfig replicaGroupPartitionConfig) { + int numReplicaGroups = replicaGroupPartitionConfig.getNumReplicaGroups(); + Preconditions.checkState(numReplicaGroups > 0, "Number of replica-groups must be positive"); + int numInstancesPerReplicaGroup = replicaGroupPartitionConfig.getNumInstancesPerReplicaGroup(); + if (numInstancesPerReplicaGroup > 0) { + int numInstancesToSelect = numInstancesPerReplicaGroup * numReplicaGroups; + + Preconditions.checkState(numInstancesToSelect <= numTotalInstances, + "Not enough qualified instances (%s in all FDs, asked for %s)", numTotalInstances, numInstancesToSelect); Review Comment: added ########## pinot-spi/src/main/java/org/apache/pinot/spi/config/table/assignment/InstanceAssignmentConfig.java: ########## @@ -28,6 +28,8 @@ public class InstanceAssignmentConfig extends BaseJsonConfig { + @JsonPropertyDescription("Configuration for the strategy to assign instances to partitions") + public final PartitionSelector _partitionSelector; Review Comment: added -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
