diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/BalancedInstanceSelector.java b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/BalancedInstanceSelector.java index 13c5c0668d9d..10526285acfd 100644 --- a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/BalancedInstanceSelector.java +++ b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/BalancedInstanceSelector.java @@ -21,7 +21,6 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import org.apache.commons.lang3.tuple.Pair; import org.apache.pinot.broker.routing.adaptiveserverselector.ServerSelectionContext; import org.apache.pinot.common.metrics.BrokerMeter; import org.apache.pinot.common.metrics.BrokerMetrics; @@ -45,7 +44,7 @@ public class BalancedInstanceSelector extends BaseInstanceSelector { @Override - public Pair, Map> select(List segments, int requestId, + public InstanceMapping select(List segments, int requestId, SegmentStates segmentStates, Map queryOptions) { Map segmentToSelectedInstanceMap = new HashMap<>(HashUtil.getHashMapCapacity(segments.size())); // No need to adjust this map per total segment numbers, as optional segments should be empty most of the time. @@ -88,6 +87,6 @@ public Pair, Map> select(List segmen _brokerMetrics.addMeteredValue(BrokerMeter.POOL_SEG_QUERIES, entry.getValue(), BrokerMetrics.getTagForPreferredPool(queryOptions), String.valueOf(entry.getKey())); } - return Pair.of(segmentToSelectedInstanceMap, optionalSegmentToInstanceMap); + return new InstanceMapping(segmentToSelectedInstanceMap, optionalSegmentToInstanceMap); } } diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/BaseInstanceSelector.java b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/BaseInstanceSelector.java index 3108e7e6b840..7382ff4290fb 100644 --- a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/BaseInstanceSelector.java +++ b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/BaseInstanceSelector.java @@ -30,7 +30,6 @@ import java.util.TreeMap; import java.util.TreeSet; import javax.annotation.Nullable; -import org.apache.commons.lang3.tuple.Pair; import org.apache.helix.AccessOption; import org.apache.helix.model.ExternalView; import org.apache.helix.model.IdealState; @@ -465,20 +464,20 @@ public SelectionResult select(BrokerRequest brokerRequest, List segments // Copy the volatile reference so that segmentToInstanceMap and unavailableSegments can have a consistent view of // the state. SegmentStates segmentStates = _segmentStates; - Pair, Map> segmentToInstanceMap = - select(segments, requestIdInt, segmentStates, queryOptions); + InstanceMapping mapping = select(segments, requestIdInt, segmentStates, queryOptions); Set unavailableSegments = segmentStates.getUnavailableSegments(); + List mappingUnavailable = mapping.unavailableSegments(); - if (unavailableSegments.isEmpty()) { - return new SelectionResult(segmentToInstanceMap, List.of(), 0); + if (unavailableSegments.isEmpty() && mappingUnavailable.isEmpty()) { + return new SelectionResult(mapping, List.of(), 0); } else { - List unavailableSegmentsForRequest = new ArrayList<>(); + List unavailableSegmentsForRequest = new ArrayList<>(mappingUnavailable); for (String segment : segments) { if (unavailableSegments.contains(segment)) { unavailableSegmentsForRequest.add(segment); } } - return new SelectionResult(segmentToInstanceMap, unavailableSegmentsForRequest, 0); + return new SelectionResult(mapping, unavailableSegmentsForRequest, 0); } } @@ -501,11 +500,11 @@ int getPool(String instanceID) { } /** - * Selects the server instances for the given segments based on the request id and segment states. Returns two maps - * from segment to selected server instance hosting the segment. The 2nd map is for optional segments. The optional - * segments are used to get the new segments that are not online yet. Instead of simply skipping them by broker at - * routing time, we can send them to servers and let servers decide how to handle them. + * Selects the server instances for the given segments based on the request id and segment states. Returns an + * {@link InstanceSelector.InstanceMapping} containing two maps from segment to selected server instance. The optional + * map covers new segments that are not online yet; instead of simply skipping them at the broker, we can send them to + * servers and let servers decide how to handle them. */ - protected abstract Pair, Map/*optional segments*/> select(List segments, - int requestId, SegmentStates segmentStates, Map queryOptions); + protected abstract InstanceMapping select(List segments, int requestId, SegmentStates segmentStates, + Map queryOptions); } diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelector.java b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelector.java index 664ea8463681..5622007d2b08 100644 --- a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelector.java +++ b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelector.java @@ -23,7 +23,6 @@ import java.util.Map; import java.util.Set; import javax.annotation.Nullable; -import org.apache.commons.lang3.tuple.Pair; import org.apache.helix.model.ExternalView; import org.apache.helix.model.IdealState; import org.apache.helix.store.zk.ZkHelixPropertyStore; @@ -83,23 +82,43 @@ void init(TableConfig tableConfig, ZkHelixPropertyStore propertyStore, */ Set getServingInstances(); + /** + * Holds the result of an instance selection: {@code segmentToInstanceMap} maps each segment to its selected server + * instance, {@code optionalSegmentToInstanceMap} maps segments not yet fully online that the server may skip, and + * {@code unavailableSegments} lists segments that have candidates but could not be routed. + */ + record InstanceMapping(Map segmentToInstanceMap, + Map optionalSegmentToInstanceMap, + List unavailableSegments) { + static final InstanceMapping EMPTY = + new InstanceMapping(Map.of(), Map.of(), List.of()); + + InstanceMapping(Map segmentToInstanceMap, Map optionalSegmentToInstanceMap) { + this(segmentToInstanceMap, optionalSegmentToInstanceMap, List.of()); + } + } + class SelectionResult { - private final Pair, Map/*optional segments*/> _segmentToInstanceMap; + private final InstanceMapping _instanceMapping; private final List _unavailableSegments; private int _numPrunedSegments; - public SelectionResult(Pair, Map> segmentToInstanceMap, + public SelectionResult(InstanceMapping instanceMapping, List unavailableSegments, int numPrunedSegments) { - _segmentToInstanceMap = segmentToInstanceMap; + _instanceMapping = instanceMapping; _unavailableSegments = unavailableSegments; _numPrunedSegments = numPrunedSegments; } + public static SelectionResult empty(int numPrunedSegments) { + return new SelectionResult(InstanceMapping.EMPTY, List.of(), numPrunedSegments); + } + /** * Returns the map from segment to selected server instance hosting the segment. */ public Map getSegmentToInstanceMap() { - return _segmentToInstanceMap.getLeft(); + return _instanceMapping.segmentToInstanceMap(); } /** @@ -107,7 +126,7 @@ public Map getSegmentToInstanceMap() { * Optional segments can be skipped by broker or server upon any issue w/o failing the query. */ public Map getOptionalSegmentToInstanceMap() { - return _segmentToInstanceMap.getRight(); + return _instanceMapping.optionalSegmentToInstanceMap(); } /** diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelectorFactory.java b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelectorFactory.java index 49536b93a8fc..bd8923960a04 100644 --- a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelectorFactory.java +++ b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelectorFactory.java @@ -106,6 +106,14 @@ public static InstanceSelector getInstanceSelector(TableConfig tableConfig, } case RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE: { LOGGER.info("Using StrictReplicaGroupInstanceSelector for table: {}", tableNameWithType); + boolean enableStrictReplicaGroupAdaptiveRouting = brokerConfig.getProperty( + CommonConstants.Broker.AdaptiveServerSelector.CONFIG_OF_ENABLE_STRICT_REPLICA_GROUP, + CommonConstants.Broker.AdaptiveServerSelector.DEFAULT_ENABLE_STRICT_REPLICA_GROUP); + if (!enableStrictReplicaGroupAdaptiveRouting && adaptiveServerSelector != null) { + LOGGER.info("Adaptive routing disabled for StrictReplicaGroupInstanceSelector table: {}", + tableNameWithType); + adaptiveServerSelector = null; + } instanceSelector = new StrictReplicaGroupInstanceSelector(); break; } diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/MultiStageReplicaGroupSelector.java b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/MultiStageReplicaGroupSelector.java index 451b47dfe89b..6b0087656c7a 100644 --- a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/MultiStageReplicaGroupSelector.java +++ b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/MultiStageReplicaGroupSelector.java @@ -29,7 +29,6 @@ import java.util.Set; import javax.annotation.Nullable; import org.apache.commons.collections4.CollectionUtils; -import org.apache.commons.lang3.tuple.Pair; import org.apache.helix.model.ExternalView; import org.apache.helix.model.IdealState; import org.apache.helix.store.zk.ZkHelixPropertyStore; @@ -86,7 +85,7 @@ public void onAssignmentChange(IdealState idealState, ExternalView externalView, } @Override - public Pair, Map> select(List segments, int requestId, + public InstanceMapping select(List segments, int requestId, SegmentStates segmentStates, Map queryOptions) { // Create a copy of InstancePartitions to avoid race-condition with event-listeners above. InstancePartitions instancePartitions = _instancePartitions; @@ -110,8 +109,9 @@ public Pair, Map> select(List segmen * Returns a map from the segmentName to the corresponding server. It tries to select all servers from the * preferredReplicaGroup, but if it fails, it will try to select the relevant server from other instance partitions. * - * @return A pair of maps, where the first map contains the segments that are assigned to a server and the second - * map contains the segments that are optional (i.e., the server is not online to serve that segment). + * @return An {@link InstanceMapping} whose {@code segmentToInstanceMap} contains the segments assigned to a server + * and whose {@code optionalSegmentToInstanceMap} contains segments that are optional (i.e., the server is not + * online to serve that segment). * Example: * { * "required_segments": { @@ -124,7 +124,7 @@ public Pair, Map> select(List segmen * } * } */ - private Pair, Map> assign(Set segments, + private InstanceMapping assign(Set segments, SegmentStates segmentStates, InstancePartitions instancePartitions, int preferredReplicaId) { Map instanceToPartitionMap = instancePartitions.getInstanceToPartitionIdMap(); Map> instanceToSegmentsMap = new HashMap<>(); @@ -229,7 +229,7 @@ private void getSelectedInstancesForPartition(Map> instanceT * Based on whether the selected instance for the segment is online to serve that segment or not, * this method computes the segments that are optional and the segments that are not. */ - private Pair, Map> computeOptionalSegments( + private InstanceMapping computeOptionalSegments( Map segmentToSelectedInstanceMap, SegmentStates segmentStates) { Map segmentsToInstanceMap = new HashMap<>(); @@ -258,7 +258,7 @@ private Pair, Map> computeOptionalSegments( } } - return Pair.of(segmentsToInstanceMap, optionalSegmentToInstanceMap); + return new InstanceMapping(segmentsToInstanceMap, optionalSegmentToInstanceMap); } @VisibleForTesting diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/ReplicaGroupInstanceSelector.java b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/ReplicaGroupInstanceSelector.java index 8f35ba28b6af..4726a7469f4a 100644 --- a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/ReplicaGroupInstanceSelector.java +++ b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/ReplicaGroupInstanceSelector.java @@ -27,7 +27,6 @@ import java.util.Set; import javax.annotation.Nullable; import org.apache.commons.collections.MapUtils; -import org.apache.commons.lang3.tuple.Pair; import org.apache.helix.model.ExternalView; import org.apache.helix.model.IdealState; import org.apache.pinot.broker.routing.adaptiveserverselector.ServerSelectionContext; @@ -70,7 +69,7 @@ public class ReplicaGroupInstanceSelector extends BaseInstanceSelector { private static final Logger LOGGER = LoggerFactory.getLogger(ReplicaGroupInstanceSelector.class); @Override - public Pair, Map> select(List segments, int requestId, + public InstanceMapping select(List segments, int requestId, SegmentStates segmentStates, Map queryOptions) { ServerSelectionContext ctx = new ServerSelectionContext(queryOptions, _config); if (_adaptiveServerSelector != null) { @@ -91,7 +90,7 @@ public Pair, Map> select(List segmen } } - private Pair, Map> selectServers(List segments, int requestId, + protected InstanceMapping selectServers(List segments, int requestId, SegmentStates segmentStates, @Nullable Map serverRankMap, ServerSelectionContext ctx) { Map segmentToSelectedInstanceMap = new HashMap<>(HashUtil.getHashMapCapacity(segments.size())); @@ -121,7 +120,7 @@ private Pair, Map> selectServers(List !serverRankMap.containsKey(candidate.getInstance())) ? selectedInstance @@ -147,7 +146,7 @@ private Pair, Map> selectServers(List fetchCandidateServersForQuery(List segments, diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/StrictReplicaGroupInstanceSelector.java b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/StrictReplicaGroupInstanceSelector.java index 04e3b25b359d..249618d93435 100644 --- a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/StrictReplicaGroupInstanceSelector.java +++ b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/StrictReplicaGroupInstanceSelector.java @@ -18,10 +18,24 @@ */ package org.apache.pinot.broker.routing.instanceselector; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; import java.util.Set; +import java.util.stream.Collectors; import org.apache.helix.model.ExternalView; import org.apache.helix.model.IdealState; +import org.apache.pinot.broker.routing.adaptiveserverselector.ServerSelectionContext; +import org.apache.pinot.common.metrics.BrokerMeter; +import org.apache.pinot.common.metrics.BrokerMetrics; +import org.apache.pinot.common.utils.HashUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + /** * Instance selector for strict replica-group routing strategy. @@ -50,14 +64,175 @@ * segments with the same assignment ([S1, S2, S3]) down on S1 to ensure that we always route the segments to the same * replica-group. * + * When adaptive server selection is enabled, this selector uses replica-group-level adaptive routing: it picks the best + * replica group for the entire query (using worst-case server rank within each group) and routes all segments to + * that group. This preserves the same-replica-group guarantee while benefiting from adaptive routing intelligence. + * * Note that new segments won't be used to exclude instances from serving when the segment is unavailable. * */ public class StrictReplicaGroupInstanceSelector extends ReplicaGroupInstanceSelector { + private static final Logger LOGGER = LoggerFactory.getLogger(StrictReplicaGroupInstanceSelector.class); @Override void updateSegmentMaps(IdealState idealState, ExternalView externalView, Set onlineSegments, Map newSegmentCreationTimeMap) { super.updateSegmentMapsForUpsertTable(idealState, externalView, onlineSegments, newSegmentCreationTimeMap); } + + @Override + public InstanceMapping select(List segments, int requestId, + SegmentStates segmentStates, Map queryOptions) { + + if (_adaptiveServerSelector == null || _priorityPoolInstanceSelector == null) { + ServerSelectionContext ctx = new ServerSelectionContext(queryOptions, _config); + return selectServers(segments, requestId, segmentStates, null, ctx); + } + + // Build a map: idealStateReplicaId (replica group) -> distinct servers in that group that are candidates for this + // query. Keyed by instance name to deduplicate (a server hosting multiple segments appears once per group). + // Also track how many query segments each replica group can serve (segment coverage) in a single pass. + // The guard defends against multiple candidates per replica group per segment (e.g., during rebalancing). + Map> replicaGroupToQueryServers = new LinkedHashMap<>(); + Map replicaGroupSegmentCount = new HashMap<>(); + int totalSegmentsWithCandidates = 0; + + Set seenReplicaIds = new HashSet<>(); // allocate once and clear per segment to avoid O(n) allocations + for (String segment : segments) { + List candidates = segmentStates.getCandidates(segment); + if (candidates == null) { + continue; + } + + seenReplicaIds.clear(); + + totalSegmentsWithCandidates++; + for (SegmentInstanceCandidate candidate : candidates) { + int replicaId = candidate.getIdealStateReplicaId(); + replicaGroupToQueryServers + .computeIfAbsent(replicaId, k -> new LinkedHashMap<>()) + .putIfAbsent(candidate.getInstance(), candidate); + if (seenReplicaIds.add(replicaId)) { + replicaGroupSegmentCount.merge(replicaId, 1, Integer::sum); + } + } + } + + if (replicaGroupToQueryServers.isEmpty()) { + return new InstanceMapping(Map.of(), Map.of()); + } + + // Collect all distinct query-relevant candidates from replicaGroupToQueryServers, which already + // deduplicates by instance name. This avoids a second traversal of segments. + List allQueryCandidates = replicaGroupToQueryServers.values().stream() + .flatMap(m -> m.values().stream()) + .collect(Collectors.toList()); + ServerSelectionContext ctx = new ServerSelectionContext(queryOptions, _config); + int bestReplicaGroupId = chooseBestReplicaGroup( + replicaGroupToQueryServers, replicaGroupSegmentCount, totalSegmentsWithCandidates, + allQueryCandidates, ctx, requestId); + return selectServersForReplicaGroup(segments, bestReplicaGroupId, segmentStates, queryOptions); + } + + /** + * Scores and ranks replica groups using adaptive server selection stats. + * + * Each replica group is scored by the worst (maximum) rank among its query-relevant servers. + * Since scatter-gather query latency is bounded by the slowest server, we pick the group + * whose bottleneck server is the best (lowest worst-case rank). + * + * Groups that can serve ALL query segments (full coverage) are preferred over groups that + * would drop segments. Among full-coverage groups, the one with the best worst-case rank wins. + * If no group has full coverage, the best-ranked group is chosen regardless of coverage. + */ + private int chooseBestReplicaGroup( + Map> replicaGroupToQueryServers, + Map replicaGroupSegmentCount, + int totalSegmentsWithCandidates, + List allQueryCandidates, + ServerSelectionContext ctx, + int requestId) { + + List rankedServers = _priorityPoolInstanceSelector.rank(ctx, allQueryCandidates); + + // If no stats are available yet (empty ranking), fall back to pool based round-robin. + if (rankedServers.isEmpty()) { + List groupIds = new ArrayList<>(replicaGroupToQueryServers.keySet()); + return groupIds.get(Math.abs(requestId) % groupIds.size()); + } + + Map serverRankMap = new HashMap<>(HashUtil.getHashMapCapacity(rankedServers.size())); + for (int i = 0; i < rankedServers.size(); i++) { + serverRankMap.put(rankedServers.get(i), i); + } + + // Pick the group with full coverage first, then best worst-case rank. + // getOrDefault guards against AdaptiveServerSelector implementations that may not return every + // submitted server — unranked servers get rank -1 (best), matching HybridSelector's convention. + // As of 8 July 2026, this fallback is unreachable, but new implementations could require it. + return replicaGroupToQueryServers.entrySet().stream() + .min(Comparator.>>comparingInt( + e -> hasFullCoverage(e.getKey(), replicaGroupSegmentCount, totalSegmentsWithCandidates) ? 0 : 1) + .thenComparingInt(e -> e.getValue().values().stream() + .mapToInt(c -> serverRankMap.getOrDefault(c.getInstance(), -1)) + .max().orElse(Integer.MAX_VALUE))) + .map(Map.Entry::getKey) + .orElse(-1); + } + + private boolean hasFullCoverage(int pool, Map replicaGroupSegmentCount, + int totalSegmentsWithCandidates) { + return replicaGroupSegmentCount.getOrDefault(pool, 0) >= totalSegmentsWithCandidates; + } + + /** + * Routes all segments to the chosen replica group. For each segment, picks the candidate whose + * idealStateReplicaId matches the selected replica group ID. If no candidate matches (transitional state during + * rebalancing), the segment is reported as unavailable. + */ + private InstanceMapping selectServersForReplicaGroup( + List segments, int replicaGroupId, SegmentStates segmentStates, Map queryOptions) { + + Map segmentToInstance = new HashMap<>(HashUtil.getHashMapCapacity(segments.size())); + Map optionalSegmentToInstance = new HashMap<>(); + List unavailableSegments = new ArrayList<>(); + Map poolToSegmentCount = new HashMap<>(); + + for (String segment : segments) { + List candidates = segmentStates.getCandidates(segment); + if (candidates == null) { + continue; + } + + SegmentInstanceCandidate selected = candidates.stream() + .filter(c -> c.getIdealStateReplicaId() == replicaGroupId) + .findFirst() + .orElse(null); + + // Skip segments with no candidate in the chosen replica group to preserve the strict same-replica-group + // invariant. We prefer to use groups that cover all queried segments, so this warning only occurs when we've + // selected a partial group. + if (selected == null) { + LOGGER.debug("No candidate found in replica group {} for segment {}; reporting as unavailable", + replicaGroupId, segment); + unavailableSegments.add(segment); + continue; + } + + if (selected.isOnline()) { + segmentToInstance.put(segment, selected.getInstance()); + } else { + optionalSegmentToInstance.put(segment, selected.getInstance()); + } + poolToSegmentCount.merge(selected.getPool(), 1, Integer::sum); + } + + // Emit per-pool POOL_SEG_QUERIES metric for observability, matching the parent's pattern. + for (Map.Entry entry : poolToSegmentCount.entrySet()) { + _brokerMetrics.addMeteredValue(BrokerMeter.POOL_SEG_QUERIES, entry.getValue(), + BrokerMetrics.getTagForPreferredPool(queryOptions), String.valueOf(entry.getKey())); + } + + return new InstanceMapping(segmentToInstance, optionalSegmentToInstance, unavailableSegments); + } } diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java index aae1fa8ef7a5..fb9908c4ec64 100644 --- a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java +++ b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java @@ -44,7 +44,6 @@ import javax.annotation.concurrent.GuardedBy; import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.lang3.StringUtils; -import org.apache.commons.lang3.tuple.Pair; import org.apache.helix.AccessOption; import org.apache.helix.BaseDataAccessor; import org.apache.helix.HelixConstants.ChangeType; @@ -768,6 +767,8 @@ private void buildRoutingInternal(String tableNameWithType) { AdaptiveServerSelector adaptiveServerSelector = AdaptiveServerSelectorFactory.getAdaptiveServerSelector(_serverRoutingStatsManager, _pinotConfig); + // For StrictReplicaGroupInstanceSelector tables, pool-level adaptive routing is used, + // preserving the same-replica-group guarantee while benefiting from adaptive server selection. InstanceSelector instanceSelector = InstanceSelectorFactory.getInstanceSelector(tableConfig, _propertyStore, _brokerMetrics, adaptiveServerSelector, _pinotConfig, _routableServerInstanceMap.keySet(), _enabledServerInstanceMap, @@ -1488,8 +1489,7 @@ InstanceSelector.SelectionResult calculateRouting(BrokerRequest brokerRequest, l selectionResult.setNumPrunedSegments(numPrunedSegments); return selectionResult; } else { - return new InstanceSelector.SelectionResult(Pair.of(Map.of(), Map.of()), - List.of(), numPrunedSegments); + return InstanceSelector.SelectionResult.empty(numPrunedSegments); } } diff --git a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelectorTest.java b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelectorTest.java index 03e35d251da4..1097556eeddc 100644 --- a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelectorTest.java +++ b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelectorTest.java @@ -34,13 +34,11 @@ import java.util.Set; import java.util.TreeMap; import java.util.concurrent.TimeUnit; -import org.apache.commons.lang3.tuple.ImmutablePair; import org.apache.commons.lang3.tuple.Pair; import org.apache.helix.model.ExternalView; import org.apache.helix.model.IdealState; import org.apache.helix.store.zk.ZkHelixPropertyStore; import org.apache.helix.zookeeper.datamodel.ZNRecord; -import org.apache.pinot.broker.routing.adaptiveserverselector.HybridSelector; import org.apache.pinot.common.metadata.ZKMetadataProvider; import org.apache.pinot.common.metadata.segment.SegmentZKMetadata; import org.apache.pinot.common.metrics.BrokerMetrics; @@ -103,13 +101,6 @@ public class InstanceSelectorTest { private static final String BALANCED_INSTANCE_SELECTOR = "balanced"; - private static List getSegments() { - return SEGMENTS; - } - - private final static List SEGMENTS = - Arrays.asList("segment0", "segment1", "segment2", "segment3", "segment4", "segment5", "segment6", "segment7", - "segment8", "segment9", "segment10", "segment11"); private static final long NEW_SEGMENT_EXPIRATION_SECONDS = 300; private static final long NEW_SEGMENT_EXPIRATION_MILLIS = TimeUnit.SECONDS.toMillis(NEW_SEGMENT_EXPIRATION_SECONDS); private final static InstanceSelectorConfig INSTANCE_SELECTOR_CONFIG = @@ -763,238 +754,6 @@ public void testInstanceSelector() { assertTrue(selectionResult.getUnavailableSegments().isEmpty()); } - @Test - public void testReplicaGroupInstanceSelectorNumReplicaGroupsToQuery() { - String offlineTableName = "testTable_OFFLINE"; - ZkHelixPropertyStore propertyStore = mock(ZkHelixPropertyStore.class); - BrokerMetrics brokerMetrics = mock(BrokerMetrics.class); - BrokerRequest brokerRequest = mock(BrokerRequest.class); - PinotQuery pinotQuery = mock(PinotQuery.class); - Map queryOptions = new HashMap<>(); - // numReplicas = 3, fanning the query to 2 replica groups - queryOptions.put("numReplicaGroupsToQuery", "2"); - when(brokerRequest.getPinotQuery()).thenReturn(pinotQuery); - when(pinotQuery.getQueryOptions()).thenReturn(queryOptions); - - ReplicaGroupInstanceSelector replicaGroupInstanceSelector = new ReplicaGroupInstanceSelector(); - - Set enabledInstances = new HashSet<>(); - IdealState idealState = new IdealState(offlineTableName); - Map> idealStateSegmentAssignment = idealState.getRecord().getMapFields(); - ExternalView externalView = new ExternalView(offlineTableName); - Map> externalViewSegmentAssignment = externalView.getRecord().getMapFields(); - Set onlineSegments = new HashSet<>(); - - // 12 online segments with each segment having all 3 instances as online - // replicas are 3 - String instance0 = "instance0"; - String instance1 = "instance1"; - String instance2 = "instance2"; - enabledInstances.add(instance0); - enabledInstances.add(instance1); - enabledInstances.add(instance2); - - Map idealStateInstanceStateMap0 = new TreeMap<>(); - Map externalViewInstanceStateMap0 = new TreeMap<>(); - - for (String instance : enabledInstances) { - idealStateInstanceStateMap0.put(instance, ONLINE); - externalViewInstanceStateMap0.put(instance, ONLINE); - } - - List segments = getSegments(); - // add all segments to both idealStateSegmentAssignment and externalViewSegmentAssignment maps and also to online - // segments - for (String segment : segments) { - idealStateSegmentAssignment.put(segment, idealStateInstanceStateMap0); - externalViewSegmentAssignment.put(segment, externalViewInstanceStateMap0); - onlineSegments.add(segment); - } - - replicaGroupInstanceSelector.init(_tableConfig, propertyStore, brokerMetrics, null, Clock.systemUTC(), - INSTANCE_SELECTOR_CONFIG, enabledInstances, EMPTY_SERVER_MAP, idealState, externalView, onlineSegments); - // ReplicaGroupInstanceSelector - // segment0 -> instance0 - // segment2 -> instance0 - // segment4 -> instance0 - // segment6 -> instance0 - // segment8 -> instance0 - // segment10 -> instance0 - // segment1 -> instance1 - // segment3 -> instance1 - // segment5 -> instance1 - // segment7 -> instance1 - // segment9 -> instance1 - // segment11 -> instance1 - - Map expectedReplicaGroupInstanceSelectorResult = new HashMap<>(); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(0), instance0); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(1), instance1); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(2), instance0); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(3), instance1); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(4), instance0); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(5), instance1); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(6), instance0); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(7), instance1); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(8), instance0); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(9), instance1); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(10), instance0); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(11), instance1); - InstanceSelector.SelectionResult selectionResult = replicaGroupInstanceSelector.select(brokerRequest, segments, 0); - assertEquals(selectionResult.getSegmentToInstanceMap(), expectedReplicaGroupInstanceSelectorResult); - assertTrue(selectionResult.getUnavailableSegments().isEmpty()); - } - - @Test - public void testReplicaGroupInstanceSelectorNumReplicaGroupsToQueryGreaterThanReplicas() { - String offlineTableName = "testTable_OFFLINE"; - ZkHelixPropertyStore propertyStore = mock(ZkHelixPropertyStore.class); - BrokerMetrics brokerMetrics = mock(BrokerMetrics.class); - BrokerRequest brokerRequest = mock(BrokerRequest.class); - PinotQuery pinotQuery = mock(PinotQuery.class); - Map queryOptions = new HashMap<>(); - queryOptions.put("numReplicaGroupsToQuery", "4"); - - when(brokerRequest.getPinotQuery()).thenReturn(pinotQuery); - when(pinotQuery.getQueryOptions()).thenReturn(queryOptions); - - ReplicaGroupInstanceSelector replicaGroupInstanceSelector = new ReplicaGroupInstanceSelector(); - - Set enabledInstances = new HashSet<>(); - IdealState idealState = new IdealState(offlineTableName); - Map> idealStateSegmentAssignment = idealState.getRecord().getMapFields(); - ExternalView externalView = new ExternalView(offlineTableName); - Map> externalViewSegmentAssignment = externalView.getRecord().getMapFields(); - Set onlineSegments = new HashSet<>(); - - // 12 online segments with each segment having all 3 instances as online - // replicas are 3 - String instance0 = "instance0"; - String instance1 = "instance1"; - String instance2 = "instance2"; - enabledInstances.add(instance0); - enabledInstances.add(instance1); - enabledInstances.add(instance2); - - List segments = getSegments(); - - Map idealStateInstanceStateMap0 = new TreeMap<>(); - Map externalViewInstanceStateMap0 = new TreeMap<>(); - - for (String instance : enabledInstances) { - idealStateInstanceStateMap0.put(instance, ONLINE); - externalViewInstanceStateMap0.put(instance, ONLINE); - } - - // add all segments to both idealStateSegmentAssignment and externalViewSegmentAssignment maps and also to online - // segments - for (String segment : segments) { - idealStateSegmentAssignment.put(segment, idealStateInstanceStateMap0); - externalViewSegmentAssignment.put(segment, externalViewInstanceStateMap0); - onlineSegments.add(segment); - } - - replicaGroupInstanceSelector.init(_tableConfig, propertyStore, brokerMetrics, null, Clock.systemUTC(), - INSTANCE_SELECTOR_CONFIG, enabledInstances, EMPTY_SERVER_MAP, idealState, externalView, onlineSegments); - // ReplicaGroupInstanceSelector - // segment0 -> instance0 - // segment3 -> instance0 - // segment6 -> instance0 - // segment9 -> instance0 - // segment1 -> instance1 - // segment4 -> instance1 - // segment7 -> instance1 - // segment10 -> instance1 - // segment2 -> instance2 - // segment5 -> instance2 - // segment8 -> instance2 - // segment11 -> instance2 - - Map expectedReplicaGroupInstanceSelectorResult = new HashMap<>(); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(0), instance0); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(1), instance1); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(2), instance2); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(3), instance0); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(4), instance1); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(5), instance2); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(6), instance0); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(7), instance1); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(8), instance2); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(9), instance0); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(10), instance1); - expectedReplicaGroupInstanceSelectorResult.put(segments.get(11), instance2); - InstanceSelector.SelectionResult selectionResult = replicaGroupInstanceSelector.select(brokerRequest, segments, 0); - assertEquals(selectionResult.getSegmentToInstanceMap(), expectedReplicaGroupInstanceSelectorResult); - assertTrue(selectionResult.getUnavailableSegments().isEmpty()); - } - - @Test - public void testReplicaGroupInstanceSelectorNumReplicaGroupsNotSet() { - String offlineTableName = "testTable_OFFLINE"; - ZkHelixPropertyStore propertyStore = mock(ZkHelixPropertyStore.class); - BrokerMetrics brokerMetrics = mock(BrokerMetrics.class); - BrokerRequest brokerRequest = mock(BrokerRequest.class); - PinotQuery pinotQuery = mock(PinotQuery.class); - Map queryOptions = new HashMap<>(); - - when(brokerRequest.getPinotQuery()).thenReturn(pinotQuery); - when(pinotQuery.getQueryOptions()).thenReturn(queryOptions); - - ReplicaGroupInstanceSelector replicaGroupInstanceSelector = new ReplicaGroupInstanceSelector(); - - Set enabledInstances = new HashSet<>(); - IdealState idealState = new IdealState(offlineTableName); - Map> idealStateSegmentAssignment = idealState.getRecord().getMapFields(); - ExternalView externalView = new ExternalView(offlineTableName); - Map> externalViewSegmentAssignment = externalView.getRecord().getMapFields(); - Set onlineSegments = new HashSet<>(); - - // 12 online segments with each segment having all 3 instances as online - // replicas are 3 - String instance0 = "instance0"; - String instance1 = "instance1"; - String instance2 = "instance2"; - enabledInstances.add(instance0); - enabledInstances.add(instance1); - enabledInstances.add(instance2); - - List segments = getSegments(); - - Map idealStateInstanceStateMap0 = new TreeMap<>(); - Map externalViewInstanceStateMap0 = new TreeMap<>(); - - for (String instance : enabledInstances) { - idealStateInstanceStateMap0.put(instance, ONLINE); - externalViewInstanceStateMap0.put(instance, ONLINE); - } - - // add all segments to both idealStateSegmentAssignment and externalViewSegmentAssignment maps and also to online - // segments - for (String segment : segments) { - idealStateSegmentAssignment.put(segment, idealStateInstanceStateMap0); - externalViewSegmentAssignment.put(segment, externalViewInstanceStateMap0); - onlineSegments.add(segment); - } - - replicaGroupInstanceSelector.init(_tableConfig, propertyStore, brokerMetrics, null, Clock.systemUTC(), - INSTANCE_SELECTOR_CONFIG, enabledInstances, EMPTY_SERVER_MAP, idealState, externalView, onlineSegments); - // since numReplicaGroupsToQuery is not set, first query should go to first replica group, - // 2nd query should go to next replica group - - Map expectedReplicaGroupInstanceSelectorResult = new HashMap<>(); - for (String segment : segments) { - expectedReplicaGroupInstanceSelectorResult.put(segment, instance0); - } - InstanceSelector.SelectionResult selectionResult = replicaGroupInstanceSelector.select(brokerRequest, segments, 0); - assertEquals(selectionResult.getSegmentToInstanceMap(), expectedReplicaGroupInstanceSelectorResult); - - for (String segment : segments) { - expectedReplicaGroupInstanceSelectorResult.put(segment, instance1); - } - selectionResult = replicaGroupInstanceSelector.select(brokerRequest, segments, 1); - assertEquals(selectionResult.getSegmentToInstanceMap(), expectedReplicaGroupInstanceSelectorResult); - } - @Test public void testUnavailableSegments() { String offlineTableName = "testTable_OFFLINE"; @@ -1857,74 +1616,4 @@ public void testExcludeIdealStateOffline(String selectorType) { assertEquals(selectionResult.getSegmentToInstanceMap(), expectedBalancedInstanceSelectorResult); assertTrue(selectionResult.getUnavailableSegments().isEmpty()); } - - @Test - public void testReplicaGroupAdaptiveServerSelector() { - // Arrange - ZkHelixPropertyStore propertyStore = mock(ZkHelixPropertyStore.class); - BrokerMetrics brokerMetrics = mock(BrokerMetrics.class); - HybridSelector hybridSelector = mock(HybridSelector.class); - ReplicaGroupInstanceSelector instanceSelector = new ReplicaGroupInstanceSelector(); - - // Define instances and segments - String instance0 = "instance0"; - String instance1 = "instance1"; - String instance2 = "instance2"; - String instance3 = "instance3"; - String instance4 = "instance4"; - String segment0 = "segment0"; - String segment1 = "segment1"; - String segment2 = "segment2"; - List segments = Arrays.asList(segment0, segment1, segment2); - - // Define candidates for each segment - Map> instanceCandidatesMap = new HashMap<>(); - // segment0 -> instance0, instance1 - instanceCandidatesMap.put(segment0, - Arrays.asList(new SegmentInstanceCandidate(instance0, true), new SegmentInstanceCandidate(instance1, true))); - // segment1 -> instance2, instance3 - instanceCandidatesMap.put(segment1, - Arrays.asList(new SegmentInstanceCandidate(instance2, true), new SegmentInstanceCandidate(instance3, true))); - // segment2 -> instance3, instance4 // instance4 is not in the hybrid selector's server ranking - instanceCandidatesMap.put(segment2, - Arrays.asList(new SegmentInstanceCandidate(instance4, true), new SegmentInstanceCandidate(instance3, true))); - - IdealState idealState = createIdealState( - Map.of(segment0, List.of(Pair.of(instance0, ONLINE), Pair.of(instance1, ONLINE)), segment1, - List.of(Pair.of(instance2, ONLINE), Pair.of(instance3, ONLINE)), segment2, - List.of(Pair.of(instance3, ONLINE), Pair.of(instance4, ONLINE)))); - - ExternalView externalView = createExternalView( - Map.of(segment0, List.of(Pair.of(instance0, ONLINE), Pair.of(instance1, ONLINE)), segment1, - List.of(Pair.of(instance2, ONLINE), Pair.of(instance3, ONLINE)), segment2, - List.of(Pair.of(instance3, ONLINE), Pair.of(instance4, ONLINE)))); - - instanceSelector.init(_tableConfig, propertyStore, brokerMetrics, hybridSelector, Clock.systemUTC(), - INSTANCE_SELECTOR_CONFIG, Set.of(instance0, instance1, instance2, instance3, instance4), EMPTY_SERVER_MAP, - idealState, externalView, new HashSet<>(segments)); - - // Define the segment states - SegmentStates segmentStates = new SegmentStates(instanceCandidatesMap, new HashSet<>(segments), null); - - // Define server rankings - List> serverRanks = Arrays.asList( - new ImmutablePair<>(instance3, 1.0), - new ImmutablePair<>(instance2, 2.0), - new ImmutablePair<>(instance1, 3.0), - new ImmutablePair<>(instance0, 4.0) - ); - when(hybridSelector.fetchServerRankingsWithScores(any())).thenReturn(serverRanks); - - // Act - Pair, Map> selectedResult = - instanceSelector.select(segments, 0, segmentStates, null); - - // Assert - Map expectedSelection = new HashMap<>(); - expectedSelection.put(segment0, instance1); - expectedSelection.put(segment1, instance3); - expectedSelection.put(segment2, instance4); - - assertEquals(selectedResult.getLeft(), expectedSelection); - } } diff --git a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/instanceselector/ReplicaGroupSelectorTest.java b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/instanceselector/ReplicaGroupSelectorTest.java new file mode 100644 index 000000000000..68cf4b33a161 --- /dev/null +++ b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/instanceselector/ReplicaGroupSelectorTest.java @@ -0,0 +1,913 @@ +/** + * 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.broker.routing.instanceselector; + +import java.time.Clock; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeMap; +import org.apache.commons.lang3.tuple.ImmutablePair; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.helix.model.ExternalView; +import org.apache.helix.model.IdealState; +import org.apache.helix.store.zk.ZkHelixPropertyStore; +import org.apache.helix.zookeeper.datamodel.ZNRecord; +import org.apache.pinot.broker.routing.adaptiveserverselector.HybridSelector; +import org.apache.pinot.common.metrics.BrokerMeter; +import org.apache.pinot.common.metrics.BrokerMetrics; +import org.apache.pinot.common.request.BrokerRequest; +import org.apache.pinot.common.request.PinotQuery; +import org.apache.pinot.core.transport.ServerInstance; +import org.apache.pinot.spi.config.table.RoutingConfig; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.env.PinotConfiguration; +import org.apache.pinot.spi.utils.CommonConstants; +import org.apache.pinot.spi.utils.builder.TableConfigBuilder; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import static org.apache.pinot.spi.config.table.RoutingConfig.REPLICA_GROUP_INSTANCE_SELECTOR_TYPE; +import static org.apache.pinot.spi.config.table.RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE; +import static org.apache.pinot.spi.utils.CommonConstants.Broker.FALLBACK_POOL_ID; +import static org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.SegmentStateModel.ONLINE; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertNull; +import static org.testng.Assert.assertTrue; + + +/** + * Tests for {@link ReplicaGroupInstanceSelector} and {@link StrictReplicaGroupInstanceSelector}, + * including adaptive server selection (replica-group-level routing for strict replica groups). + */ +@SuppressWarnings("unchecked") +public class ReplicaGroupSelectorTest { + private AutoCloseable _mocks; + + @Mock + private TableConfig _tableConfig; + + private static final String TABLE_NAME = "testTable_OFFLINE"; + private static final Map EMPTY_SERVER_MAP = Collections.EMPTY_MAP; + private static final InstanceSelectorConfig INSTANCE_SELECTOR_CONFIG = new InstanceSelectorConfig(false, 300, false); + private static final List SEGMENTS = + Arrays.asList("segment0", "segment1", "segment2", "segment3", "segment4", "segment5", "segment6", "segment7", + "segment8", "segment9", "segment10", "segment11"); + + @BeforeMethod + public void setUp() { + _mocks = MockitoAnnotations.openMocks(this); + when(_tableConfig.getTableName()).thenReturn(TABLE_NAME); + } + + @AfterMethod + public void tearDown() + throws Exception { + _mocks.close(); + } + + // --- Shared helpers --- + + static IdealState createIdealState(Map>> segmentState) { + IdealState idealState = new IdealState(TABLE_NAME); + Map> idealStateSegmentAssignment = idealState.getRecord().getMapFields(); + for (Map.Entry>> entry : segmentState.entrySet()) { + Map instanceStateMap = new TreeMap<>(); + for (Pair instanceState : entry.getValue()) { + instanceStateMap.put(instanceState.getLeft(), instanceState.getRight()); + } + idealStateSegmentAssignment.put(entry.getKey(), instanceStateMap); + } + return idealState; + } + + static ExternalView createExternalView(Map>> segmentState) { + ExternalView externalView = new ExternalView(TABLE_NAME); + Map> externalViewSegmentAssignment = externalView.getRecord().getMapFields(); + for (Map.Entry>> entry : segmentState.entrySet()) { + Map instanceStateMap = new TreeMap<>(); + for (Pair instanceState : entry.getValue()) { + instanceStateMap.put(instanceState.getLeft(), instanceState.getRight()); + } + externalViewSegmentAssignment.put(entry.getKey(), instanceStateMap); + } + return externalView; + } + + // --- ReplicaGroupInstanceSelector: numReplicaGroupsToQuery tests --- + + @Test + public void testReplicaGroupInstanceSelectorNumReplicaGroupsToQuery() { + String offlineTableName = "testTable_OFFLINE"; + ZkHelixPropertyStore propertyStore = mock(ZkHelixPropertyStore.class); + BrokerMetrics brokerMetrics = mock(BrokerMetrics.class); + BrokerRequest brokerRequest = mock(BrokerRequest.class); + PinotQuery pinotQuery = mock(PinotQuery.class); + Map queryOptions = new HashMap<>(); + // numReplicas = 3, fanning the query to 2 replica groups + queryOptions.put("numReplicaGroupsToQuery", "2"); + when(brokerRequest.getPinotQuery()).thenReturn(pinotQuery); + when(pinotQuery.getQueryOptions()).thenReturn(queryOptions); + + ReplicaGroupInstanceSelector replicaGroupInstanceSelector = new ReplicaGroupInstanceSelector(); + + Set enabledInstances = new HashSet<>(); + IdealState idealState = new IdealState(offlineTableName); + Map> idealStateSegmentAssignment = idealState.getRecord().getMapFields(); + ExternalView externalView = new ExternalView(offlineTableName); + Map> externalViewSegmentAssignment = externalView.getRecord().getMapFields(); + Set onlineSegments = new HashSet<>(); + + // 12 online segments with each segment having all 3 instances as online + // replicas are 3 + String instance0 = "instance0"; + String instance1 = "instance1"; + String instance2 = "instance2"; + enabledInstances.add(instance0); + enabledInstances.add(instance1); + enabledInstances.add(instance2); + + Map idealStateInstanceStateMap0 = new TreeMap<>(); + Map externalViewInstanceStateMap0 = new TreeMap<>(); + + for (String instance : enabledInstances) { + idealStateInstanceStateMap0.put(instance, ONLINE); + externalViewInstanceStateMap0.put(instance, ONLINE); + } + + List segments = SEGMENTS; + // add all segments to both idealStateSegmentAssignment and externalViewSegmentAssignment maps and also to online + // segments + for (String segment : segments) { + idealStateSegmentAssignment.put(segment, idealStateInstanceStateMap0); + externalViewSegmentAssignment.put(segment, externalViewInstanceStateMap0); + onlineSegments.add(segment); + } + + replicaGroupInstanceSelector.init(_tableConfig, propertyStore, brokerMetrics, null, Clock.systemUTC(), + INSTANCE_SELECTOR_CONFIG, enabledInstances, EMPTY_SERVER_MAP, idealState, externalView, onlineSegments); + + Map expectedReplicaGroupInstanceSelectorResult = new HashMap<>(); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(0), instance0); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(1), instance1); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(2), instance0); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(3), instance1); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(4), instance0); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(5), instance1); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(6), instance0); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(7), instance1); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(8), instance0); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(9), instance1); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(10), instance0); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(11), instance1); + InstanceSelector.SelectionResult selectionResult = replicaGroupInstanceSelector.select(brokerRequest, segments, 0); + assertEquals(selectionResult.getSegmentToInstanceMap(), expectedReplicaGroupInstanceSelectorResult); + assertTrue(selectionResult.getUnavailableSegments().isEmpty()); + } + + @Test + public void testReplicaGroupInstanceSelectorNumReplicaGroupsToQueryGreaterThanReplicas() { + String offlineTableName = "testTable_OFFLINE"; + ZkHelixPropertyStore propertyStore = mock(ZkHelixPropertyStore.class); + BrokerMetrics brokerMetrics = mock(BrokerMetrics.class); + BrokerRequest brokerRequest = mock(BrokerRequest.class); + PinotQuery pinotQuery = mock(PinotQuery.class); + Map queryOptions = new HashMap<>(); + queryOptions.put("numReplicaGroupsToQuery", "4"); + + when(brokerRequest.getPinotQuery()).thenReturn(pinotQuery); + when(pinotQuery.getQueryOptions()).thenReturn(queryOptions); + + ReplicaGroupInstanceSelector replicaGroupInstanceSelector = new ReplicaGroupInstanceSelector(); + + Set enabledInstances = new HashSet<>(); + IdealState idealState = new IdealState(offlineTableName); + Map> idealStateSegmentAssignment = idealState.getRecord().getMapFields(); + ExternalView externalView = new ExternalView(offlineTableName); + Map> externalViewSegmentAssignment = externalView.getRecord().getMapFields(); + Set onlineSegments = new HashSet<>(); + + // 12 online segments with each segment having all 3 instances as online + // replicas are 3 + String instance0 = "instance0"; + String instance1 = "instance1"; + String instance2 = "instance2"; + enabledInstances.add(instance0); + enabledInstances.add(instance1); + enabledInstances.add(instance2); + + List segments = SEGMENTS; + + Map idealStateInstanceStateMap0 = new TreeMap<>(); + Map externalViewInstanceStateMap0 = new TreeMap<>(); + + for (String instance : enabledInstances) { + idealStateInstanceStateMap0.put(instance, ONLINE); + externalViewInstanceStateMap0.put(instance, ONLINE); + } + + // add all segments to both idealStateSegmentAssignment and externalViewSegmentAssignment maps and also to online + // segments + for (String segment : segments) { + idealStateSegmentAssignment.put(segment, idealStateInstanceStateMap0); + externalViewSegmentAssignment.put(segment, externalViewInstanceStateMap0); + onlineSegments.add(segment); + } + + replicaGroupInstanceSelector.init(_tableConfig, propertyStore, brokerMetrics, null, Clock.systemUTC(), + INSTANCE_SELECTOR_CONFIG, enabledInstances, EMPTY_SERVER_MAP, idealState, externalView, onlineSegments); + + Map expectedReplicaGroupInstanceSelectorResult = new HashMap<>(); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(0), instance0); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(1), instance1); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(2), instance2); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(3), instance0); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(4), instance1); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(5), instance2); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(6), instance0); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(7), instance1); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(8), instance2); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(9), instance0); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(10), instance1); + expectedReplicaGroupInstanceSelectorResult.put(segments.get(11), instance2); + InstanceSelector.SelectionResult selectionResult = replicaGroupInstanceSelector.select(brokerRequest, segments, 0); + assertEquals(selectionResult.getSegmentToInstanceMap(), expectedReplicaGroupInstanceSelectorResult); + assertTrue(selectionResult.getUnavailableSegments().isEmpty()); + } + + @Test + public void testReplicaGroupInstanceSelectorNumReplicaGroupsNotSet() { + String offlineTableName = "testTable_OFFLINE"; + ZkHelixPropertyStore propertyStore = mock(ZkHelixPropertyStore.class); + BrokerMetrics brokerMetrics = mock(BrokerMetrics.class); + BrokerRequest brokerRequest = mock(BrokerRequest.class); + PinotQuery pinotQuery = mock(PinotQuery.class); + Map queryOptions = new HashMap<>(); + + when(brokerRequest.getPinotQuery()).thenReturn(pinotQuery); + when(pinotQuery.getQueryOptions()).thenReturn(queryOptions); + + ReplicaGroupInstanceSelector replicaGroupInstanceSelector = new ReplicaGroupInstanceSelector(); + + Set enabledInstances = new HashSet<>(); + IdealState idealState = new IdealState(offlineTableName); + Map> idealStateSegmentAssignment = idealState.getRecord().getMapFields(); + ExternalView externalView = new ExternalView(offlineTableName); + Map> externalViewSegmentAssignment = externalView.getRecord().getMapFields(); + Set onlineSegments = new HashSet<>(); + + // 12 online segments with each segment having all 3 instances as online + // replicas are 3 + String instance0 = "instance0"; + String instance1 = "instance1"; + String instance2 = "instance2"; + enabledInstances.add(instance0); + enabledInstances.add(instance1); + enabledInstances.add(instance2); + + List segments = SEGMENTS; + + Map idealStateInstanceStateMap0 = new TreeMap<>(); + Map externalViewInstanceStateMap0 = new TreeMap<>(); + + for (String instance : enabledInstances) { + idealStateInstanceStateMap0.put(instance, ONLINE); + externalViewInstanceStateMap0.put(instance, ONLINE); + } + + // add all segments to both idealStateSegmentAssignment and externalViewSegmentAssignment maps and also to online + // segments + for (String segment : segments) { + idealStateSegmentAssignment.put(segment, idealStateInstanceStateMap0); + externalViewSegmentAssignment.put(segment, externalViewInstanceStateMap0); + onlineSegments.add(segment); + } + + replicaGroupInstanceSelector.init(_tableConfig, propertyStore, brokerMetrics, null, Clock.systemUTC(), + INSTANCE_SELECTOR_CONFIG, enabledInstances, EMPTY_SERVER_MAP, idealState, externalView, onlineSegments); + // since numReplicaGroupsToQuery is not set, first query should go to first replica group, + // 2nd query should go to next replica group + + Map expectedReplicaGroupInstanceSelectorResult = new HashMap<>(); + for (String segment : segments) { + expectedReplicaGroupInstanceSelectorResult.put(segment, instance0); + } + InstanceSelector.SelectionResult selectionResult = replicaGroupInstanceSelector.select(brokerRequest, segments, 0); + assertEquals(selectionResult.getSegmentToInstanceMap(), expectedReplicaGroupInstanceSelectorResult); + + for (String segment : segments) { + expectedReplicaGroupInstanceSelectorResult.put(segment, instance1); + } + selectionResult = replicaGroupInstanceSelector.select(brokerRequest, segments, 1); + assertEquals(selectionResult.getSegmentToInstanceMap(), expectedReplicaGroupInstanceSelectorResult); + } + + // --- Adaptive server selection tests --- + + // Shared topology for AR tests: 3 segments across 5 instances in two pools / replica groups. + // segment2 intentionally has AR_P0_RG0_SERVER_E (unranked) so AR falls back to round-robin for that segment. + private static final String AR_P0_RG0_SERVER_A = "ar_p0_rg0_server_a"; + private static final String AR_P1_RG1_SERVER_B = "ar_p1_rg1_server_b"; + private static final String AR_P0_RG0_SERVER_C = "ar_p0_rg0_server_c"; + private static final String AR_P1_RG1_SERVER_D = "ar_p1_rg1_server_d"; + private static final String AR_P0_RG0_SERVER_E = "ar_p0_rg0_server_e"; + private static final String AR_SEGMENT0 = "segment0"; + private static final String AR_SEGMENT1 = "segment1"; + private static final String AR_SEGMENT2 = "segment2"; + private static final List AR_SEGMENTS = + Arrays.asList(AR_SEGMENT0, AR_SEGMENT1, AR_SEGMENT2); + // Rankings: D best → C → B → A worst; E absent (triggers AR fallback) + private static final List> AR_SERVER_RANKS = Arrays.asList( + new ImmutablePair<>(AR_P1_RG1_SERVER_D, 1.0), + new ImmutablePair<>(AR_P0_RG0_SERVER_C, 2.0), + new ImmutablePair<>(AR_P1_RG1_SERVER_B, 3.0), + new ImmutablePair<>(AR_P0_RG0_SERVER_A, 4.0) + ); + + private SegmentStates buildArSegmentStates() { + Map> candidatesMap = new HashMap<>(); + // segment0 → pool 0 / replica group 0 server A, pool 1 / replica group 1 server B + candidatesMap.put(AR_SEGMENT0, Arrays.asList( + new SegmentInstanceCandidate(AR_P0_RG0_SERVER_A, true, 0, 0), + new SegmentInstanceCandidate(AR_P1_RG1_SERVER_B, true, 1, 1))); + // segment1 → pool 0 / replica group 0 server C, pool 1 / replica group 1 server D + candidatesMap.put(AR_SEGMENT1, Arrays.asList( + new SegmentInstanceCandidate(AR_P0_RG0_SERVER_C, true, 0, 0), + new SegmentInstanceCandidate(AR_P1_RG1_SERVER_D, true, 1, 1))); + // segment2 → pool 0 / replica group 0 server E (unranked), pool 1 / replica group 1 server D + candidatesMap.put(AR_SEGMENT2, Arrays.asList( + new SegmentInstanceCandidate(AR_P0_RG0_SERVER_E, true, 0, 0), + new SegmentInstanceCandidate(AR_P1_RG1_SERVER_D, true, 1, 1))); + return new SegmentStates(candidatesMap, new HashSet<>(AR_SEGMENTS), null); + } + + private ReplicaGroupInstanceSelector buildArSelector(String selectorType, HybridSelector hybridSelector) { + return buildArSelector(selectorType, hybridSelector, new PinotConfiguration(Map.of())); + } + + private ReplicaGroupInstanceSelector buildArSelector(String selectorType, HybridSelector hybridSelector, + PinotConfiguration brokerConfig) { + RoutingConfig routingConfig = new RoutingConfig(null, null, selectorType, false); + TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName("testTable") + .setRoutingConfig(routingConfig).build(); + IdealState idealState = createIdealState(Map.of( + AR_SEGMENT0, List.of(Pair.of(AR_P0_RG0_SERVER_A, ONLINE), Pair.of(AR_P1_RG1_SERVER_B, ONLINE)), + AR_SEGMENT1, List.of(Pair.of(AR_P0_RG0_SERVER_C, ONLINE), Pair.of(AR_P1_RG1_SERVER_D, ONLINE)), + AR_SEGMENT2, List.of(Pair.of(AR_P1_RG1_SERVER_D, ONLINE), Pair.of(AR_P0_RG0_SERVER_E, ONLINE)))); + ExternalView externalView = createExternalView(Map.of( + AR_SEGMENT0, List.of(Pair.of(AR_P0_RG0_SERVER_A, ONLINE), Pair.of(AR_P1_RG1_SERVER_B, ONLINE)), + AR_SEGMENT1, List.of(Pair.of(AR_P0_RG0_SERVER_C, ONLINE), Pair.of(AR_P1_RG1_SERVER_D, ONLINE)), + AR_SEGMENT2, List.of(Pair.of(AR_P1_RG1_SERVER_D, ONLINE), Pair.of(AR_P0_RG0_SERVER_E, ONLINE)))); + ServerInstance serverA = mock(ServerInstance.class); + when(serverA.getPool()).thenReturn(0); + ServerInstance serverB = mock(ServerInstance.class); + when(serverB.getPool()).thenReturn(1); + ServerInstance serverC = mock(ServerInstance.class); + when(serverC.getPool()).thenReturn(0); + ServerInstance serverD = mock(ServerInstance.class); + when(serverD.getPool()).thenReturn(1); + ServerInstance serverE = mock(ServerInstance.class); + when(serverE.getPool()).thenReturn(0); + Map serverMap = Map.of( + AR_P0_RG0_SERVER_A, serverA, + AR_P1_RG1_SERVER_B, serverB, + AR_P0_RG0_SERVER_C, serverC, + AR_P1_RG1_SERVER_D, serverD, + AR_P0_RG0_SERVER_E, serverE); + return (ReplicaGroupInstanceSelector) InstanceSelectorFactory.getInstanceSelector(tableConfig, + mock(ZkHelixPropertyStore.class), mock(BrokerMetrics.class), hybridSelector, + brokerConfig, + Set.of(AR_P0_RG0_SERVER_A, AR_P1_RG1_SERVER_B, AR_P0_RG0_SERVER_C, AR_P1_RG1_SERVER_D, AR_P0_RG0_SERVER_E), + serverMap, idealState, externalView, new HashSet<>(AR_SEGMENTS)); + } + + @Test + public void testReplicaGroupAdaptiveServerSelector() { + HybridSelector hybridSelector = mock(HybridSelector.class); + ReplicaGroupInstanceSelector instanceSelector = + buildArSelector(REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, hybridSelector); + + assertTrue(instanceSelector instanceof ReplicaGroupInstanceSelector); + assertFalse(instanceSelector instanceof StrictReplicaGroupInstanceSelector); + assertNotNull(instanceSelector._adaptiveServerSelector); + assertNotNull(instanceSelector._priorityPoolInstanceSelector); + + when(hybridSelector.fetchServerRankingsWithScores(any())).thenReturn(AR_SERVER_RANKS); + InstanceSelector.InstanceMapping result = + instanceSelector.select(AR_SEGMENTS, 0, buildArSegmentStates(), null); + + // AR prefers the better-ranked server when all candidates are ranked: + // segment0: B (rank 3) over A (rank 4) + // segment1: D (rank 1) over C (rank 2) + // segment2: E unranked → AR falls back to round-robin → index 0 = E + assertEquals(result.segmentToInstanceMap(), Map.of( + AR_SEGMENT0, AR_P1_RG1_SERVER_B, + AR_SEGMENT1, AR_P1_RG1_SERVER_D, + AR_SEGMENT2, AR_P0_RG0_SERVER_E)); + } + + @Test + public void testStrictReplicaGroupAdaptiveDisabledByFeatureFlag() { + // When the feature flag is disabled, the factory nulls out the adaptive selector so the selector falls back to the + // strict selector's non-adaptive, round-robin-by-replica-index path. + HybridSelector hybridSelector = mock(HybridSelector.class); + BrokerMetrics brokerMetrics = mock(BrokerMetrics.class); + PinotConfiguration brokerConfig = new PinotConfiguration(Map.of( + CommonConstants.Broker.AdaptiveServerSelector.CONFIG_OF_ENABLE_STRICT_REPLICA_GROUP, "false")); + StrictReplicaGroupInstanceSelector instanceSelector = + buildStrictReplicaGroupArSelector(hybridSelector, brokerMetrics, brokerConfig); + + assertNull(instanceSelector._adaptiveServerSelector); + assertNull(instanceSelector._priorityPoolInstanceSelector); + + BaseInstanceSelector.InstanceMapping result = + instanceSelector.select(STRICT_SEGMENTS, 0, buildStrictReplicaGroupSegmentStates(), null); + + verify(hybridSelector, never()).fetchServerRankingsWithScores(any()); + assertEquals(result.segmentToInstanceMap(), Map.of( + STRICT_SEGMENT0, STRICT_RG0_SERVER_A, + STRICT_SEGMENT1, STRICT_RG0_SERVER_A, + STRICT_SEGMENT2, STRICT_RG0_SERVER_B)); + + result = instanceSelector.select(STRICT_SEGMENTS, 1, buildStrictReplicaGroupSegmentStates(), null); + assertEquals(result.segmentToInstanceMap(), Map.of( + STRICT_SEGMENT0, STRICT_RG1_SERVER_C, + STRICT_SEGMENT1, STRICT_RG1_SERVER_C, + STRICT_SEGMENT2, STRICT_RG1_SERVER_D)); + } + + @Test + public void testStrictReplicaGroupFactoryEnablesAdaptiveRouting() { + HybridSelector hybridSelector = mock(HybridSelector.class); + ReplicaGroupInstanceSelector instanceSelector = + buildArSelector(STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, hybridSelector); + + assertTrue(instanceSelector instanceof StrictReplicaGroupInstanceSelector); + assertNotNull(instanceSelector._adaptiveServerSelector); + assertNotNull(instanceSelector._priorityPoolInstanceSelector); + } + + // --- Strict replica group adaptive routing tests --- + + // Topology: 2 replica groups, 3 segments + // Replica group 0: server_a (segment0, segment1), server_b (segment2) + // Replica group 1: server_c (segment0, segment1), server_d (segment2) + // If a strict-fixture server name omits `P#`, its pool is FALLBACK_POOL_ID. + private static final String STRICT_RG0_SERVER_A = "strict_rg0_server_a"; + private static final String STRICT_RG0_SERVER_B = "strict_rg0_server_b"; + private static final String STRICT_RG1_SERVER_C = "strict_rg1_server_c"; + private static final String STRICT_RG1_SERVER_D = "strict_rg1_server_d"; + private static final String STRICT_SEGMENT0 = "seg0"; + private static final String STRICT_SEGMENT1 = "seg1"; + private static final String STRICT_SEGMENT2 = "seg2"; + private static final List STRICT_SEGMENTS = Arrays.asList(STRICT_SEGMENT0, STRICT_SEGMENT1, STRICT_SEGMENT2); + + private List buildStrictReplicaGroupCandidates(String replicaGroup0Instance, + String replicaGroup1Instance) { + return Arrays.asList( + new SegmentInstanceCandidate(replicaGroup0Instance, true, FALLBACK_POOL_ID, 0), + new SegmentInstanceCandidate(replicaGroup1Instance, true, FALLBACK_POOL_ID, 1)); + } + + private SegmentStates buildStrictReplicaGroupSegmentStates() { + Map> candidatesMap = new HashMap<>(); + // All strict-fixture servers intentionally share FALLBACK_POOL_ID so these tests fail if strict routing regresses + // back to grouping or filtering by pool instead of idealStateReplicaId. + candidatesMap.put(STRICT_SEGMENT0, + buildStrictReplicaGroupCandidates(STRICT_RG0_SERVER_A, STRICT_RG1_SERVER_C)); + candidatesMap.put(STRICT_SEGMENT1, + buildStrictReplicaGroupCandidates(STRICT_RG0_SERVER_A, STRICT_RG1_SERVER_C)); + candidatesMap.put(STRICT_SEGMENT2, + buildStrictReplicaGroupCandidates(STRICT_RG0_SERVER_B, STRICT_RG1_SERVER_D)); + return new SegmentStates(candidatesMap, new HashSet<>(STRICT_SEGMENTS), null); + } + + private StrictReplicaGroupInstanceSelector buildStrictReplicaGroupArSelector(HybridSelector hybridSelector) { + return buildStrictReplicaGroupArSelector(hybridSelector, mock(BrokerMetrics.class)); + } + + private StrictReplicaGroupInstanceSelector buildStrictReplicaGroupArSelector(HybridSelector hybridSelector, + BrokerMetrics brokerMetrics) { + return buildStrictReplicaGroupArSelector(hybridSelector, brokerMetrics, + new PinotConfiguration(Map.of())); + } + + private StrictReplicaGroupInstanceSelector buildStrictReplicaGroupArSelector(HybridSelector hybridSelector, + BrokerMetrics brokerMetrics, PinotConfiguration brokerConfig) { + RoutingConfig routingConfig = new RoutingConfig(null, null, STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false); + TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName("testUpsertTable") + .setRoutingConfig(routingConfig).build(); + // Ideal state: mirrors the topology above + IdealState idealState = createIdealState(Map.of( + STRICT_SEGMENT0, List.of( + Pair.of(STRICT_RG0_SERVER_A, ONLINE), + Pair.of(STRICT_RG1_SERVER_C, ONLINE)), + STRICT_SEGMENT1, List.of( + Pair.of(STRICT_RG0_SERVER_A, ONLINE), + Pair.of(STRICT_RG1_SERVER_C, ONLINE)), + STRICT_SEGMENT2, List.of( + Pair.of(STRICT_RG0_SERVER_B, ONLINE), + Pair.of(STRICT_RG1_SERVER_D, ONLINE)))); + ExternalView externalView = createExternalView(Map.of( + STRICT_SEGMENT0, List.of( + Pair.of(STRICT_RG0_SERVER_A, ONLINE), + Pair.of(STRICT_RG1_SERVER_C, ONLINE)), + STRICT_SEGMENT1, List.of( + Pair.of(STRICT_RG0_SERVER_A, ONLINE), + Pair.of(STRICT_RG1_SERVER_C, ONLINE)), + STRICT_SEGMENT2, List.of( + Pair.of(STRICT_RG0_SERVER_B, ONLINE), + Pair.of(STRICT_RG1_SERVER_D, ONLINE)))); + // All servers use the fallback pool to prove strict routing keys off idealStateReplicaId rather than pool. + ServerInstance serverA = mock(ServerInstance.class); + when(serverA.getPool()).thenReturn(FALLBACK_POOL_ID); + ServerInstance serverB = mock(ServerInstance.class); + when(serverB.getPool()).thenReturn(FALLBACK_POOL_ID); + ServerInstance serverC = mock(ServerInstance.class); + when(serverC.getPool()).thenReturn(FALLBACK_POOL_ID); + ServerInstance serverD = mock(ServerInstance.class); + when(serverD.getPool()).thenReturn(FALLBACK_POOL_ID); + Map serverMap = Map.of( + STRICT_RG0_SERVER_A, serverA, STRICT_RG0_SERVER_B, serverB, + STRICT_RG1_SERVER_C, serverC, STRICT_RG1_SERVER_D, serverD); + return (StrictReplicaGroupInstanceSelector) InstanceSelectorFactory.getInstanceSelector(tableConfig, + mock(ZkHelixPropertyStore.class), brokerMetrics, hybridSelector, + brokerConfig, + Set.of(STRICT_RG0_SERVER_A, STRICT_RG0_SERVER_B, + STRICT_RG1_SERVER_C, STRICT_RG1_SERVER_D), + serverMap, idealState, externalView, new HashSet<>(STRICT_SEGMENTS)); + } + + @Test + public void testStrictReplicaGroupAdaptivePicksBestReplicaGroupByWorstCaseServer() { + // Replica group 0 servers: server_a (rank 2), server_b (rank 3) → worst = 3 + // Replica group 1 servers: server_c (rank 0), server_d (rank 1) → worst = 1 + // Replica group 1 is better (lower worst-case rank), so all segments should route there + HybridSelector hybridSelector = mock(HybridSelector.class); + BrokerMetrics brokerMetrics = mock(BrokerMetrics.class); + StrictReplicaGroupInstanceSelector instanceSelector = buildStrictReplicaGroupArSelector(hybridSelector, + brokerMetrics); + + when(hybridSelector.fetchServerRankingsWithScores(any())).thenReturn(Arrays.asList( + new ImmutablePair<>(STRICT_RG1_SERVER_C, 1.0), // rank 0 (best) + new ImmutablePair<>(STRICT_RG1_SERVER_D, 2.0), // rank 1 + new ImmutablePair<>(STRICT_RG0_SERVER_A, 3.0), // rank 2 + new ImmutablePair<>(STRICT_RG0_SERVER_B, 4.0) // rank 3 (worst) + )); + + InstanceSelector.InstanceMapping result = + instanceSelector.select(STRICT_SEGMENTS, 0, buildStrictReplicaGroupSegmentStates(), null); + + // All segments routed to replica group 1 + assertEquals(result.segmentToInstanceMap(), Map.of( + STRICT_SEGMENT0, STRICT_RG1_SERVER_C, + STRICT_SEGMENT1, STRICT_RG1_SERVER_C, + STRICT_SEGMENT2, STRICT_RG1_SERVER_D)); + // Metrics are still reported per actual pool, even though routing is chosen by replica group. + verify(brokerMetrics).addMeteredValue(eq(BrokerMeter.POOL_SEG_QUERIES), eq(3L), + eq(BrokerMetrics.getTagForPreferredPool(null)), eq(String.valueOf(FALLBACK_POOL_ID))); + } + + @Test + public void testStrictReplicaGroupAdaptivePicksGroup0WhenItHasBetterWorstCase() { + // Replica group 0 servers: server_a (rank 0), server_b (rank 1) → worst = 1 + // Replica group 1 servers: server_c (rank 2), server_d (rank 3) → worst = 3 + // Replica group 0 is better + HybridSelector hybridSelector = mock(HybridSelector.class); + StrictReplicaGroupInstanceSelector instanceSelector = buildStrictReplicaGroupArSelector(hybridSelector); + + when(hybridSelector.fetchServerRankingsWithScores(any())).thenReturn(Arrays.asList( + new ImmutablePair<>(STRICT_RG0_SERVER_A, 1.0), + new ImmutablePair<>(STRICT_RG0_SERVER_B, 2.0), + new ImmutablePair<>(STRICT_RG1_SERVER_C, 3.0), + new ImmutablePair<>(STRICT_RG1_SERVER_D, 4.0) + )); + + InstanceSelector.InstanceMapping result = + instanceSelector.select(STRICT_SEGMENTS, 0, buildStrictReplicaGroupSegmentStates(), null); + + // All segments routed to replica group 0 + assertEquals(result.segmentToInstanceMap(), Map.of( + STRICT_SEGMENT0, STRICT_RG0_SERVER_A, + STRICT_SEGMENT1, STRICT_RG0_SERVER_A, + STRICT_SEGMENT2, STRICT_RG0_SERVER_B)); + } + + @Test + public void testStrictReplicaGroupAdaptiveEmptyRankingsFallBackToRoundRobin() { + // When adaptive ranking returns no servers, fall back to round-robin by replica-group index. + HybridSelector hybridSelector = mock(HybridSelector.class); + StrictReplicaGroupInstanceSelector instanceSelector = buildStrictReplicaGroupArSelector(hybridSelector); + + when(hybridSelector.fetchServerRankingsWithScores(any())).thenReturn(List.of()); + + // requestId=0 → picks group at index 0 (replica group 0) + InstanceSelector.InstanceMapping result = + instanceSelector.select(STRICT_SEGMENTS, 0, buildStrictReplicaGroupSegmentStates(), null); + assertEquals(result.segmentToInstanceMap(), Map.of( + STRICT_SEGMENT0, STRICT_RG0_SERVER_A, + STRICT_SEGMENT1, STRICT_RG0_SERVER_A, + STRICT_SEGMENT2, STRICT_RG0_SERVER_B)); + + // requestId=1 → picks group at index 1 (replica group 1) + result = instanceSelector.select(STRICT_SEGMENTS, 1, buildStrictReplicaGroupSegmentStates(), null); + assertEquals(result.segmentToInstanceMap(), Map.of( + STRICT_SEGMENT0, STRICT_RG1_SERVER_C, + STRICT_SEGMENT1, STRICT_RG1_SERVER_C, + STRICT_SEGMENT2, STRICT_RG1_SERVER_D)); + } + + @Test + public void testStrictReplicaGroupAdaptiveConsistencyAllSegmentsSameReplicaGroup() { + // When adaptive scores favor replica group 1, all segments must route to that same replica group. + HybridSelector hybridSelector = mock(HybridSelector.class); + StrictReplicaGroupInstanceSelector instanceSelector = buildStrictReplicaGroupArSelector(hybridSelector); + + when(hybridSelector.fetchServerRankingsWithScores(any())).thenReturn(Arrays.asList( + new ImmutablePair<>(STRICT_RG1_SERVER_C, 1.0), + new ImmutablePair<>(STRICT_RG1_SERVER_D, 2.0), + new ImmutablePair<>(STRICT_RG0_SERVER_A, 3.0), + new ImmutablePair<>(STRICT_RG0_SERVER_B, 4.0) + )); + + InstanceSelector.InstanceMapping result = + instanceSelector.select(STRICT_SEGMENTS, 42, buildStrictReplicaGroupSegmentStates(), null); + + assertEquals(result.segmentToInstanceMap(), Map.of( + STRICT_SEGMENT0, STRICT_RG1_SERVER_C, + STRICT_SEGMENT1, STRICT_RG1_SERVER_C, + STRICT_SEGMENT2, STRICT_RG1_SERVER_D)); + } + + @Test + public void testStrictReplicaGroupAdaptivePrefersFullCoverageGroup() { + // Topology: replica group 0 has all 3 segments, replica group 1 only has segment0 and segment1. + // Even though replica group 1 has better adaptive ranks, replica group 0 should be chosen (full coverage). + // Replica group 0: server_a (seg0, seg1), server_b (seg2) — full coverage + // Replica group 1: server_c (seg0, seg1) — missing seg2, partial coverage + // Pools intentionally do not match replica groups: server_a and server_c share pool 0, while server_b is pool 1. + String serverC = "server_c"; + HybridSelector hybridSelector = mock(HybridSelector.class); + + RoutingConfig routingConfig = new RoutingConfig(null, null, STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false); + TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName("testUpsertTable") + .setRoutingConfig(routingConfig).build(); + // Replica group 1 only has seg0 and seg1 (no seg2) + IdealState idealState = createIdealState(Map.of( + STRICT_SEGMENT0, List.of(Pair.of(STRICT_RG0_SERVER_A, ONLINE), Pair.of(serverC, ONLINE)), + STRICT_SEGMENT1, List.of(Pair.of(STRICT_RG0_SERVER_A, ONLINE), Pair.of(serverC, ONLINE)), + STRICT_SEGMENT2, List.of(Pair.of(STRICT_RG0_SERVER_B, ONLINE)))); + ExternalView externalView = createExternalView(Map.of( + STRICT_SEGMENT0, List.of(Pair.of(STRICT_RG0_SERVER_A, ONLINE), Pair.of(serverC, ONLINE)), + STRICT_SEGMENT1, List.of(Pair.of(STRICT_RG0_SERVER_A, ONLINE), Pair.of(serverC, ONLINE)), + STRICT_SEGMENT2, List.of(Pair.of(STRICT_RG0_SERVER_B, ONLINE)))); + ServerInstance serverA = mock(ServerInstance.class); + when(serverA.getPool()).thenReturn(0); + ServerInstance serverB = mock(ServerInstance.class); + when(serverB.getPool()).thenReturn(1); + ServerInstance serverCInstance = mock(ServerInstance.class); + when(serverCInstance.getPool()).thenReturn(0); + Map serverMap = Map.of( + STRICT_RG0_SERVER_A, serverA, STRICT_RG0_SERVER_B, serverB, serverC, serverCInstance); + StrictReplicaGroupInstanceSelector instanceSelector = + (StrictReplicaGroupInstanceSelector) InstanceSelectorFactory.getInstanceSelector(tableConfig, + mock(ZkHelixPropertyStore.class), mock(BrokerMetrics.class), hybridSelector, + new PinotConfiguration(Map.of()), + Set.of(STRICT_RG0_SERVER_A, STRICT_RG0_SERVER_B, serverC), + serverMap, idealState, externalView, new HashSet<>(STRICT_SEGMENTS)); + + // Replica group 1 (server_c) has rank 0 — best rank, but only covers 2 of 3 segments + // Replica group 0 (server_a rank 1, server_b rank 2) — worse ranks, but full coverage + when(hybridSelector.fetchServerRankingsWithScores(any())).thenReturn(Arrays.asList( + new ImmutablePair<>(serverC, 1.0), // rank 0 (best) + new ImmutablePair<>(STRICT_RG0_SERVER_A, 3.0), // rank 1 + new ImmutablePair<>(STRICT_RG0_SERVER_B, 4.0) // rank 2 + )); + + // Build segment states where replica group 1 is missing seg2 + Map> candidatesMap = new HashMap<>(); + candidatesMap.put(STRICT_SEGMENT0, Arrays.asList( + new SegmentInstanceCandidate(STRICT_RG0_SERVER_A, true, 0, 0), + new SegmentInstanceCandidate(serverC, true, 0, 1))); + candidatesMap.put(STRICT_SEGMENT1, Arrays.asList( + new SegmentInstanceCandidate(STRICT_RG0_SERVER_A, true, 0, 0), + new SegmentInstanceCandidate(serverC, true, 0, 1))); + candidatesMap.put(STRICT_SEGMENT2, List.of( + new SegmentInstanceCandidate(STRICT_RG0_SERVER_B, true, 1, 0))); + SegmentStates segmentStates = new SegmentStates(candidatesMap, new HashSet<>(STRICT_SEGMENTS), null); + + InstanceSelector.InstanceMapping result = + instanceSelector.select(STRICT_SEGMENTS, 0, segmentStates, null); + + // Replica group 0 should be chosen (full coverage) even though replica group 1 has better ranks + assertEquals(result.segmentToInstanceMap(), Map.of( + STRICT_SEGMENT0, STRICT_RG0_SERVER_A, + STRICT_SEGMENT1, STRICT_RG0_SERVER_A, + STRICT_SEGMENT2, STRICT_RG0_SERVER_B)); + } + + @Test + public void testStrictReplicaGroupAdaptiveNewSegmentNotFalselyUnavailable() { + // Topology: server_a (replica group 0) hosts seg0 only. seg1 has no online instance (simulating a + // new/unassigned segment with null candidates in segmentStates). Replica group 0 is chosen (only group). + // seg1 should NOT appear in unavailable — it is a new segment, not a dropped one. + String serverA = "new_server_a"; + String seg0 = "new_seg0"; + String seg1 = "new_seg1"; + List segments = Arrays.asList(seg0, seg1); + HybridSelector hybridSelector = mock(HybridSelector.class); + + RoutingConfig routingConfig = new RoutingConfig(null, null, STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false); + TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName("testNewSegTable") + .setRoutingConfig(routingConfig).build(); + // seg1 intentionally absent from idealState/externalView → null candidates (new segment) + IdealState idealState = createIdealState(Map.of( + seg0, List.of(Pair.of(serverA, ONLINE)))); + ExternalView externalView = createExternalView(Map.of( + seg0, List.of(Pair.of(serverA, ONLINE)))); + ServerInstance serverAInstance = mock(ServerInstance.class); + when(serverAInstance.getPool()).thenReturn(FALLBACK_POOL_ID); + StrictReplicaGroupInstanceSelector instanceSelector = + (StrictReplicaGroupInstanceSelector) InstanceSelectorFactory.getInstanceSelector(tableConfig, + mock(ZkHelixPropertyStore.class), mock(BrokerMetrics.class), hybridSelector, + new PinotConfiguration(Map.of()), + Set.of(serverA), + Map.of(serverA, serverAInstance), + idealState, externalView, new HashSet<>(List.of(seg0))); + + when(hybridSelector.fetchServerRankingsWithScores(any())).thenReturn(Arrays.asList( + new ImmutablePair<>(serverA, 1.0))); + + BrokerRequest brokerRequest = mock(BrokerRequest.class); + PinotQuery pinotQuery = mock(PinotQuery.class); + when(brokerRequest.getPinotQuery()).thenReturn(pinotQuery); + when(pinotQuery.getQueryOptions()).thenReturn(null); + + InstanceSelector.SelectionResult result = instanceSelector.select(brokerRequest, segments, 0); + assertEquals(result.getSegmentToInstanceMap(), Map.of(seg0, serverA)); + // seg1 has null candidates (not yet in the selector's state) → must NOT appear as unavailable + assertTrue(result.getUnavailableSegments().isEmpty()); + } + + @Test + public void testStrictReplicaGroupAdaptiveDroppedSegmentReportedAsUnavailable() { + // Topology: no replica group (idealStateReplicaId) has full coverage. + // ReplicaId 0 (server_a): covers seg0, seg1 (missing seg2) + // ReplicaId 1 (server_c, server_d): covers seg1, seg2 (missing seg0) + // ReplicaId 1 has better rank -> chosen. seg0 has no replicaId-1 candidate -> reported unavailable. + HybridSelector hybridSelector = mock(HybridSelector.class); + StrictReplicaGroupInstanceSelector instanceSelector = buildStrictReplicaGroupArSelector(hybridSelector); + + when(hybridSelector.fetchServerRankingsWithScores(any())).thenReturn(Arrays.asList( + new ImmutablePair<>(STRICT_RG1_SERVER_C, 1.0), + new ImmutablePair<>(STRICT_RG0_SERVER_A, 3.0) + )); + + // Craft segment states with partial coverage per replica group: + // seg0: only in replicaId 0 (server_a) + // seg1: in both replicaId 0 (server_a) and replicaId 1 (server_c) + // seg2: only in replicaId 1 (server_d) + Map> candidatesMap = new HashMap<>(); + candidatesMap.put(STRICT_SEGMENT0, List.of( + new SegmentInstanceCandidate(STRICT_RG0_SERVER_A, true, FALLBACK_POOL_ID, 0))); + candidatesMap.put(STRICT_SEGMENT1, Arrays.asList( + new SegmentInstanceCandidate(STRICT_RG0_SERVER_A, true, FALLBACK_POOL_ID, 0), + new SegmentInstanceCandidate(STRICT_RG1_SERVER_C, true, FALLBACK_POOL_ID, 1))); + candidatesMap.put(STRICT_SEGMENT2, List.of( + new SegmentInstanceCandidate(STRICT_RG1_SERVER_D, true, FALLBACK_POOL_ID, 1))); + SegmentStates segmentStates = new SegmentStates(candidatesMap, new HashSet<>(STRICT_SEGMENTS), null); + + InstanceSelector.InstanceMapping result = + instanceSelector.select(STRICT_SEGMENTS, 0, segmentStates, null); + + // seg1 and seg2 routed to replicaId 1 + assertEquals(result.segmentToInstanceMap(), Map.of( + STRICT_SEGMENT1, STRICT_RG1_SERVER_C, + STRICT_SEGMENT2, STRICT_RG1_SERVER_D)); + // seg0 has candidates (replicaId 0 only) but no match in chosen replicaId 1 -> reported unavailable + assertEquals(result.unavailableSegments(), List.of(STRICT_SEGMENT0)); + } + + @Test + public void testStrictReplicaGroupAdaptiveOnlyQueryRelevantServersScored() { + // Replica group 1 has an extra server (server_e) that only hosts seg_extra, which is not in the query. + // server_e should never be scored for a query over STRICT_SEGMENTS. + // Replica group 0: server_a (rank 2), server_b (rank 3) → worst = 3 + // Replica group 1: server_c (rank 0), server_d (rank 1) → worst = 1 + // Replica group 1 should win because only query-relevant servers matter. + String rg1ServerE = "server_e"; + String segExtra = "seg_extra"; + HybridSelector hybridSelector = mock(HybridSelector.class); + + // Build selector with server_e in replica group 1, hosting seg_extra. + RoutingConfig routingConfig = new RoutingConfig(null, null, STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false); + TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName("testUpsertTable") + .setRoutingConfig(routingConfig).build(); + // server_e's pool intentionally differs from the queried servers to show pool is irrelevant. + // segExtra places server_d at position 0 and server_e at position 1 (idealStateReplicaId=1). + IdealState idealState = createIdealState(Map.of( + STRICT_SEGMENT0, List.of(Pair.of(STRICT_RG0_SERVER_A, ONLINE), Pair.of(STRICT_RG1_SERVER_C, ONLINE)), + STRICT_SEGMENT1, List.of(Pair.of(STRICT_RG0_SERVER_A, ONLINE), Pair.of(STRICT_RG1_SERVER_C, ONLINE)), + STRICT_SEGMENT2, List.of(Pair.of(STRICT_RG0_SERVER_B, ONLINE), Pair.of(STRICT_RG1_SERVER_D, ONLINE)), + segExtra, List.of(Pair.of(STRICT_RG1_SERVER_D, ONLINE), Pair.of(rg1ServerE, ONLINE)))); + ExternalView externalView = createExternalView(Map.of( + STRICT_SEGMENT0, List.of(Pair.of(STRICT_RG0_SERVER_A, ONLINE), Pair.of(STRICT_RG1_SERVER_C, ONLINE)), + STRICT_SEGMENT1, List.of(Pair.of(STRICT_RG0_SERVER_A, ONLINE), Pair.of(STRICT_RG1_SERVER_C, ONLINE)), + STRICT_SEGMENT2, List.of(Pair.of(STRICT_RG0_SERVER_B, ONLINE), Pair.of(STRICT_RG1_SERVER_D, ONLINE)), + segExtra, List.of(Pair.of(STRICT_RG1_SERVER_D, ONLINE), Pair.of(rg1ServerE, ONLINE)))); + ServerInstance serverA = mock(ServerInstance.class); + when(serverA.getPool()).thenReturn(FALLBACK_POOL_ID); + ServerInstance serverB = mock(ServerInstance.class); + when(serverB.getPool()).thenReturn(FALLBACK_POOL_ID); + ServerInstance serverC = mock(ServerInstance.class); + when(serverC.getPool()).thenReturn(FALLBACK_POOL_ID); + ServerInstance serverD = mock(ServerInstance.class); + when(serverD.getPool()).thenReturn(FALLBACK_POOL_ID); + ServerInstance serverEInstance = mock(ServerInstance.class); + when(serverEInstance.getPool()).thenReturn(99); + Map serverMap = Map.of( + STRICT_RG0_SERVER_A, serverA, STRICT_RG0_SERVER_B, serverB, + STRICT_RG1_SERVER_C, serverC, STRICT_RG1_SERVER_D, serverD, + rg1ServerE, serverEInstance); + Set allSegments = new HashSet<>(STRICT_SEGMENTS); + allSegments.add(segExtra); + StrictReplicaGroupInstanceSelector instanceSelector = + (StrictReplicaGroupInstanceSelector) InstanceSelectorFactory.getInstanceSelector(tableConfig, + mock(ZkHelixPropertyStore.class), mock(BrokerMetrics.class), hybridSelector, + new PinotConfiguration(Map.of()), + Set.of(STRICT_RG0_SERVER_A, STRICT_RG0_SERVER_B, + STRICT_RG1_SERVER_C, STRICT_RG1_SERVER_D, rg1ServerE), + serverMap, idealState, externalView, allSegments); + + // server_e has a much worse mocked score but hosts no query segments, so it should never be ranked. + when(hybridSelector.fetchServerRankingsWithScores(any())).thenReturn(Arrays.asList( + new ImmutablePair<>(STRICT_RG1_SERVER_C, 1.0), // rank 0 + new ImmutablePair<>(STRICT_RG1_SERVER_D, 2.0), // rank 1 + new ImmutablePair<>(STRICT_RG0_SERVER_A, 3.0), // rank 2 + new ImmutablePair<>(STRICT_RG0_SERVER_B, 4.0), // rank 3 + new ImmutablePair<>(rg1ServerE, 99.0) // rank 4 (not query-relevant) + )); + + Map> candidatesMap = new HashMap<>(); + candidatesMap.put(STRICT_SEGMENT0, + buildStrictReplicaGroupCandidates(STRICT_RG0_SERVER_A, STRICT_RG1_SERVER_C)); + candidatesMap.put(STRICT_SEGMENT1, + buildStrictReplicaGroupCandidates(STRICT_RG0_SERVER_A, STRICT_RG1_SERVER_C)); + candidatesMap.put(STRICT_SEGMENT2, + buildStrictReplicaGroupCandidates(STRICT_RG0_SERVER_B, STRICT_RG1_SERVER_D)); + candidatesMap.put(segExtra, Arrays.asList( + new SegmentInstanceCandidate(STRICT_RG1_SERVER_D, true, FALLBACK_POOL_ID, 0), + new SegmentInstanceCandidate(rg1ServerE, true, 99, 1))); + SegmentStates segmentStates = new SegmentStates(candidatesMap, allSegments, null); + + InstanceSelector.InstanceMapping result = + instanceSelector.select(STRICT_SEGMENTS, 0, segmentStates, null); + + // Replica group 1 wins (worst rank among query-relevant servers = 1) over replica group 0 (worst = 3) + // server_e is not considered because it hosts no query segments + assertEquals(result.segmentToInstanceMap(), Map.of( + STRICT_SEGMENT0, STRICT_RG1_SERVER_C, + STRICT_SEGMENT1, STRICT_RG1_SERVER_C, + STRICT_SEGMENT2, STRICT_RG1_SERVER_D)); + + ArgumentCaptor> rankedCandidatesCaptor = ArgumentCaptor.forClass(List.class); + verify(hybridSelector).fetchServerRankingsWithScores(rankedCandidatesCaptor.capture()); + assertEquals(new HashSet<>(rankedCandidatesCaptor.getValue()), + Set.of(STRICT_RG0_SERVER_A, STRICT_RG0_SERVER_B, + STRICT_RG1_SERVER_C, STRICT_RG1_SERVER_D)); + assertEquals(rankedCandidatesCaptor.getValue().size(), 4); + } +} diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java index ee25435988a6..3e7c547d9f50 100644 --- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java +++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java @@ -1271,6 +1271,13 @@ public enum Type { public static final String CONFIG_OF_STATS_METRIC_EXPORT_INTERVAL_MS = CONFIG_PREFIX + ".stats.metric.export.interval.ms"; public static final long DEFAULT_STATS_METRIC_EXPORT_INTERVAL_MS = 10 * 1000; + + // Controls whether pool-level adaptive routing is enabled for StrictReplicaGroupInstanceSelector. + // When false, StrictReplicaGroupInstanceSelector falls back to round-robin even if adaptive server + // selection is configured. + public static final String CONFIG_OF_ENABLE_STRICT_REPLICA_GROUP = + CONFIG_PREFIX + ".enable.strict.replica.group"; + public static final boolean DEFAULT_ENABLE_STRICT_REPLICA_GROUP = true; } public static class Grpc { diff --git a/pom.xml b/pom.xml index d28b9fd02b6c..9fead38d5c89 100644 --- a/pom.xml +++ b/pom.xml @@ -2071,7 +2071,9 @@ ,\# - + + cleanthat-javaparser-unnecessaryimport +