This is an automated email from the ASF dual-hosted git repository.
gortiz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new ba49142265f Share instance selector candidate lists between segments
with equal assignments (#19785)
ba49142265f is described below
commit ba49142265f93e4eef17c42c6dc262ab9c9e9f5b
Author: Gonzalo Ortiz Jaureguizar <[email protected]>
AuthorDate: Thu Oct 15 10:20:09 2026 +0200
Share instance selector candidate lists between segments with equal
assignments (#19785)
Part of #19782.
---
.../instanceselector/BaseInstanceSelector.java | 248 ++++++++++++++++-----
.../ReplicaGroupInstanceSelector.java | 65 +++---
.../instanceselector/InstanceSelectorTest.java | 189 ++++++++++++++++
3 files changed, 420 insertions(+), 82 deletions(-)
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 6c8d96cbf6d..263dc258457 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
@@ -25,10 +25,10 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
-import java.util.TreeSet;
import javax.annotation.Nullable;
import org.apache.helix.AccessOption;
import org.apache.helix.model.ExternalView;
@@ -50,6 +50,8 @@ import
org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.SegmentStateM
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.slf4j.Marker;
+import org.slf4j.MarkerFactory;
import static
org.apache.pinot.spi.utils.CommonConstants.Broker.FALLBACK_POOL_ID;
@@ -90,8 +92,18 @@ import static
org.apache.pinot.spi.utils.CommonConstants.Broker.FALLBACK_POOL_ID
/// which one owns the table's gauges.
public abstract class BaseInstanceSelector implements InstanceSelector {
private static final Logger LOGGER =
LoggerFactory.getLogger(BaseInstanceSelector.class);
+ /// Tags the INFO logs that print the whole new-segment map on every rebuild
("Got N new segments: ..."). To drop
+ /// them without code changes, add to the log4j2 config:
+ /// `<MarkerFilter marker="ROUTING_NEW_SEGMENTS" onMatch="DENY"
onMismatch="NEUTRAL"/>`
+ private static final Marker NEW_SEGMENTS_MARKER =
MarkerFactory.getMarker("ROUTING_NEW_SEGMENTS");
// To prevent int overflow, reset the request id once it reaches this value
protected static final long MAX_REQUEST_ID = 1_000_000_000;
+ /// Maximum number of distinct (ideal state, external view) instance state
map pairs whose candidate lists one
+ /// rebuild shares between segments. Past this bound, the rebuild stops
caching new pairs: segments with an already
+ /// cached pair keep sharing its lists, and segments with any other pair get
their own lists, as if there was no
+ /// sharing. In a realtime table, all the segments of a stream partition
usually have the same pair, so the number of
+ /// distinct pairs is close to the number of partitions, far below this
bound.
+ static final int MAX_SHARED_CANDIDATE_LISTS = 4096;
protected TableConfig _tableConfig;
protected String _tableNameWithType;
@@ -126,6 +138,9 @@ public abstract class BaseInstanceSelector implements
InstanceSelector {
// computed alongside.
protected volatile TableReplicaHealth _replicaHealth;
protected Map<String, ServerInstance> _enabledServerStore;
+ // Only read by the rebuild thread. Not final so that tests can disable or
bound the sharing.
+ @VisibleForTesting
+ int _maxSharedCandidateLists = MAX_SHARED_CANDIDATE_LISTS;
@Override
public void init(TableConfig tableConfig, ZkHelixPropertyStore<ZNRecord>
propertyStore,
@@ -235,7 +250,8 @@ public abstract class BaseInstanceSelector implements
InstanceSelector {
newSegmentCreationTimeMap.put(segmentZKMetadata.getSegmentName(),
creationTimeMs);
}
}
- LOGGER.info("Got {} new segments: {} for table: {} by reading ZK metadata,
current time: {}",
+ LOGGER.info(NEW_SEGMENTS_MARKER,
+ "Got {} new segments: {} for table: {} by reading ZK metadata, current
time: {}",
newSegmentCreationTimeMap.size(), newSegmentCreationTimeMap,
_tableNameWithType, currentTimeMs);
return newSegmentCreationTimeMap;
}
@@ -265,9 +281,9 @@ public abstract class BaseInstanceSelector implements
InstanceSelector {
}
/// Returns the online instances for routing purpose.
- static TreeSet<String> getOnlineInstances(Map<String, String>
idealStateInstanceStateMap,
+ static Set<String> getOnlineInstances(Map<String, String>
idealStateInstanceStateMap,
Map<String, String> externalViewInstanceStateMap) {
- TreeSet<String> onlineInstances = new TreeSet<>();
+ Set<String> onlineInstances = new HashSet<>();
// Only track ONLINE/CONSUMING instances within the ideal state
for (Map.Entry<String, String> entry :
idealStateInstanceStateMap.entrySet()) {
String instance = entry.getKey();
@@ -279,13 +295,58 @@ public abstract class BaseInstanceSelector implements
InstanceSelector {
return onlineInstances;
}
- /// Converts the given map into a sorted map if needed.
- static SortedMap<String, String> convertToSortedMap(Map<String, String> map)
{
+ /// Returns a map that iterates the entries of the given map in key order:
the map itself when it is a
+ /// [SortedMap] or when its iteration order is already sorted, otherwise a
sorted copy.
+ ///
+ /// The instance state maps read from ZooKeeper are insertion-ordered, and
are usually written in key order, so
+ /// the linear check avoids the copy in the common case.
+ static Map<String, String> sortByKeyIfNeeded(Map<String, String> map) {
if (map instanceof SortedMap) {
- return (SortedMap<String, String>) map;
- } else {
- return new TreeMap<>(map);
+ return map;
}
+ String previousKey = null;
+ for (String key : map.keySet()) {
+ if (previousKey != null && previousKey.compareTo(key) > 0) {
+ return new TreeMap<>(map);
+ }
+ previousKey = key;
+ }
+ return map;
+ }
+
+ /// Returns the candidates of an old segment: the instances that are
ONLINE/CONSUMING in both the ideal state and
+ /// the external view, in instance order.
+ List<SegmentInstanceCandidate> computeOldSegmentCandidates(Map<String,
String> idealStateInstanceStateMap,
+ Map<String, String> externalViewInstanceStateMap) {
+ List<SegmentInstanceCandidate> candidates = new
ArrayList<>(idealStateInstanceStateMap.size());
+ int idealStateReplicaId = 0;
+ for (Map.Entry<String, String> entry :
sortByKeyIfNeeded(idealStateInstanceStateMap).entrySet()) {
+ String instance = entry.getKey();
+ // NOTE: DO NOT check if EV matches IS because it is a valid state when
EV is CONSUMING while IS is ONLINE
+ if (isOnlineForRouting(entry.getValue()) &&
isOnlineForRouting(externalViewInstanceStateMap.get(instance))) {
+ candidates.add(new SegmentInstanceCandidate(instance, true,
getPool(instance), idealStateReplicaId));
+ }
+ idealStateReplicaId++;
+ }
+ return candidates;
+ }
+
+ /// Returns the candidates of a new segment: the instances that are
ONLINE/CONSUMING in the ideal state, in instance
+ /// order. A candidate is online when the instance is ONLINE/CONSUMING in
the external view.
+ List<SegmentInstanceCandidate> computeNewSegmentCandidates(Map<String,
String> idealStateInstanceStateMap,
+ @Nullable Map<String, String> externalViewInstanceStateMap) {
+ List<SegmentInstanceCandidate> candidates = new
ArrayList<>(idealStateInstanceStateMap.size());
+ int idealStateReplicaId = 0;
+ for (Map.Entry<String, String> entry :
sortByKeyIfNeeded(idealStateInstanceStateMap).entrySet()) {
+ if (isOnlineForRouting(entry.getValue())) {
+ String instance = entry.getKey();
+ boolean online =
+ externalViewInstanceStateMap != null &&
isOnlineForRouting(externalViewInstanceStateMap.get(instance));
+ candidates.add(new SegmentInstanceCandidate(instance, online,
getPool(instance), idealStateReplicaId));
+ }
+ idealStateReplicaId++;
+ }
+ return candidates;
}
/// Updates the segment maps based on the given ideal state, external view,
online segments (segments with
@@ -301,60 +362,31 @@ public abstract class BaseInstanceSelector implements
InstanceSelector {
Map<String, Map<String, String>> idealStateAssignment =
idealState.getRecord().getMapFields();
Map<String, Map<String, String>> externalViewAssignment =
externalView.getRecord().getMapFields();
+ InstanceStateMapsCache instanceStateMapsCache = new
InstanceStateMapsCache(_maxSharedCandidateLists);
int numSinglePoolSegments = 0;
Set<Integer> pools = new HashSet<>();
for (String segment : onlineSegments) {
Map<String, String> idealStateInstanceStateMap =
idealStateAssignment.get(segment);
- // TODO: Verify whether sorting is actually needed
- Map<String, String> sortedIdealStateMap =
convertToSortedMap(idealStateInstanceStateMap);
Long newSegmentCreationTimeMs = newSegmentCreationTimeMap.get(segment);
Map<String, String> externalViewInstanceStateMap =
externalViewAssignment.get(segment);
- if (externalViewInstanceStateMap == null) {
- if (newSegmentCreationTimeMs != null) {
- // New segment
- List<SegmentInstanceCandidate> candidates = new
ArrayList<>(idealStateInstanceStateMap.size());
- int idealStateReplicaId = 0;
- for (Map.Entry<String, String> entry :
sortedIdealStateMap.entrySet()) {
- if (isOnlineForRouting(entry.getValue())) {
- String instance = entry.getKey();
- candidates.add(new SegmentInstanceCandidate(instance, false,
getPool(instance), idealStateReplicaId));
- }
- idealStateReplicaId++;
- }
- _newSegmentStateMap.put(segment, new
NewSegmentState(newSegmentCreationTimeMs, candidates));
- } else {
- // Old segment
- putOldSegment(segment, List.of(), idealStateInstanceStateMap);
- }
+ if (newSegmentCreationTimeMs != null) {
+ // New segment
+ _newSegmentStateMap.put(segment, new
NewSegmentState(newSegmentCreationTimeMs,
+ computeNewSegmentCandidates(idealStateInstanceStateMap,
externalViewInstanceStateMap)));
+ } else if (externalViewInstanceStateMap == null) {
+ // Old segment
+ putOldSegment(segment, List.of(), idealStateInstanceStateMap);
} else {
- TreeSet<String> onlineInstances =
getOnlineInstances(idealStateInstanceStateMap, externalViewInstanceStateMap);
- if (newSegmentCreationTimeMs != null) {
- // New segment
- List<SegmentInstanceCandidate> candidates = new
ArrayList<>(idealStateInstanceStateMap.size());
- int idealStateReplicaId = 0;
- for (Map.Entry<String, String> entry :
sortedIdealStateMap.entrySet()) {
- if (isOnlineForRouting(entry.getValue())) {
- String instance = entry.getKey();
- candidates.add(
- new SegmentInstanceCandidate(instance,
onlineInstances.contains(instance), getPool(instance),
- idealStateReplicaId));
- }
- idealStateReplicaId++;
- }
- _newSegmentStateMap.put(segment, new
NewSegmentState(newSegmentCreationTimeMs, candidates));
- } else {
- // Old segment
- List<SegmentInstanceCandidate> candidates = new
ArrayList<>(onlineInstances.size());
- int idealStateReplicaId = 0;
- for (String instance : sortedIdealStateMap.keySet()) {
- if (onlineInstances.contains(instance)) {
- candidates.add(new SegmentInstanceCandidate(instance, true,
getPool(instance), idealStateReplicaId));
- }
- idealStateReplicaId++;
- }
- putOldSegment(segment, candidates, idealStateInstanceStateMap);
+ // Old segment. Segments with the same instance state maps share the
candidate list.
+ InstanceStateMaps instanceStateMaps =
+ instanceStateMapsCache.canonicalize(idealStateInstanceStateMap,
externalViewInstanceStateMap);
+ List<SegmentInstanceCandidate> candidates =
instanceStateMaps._oldSegmentCandidates;
+ if (candidates == null) {
+ candidates = computeOldSegmentCandidates(idealStateInstanceStateMap,
externalViewInstanceStateMap);
+ instanceStateMaps._oldSegmentCandidates = candidates;
}
+ putOldSegment(segment, candidates, idealStateInstanceStateMap);
}
if (_emitSinglePoolSegmentsMetric) {
pools.clear();
@@ -443,10 +475,33 @@ public abstract class BaseInstanceSelector implements
InstanceSelector {
unavailableSegments.size());
}
+ /// Returns the enabled candidates, and adds their instances to the serving
instances.
+ ///
+ /// When every candidate is enabled, returns the given list itself rather
than a copy. That publishes the list to
+ /// the query path, which is safe because candidate lists are never modified
once built. It also keeps the lists
+ /// that [#updateSegmentMaps] shares between segments shared in the
published [SegmentStates].
private List<SegmentInstanceCandidate>
getEnabledCandidatesAndAddToServingInstances(
List<SegmentInstanceCandidate> candidates, Set<String> servingInstances)
{
- List<SegmentInstanceCandidate> enabledCandidates = new
ArrayList<>(candidates.size());
- for (SegmentInstanceCandidate candidate : candidates) {
+ int numCandidates = candidates.size();
+ int numLeadingEnabledCandidates = 0;
+ while (numLeadingEnabledCandidates < numCandidates) {
+ String instance =
candidates.get(numLeadingEnabledCandidates).getInstance();
+ if (!_enabledInstances.contains(instance)) {
+ break;
+ }
+ servingInstances.add(instance);
+ numLeadingEnabledCandidates++;
+ }
+ if (numLeadingEnabledCandidates == numCandidates) {
+ return candidates;
+ }
+ List<SegmentInstanceCandidate> enabledCandidates = new
ArrayList<>(numCandidates - 1);
+ for (int i = 0; i < numLeadingEnabledCandidates; i++) {
+ enabledCandidates.add(candidates.get(i));
+ }
+ // The candidate at numLeadingEnabledCandidates is disabled
+ for (int i = numLeadingEnabledCandidates + 1; i < numCandidates; i++) {
+ SegmentInstanceCandidate candidate = candidates.get(i);
String instance = candidate.getInstance();
if (_enabledInstances.contains(instance)) {
enabledCandidates.add(candidate);
@@ -505,7 +560,8 @@ public abstract class BaseInstanceSelector implements
InstanceSelector {
}
}
}
- LOGGER.info("Got {} new segments: {} for table: {} by processing existing
states, current time: {}",
+ LOGGER.info(NEW_SEGMENTS_MARKER,
+ "Got {} new segments: {} for table: {} by processing existing states,
current time: {}",
newSegmentCreationTimeMap.size(), newSegmentCreationTimeMap,
_tableNameWithType, currentTimeMs);
return newSegmentCreationTimeMap;
}
@@ -559,4 +615,82 @@ public abstract class BaseInstanceSelector implements
InstanceSelector {
/// servers and let servers decide how to handle them.
protected abstract InstanceMapping select(List<String> segments, int
requestId, SegmentStates segmentStates,
Map<String, String> queryOptions);
+
+ /// The ideal state and external view instance state maps of a segment,
compared by content. Within one rebuild,
+ /// segments with equal maps get the same routing candidates, so the
canonical instance (see
+ /// [InstanceStateMapsCache]) caches what is derived from the maps, and the
segments share it.
+ ///
+ /// Not thread safe. Only used by the rebuild thread, for the duration of
one rebuild.
+ static final class InstanceStateMaps {
+ final Map<String, String> _idealStateInstanceStateMap;
+ @Nullable
+ final Map<String, String> _externalViewInstanceStateMap;
+ private final int _hashCode;
+ // Derived from the maps, computed on first use
+ @Nullable
+ Set<String> _onlineInstances;
+ @Nullable
+ List<SegmentInstanceCandidate> _oldSegmentCandidates;
+ // Used by the strict replica-group selector to collect the unavailable
instances once per instance state maps
+ boolean _unavailableInstancesCollected;
+
+ InstanceStateMaps(Map<String, String> idealStateInstanceStateMap,
+ @Nullable Map<String, String> externalViewInstanceStateMap) {
+ _idealStateInstanceStateMap = idealStateInstanceStateMap;
+ _externalViewInstanceStateMap = externalViewInstanceStateMap;
+ _hashCode = 31 * idealStateInstanceStateMap.hashCode() +
Objects.hashCode(externalViewInstanceStateMap);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof InstanceStateMaps)) {
+ return false;
+ }
+ InstanceStateMaps that = (InstanceStateMaps) o;
+ return _hashCode == that._hashCode &&
_idealStateInstanceStateMap.equals(that._idealStateInstanceStateMap)
+ && Objects.equals(_externalViewInstanceStateMap,
that._externalViewInstanceStateMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return _hashCode;
+ }
+ }
+
+ /// Returns one canonical [InstanceStateMaps] per distinct pair of instance
state maps, so that the segments of one
+ /// rebuild with equal maps share what is derived from them.
+ ///
+ /// The cache keeps at most `maxSize` pairs. Past that bound, it stops
caching new pairs: a pair already in the cache
+ /// still returns its canonical [InstanceStateMaps], while any other pair
returns a new [InstanceStateMaps] for each
+ /// call, which makes each segment compute its own state, as if there was no
sharing.
+ ///
+ /// Not thread safe. Only used by the rebuild thread, for the duration of
one rebuild.
+ static final class InstanceStateMapsCache {
+ private final int _maxSize;
+ private final Map<InstanceStateMaps, InstanceStateMaps> _cache = new
HashMap<>();
+
+ InstanceStateMapsCache(int maxSize) {
+ _maxSize = maxSize;
+ }
+
+ InstanceStateMaps canonicalize(Map<String, String>
idealStateInstanceStateMap,
+ @Nullable Map<String, String> externalViewInstanceStateMap) {
+ InstanceStateMaps instanceStateMaps =
+ new InstanceStateMaps(idealStateInstanceStateMap,
externalViewInstanceStateMap);
+ if (_maxSize <= 0) {
+ return instanceStateMaps;
+ }
+ InstanceStateMaps canonical = _cache.get(instanceStateMaps);
+ if (canonical != null) {
+ return canonical;
+ }
+ if (_cache.size() < _maxSize) {
+ _cache.put(instanceStateMaps, instanceStateMaps);
+ }
+ return instanceStateMaps;
+ }
+ }
}
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 038994d40cd..9f1be381206 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
@@ -208,33 +208,42 @@ public class ReplicaGroupInstanceSelector extends
BaseInstanceSelector {
Map<String, Map<String, String>> idealStateAssignment =
idealState.getRecord().getMapFields();
Map<String, Map<String, String>> externalViewAssignment =
externalView.getRecord().getMapFields();
- // Get the online instances for the segments
- Map<String, Set<String>> oldSegmentToOnlineInstancesMap =
+ // Get the online instances for the segments. Segments with the same
instance state maps share one
+ // InstanceStateMaps, which caches the online instances and the old
segment candidates derived from the maps.
+ InstanceStateMapsCache instanceStateMapsCache = new
InstanceStateMapsCache(_maxSharedCandidateLists);
+ Map<String, InstanceStateMaps> oldSegmentToInstanceStateMaps =
new HashMap<>(HashUtil.getHashMapCapacity(onlineSegments.size()));
- Map<String, Set<String>> newSegmentToOnlineInstancesMap = new
HashMap<>(newSegmentMapCapacity);
+ Map<String, InstanceStateMaps> newSegmentToInstanceStateMaps = new
HashMap<>(newSegmentMapCapacity);
for (String segment : onlineSegments) {
Map<String, String> idealStateInstanceStateMap =
idealStateAssignment.get(segment);
assert idealStateInstanceStateMap != null;
Map<String, String> externalViewInstanceStateMap =
externalViewAssignment.get(segment);
- Set<String> onlineInstances;
- if (externalViewInstanceStateMap == null) {
- onlineInstances = Set.of();
- } else {
- onlineInstances = getOnlineInstances(idealStateInstanceStateMap,
externalViewInstanceStateMap);
+ InstanceStateMaps instanceStateMaps =
+ instanceStateMapsCache.canonicalize(idealStateInstanceStateMap,
externalViewInstanceStateMap);
+ if (instanceStateMaps._onlineInstances == null) {
+ instanceStateMaps._onlineInstances = externalViewInstanceStateMap ==
null
+ ? Set.of()
+ : getOnlineInstances(idealStateInstanceStateMap,
externalViewInstanceStateMap);
}
if (newSegmentCreationTimeMap.containsKey(segment)) {
- newSegmentToOnlineInstancesMap.put(segment, onlineInstances);
+ newSegmentToInstanceStateMaps.put(segment, instanceStateMaps);
} else {
- oldSegmentToOnlineInstancesMap.put(segment, onlineInstances);
+ oldSegmentToInstanceStateMaps.put(segment, instanceStateMaps);
}
}
// Calculate the unavailable instances based on the old segments' online
instances for each combination of instances
// in the ideal state
Map<Set<String>, Set<String>> unavailableInstancesMap = new HashMap<>();
- for (Map.Entry<String, Set<String>> entry :
oldSegmentToOnlineInstancesMap.entrySet()) {
+ for (Map.Entry<String, InstanceStateMaps> entry :
oldSegmentToInstanceStateMaps.entrySet()) {
String segment = entry.getKey();
- Set<String> onlineInstances = entry.getValue();
+ InstanceStateMaps instanceStateMaps = entry.getValue();
+ if (instanceStateMaps._unavailableInstancesCollected) {
+ // Another segment with the same instance state maps already collected
the same unavailable instances
+ continue;
+ }
+ instanceStateMaps._unavailableInstancesCollected = true;
+ Set<String> onlineInstances = instanceStateMaps._onlineInstances;
Map<String, String> idealStateInstanceStateMap =
idealStateAssignment.get(segment);
Set<String> instancesInIdealState = idealStateInstanceStateMap.keySet();
Set<String> unavailableInstances =
@@ -252,30 +261,36 @@ public class ReplicaGroupInstanceSelector extends
BaseInstanceSelector {
}
// Iterate over the maps and exclude the unavailable instances
- for (Map.Entry<String, Set<String>> entry :
oldSegmentToOnlineInstancesMap.entrySet()) {
+ for (Map.Entry<String, InstanceStateMaps> entry :
oldSegmentToInstanceStateMaps.entrySet()) {
String segment = entry.getKey();
- // NOTE: onlineInstances is either a TreeSet or an EmptySet (sorted)
- Set<String> onlineInstances = entry.getValue();
+ InstanceStateMaps instanceStateMaps = entry.getValue();
Map<String, String> idealStateInstanceStateMap =
idealStateAssignment.get(segment);
- Set<String> unavailableInstances =
unavailableInstancesMap.get(idealStateInstanceStateMap.keySet());
- List<SegmentInstanceCandidate> candidates = new
ArrayList<>(onlineInstances.size());
- int idealStateReplicaId = 0;
- for (String instance :
convertToSortedMap(idealStateInstanceStateMap).keySet()) {
- if (onlineInstances.contains(instance) &&
!unavailableInstances.contains(instance)) {
- candidates.add(new SegmentInstanceCandidate(instance, true,
getPool(instance), idealStateReplicaId));
+ // The candidates only depend on the instance state maps: the
unavailable instances are a function of the
+ // instances in the ideal state, and they are final once computed above
+ List<SegmentInstanceCandidate> candidates =
instanceStateMaps._oldSegmentCandidates;
+ if (candidates == null) {
+ Set<String> onlineInstances = instanceStateMaps._onlineInstances;
+ Set<String> unavailableInstances =
unavailableInstancesMap.get(idealStateInstanceStateMap.keySet());
+ candidates = new ArrayList<>(onlineInstances.size());
+ int idealStateReplicaId = 0;
+ for (String instance :
sortByKeyIfNeeded(idealStateInstanceStateMap).keySet()) {
+ if (onlineInstances.contains(instance) &&
!unavailableInstances.contains(instance)) {
+ candidates.add(new SegmentInstanceCandidate(instance, true,
getPool(instance), idealStateReplicaId));
+ }
+ idealStateReplicaId++;
}
- idealStateReplicaId++;
+ instanceStateMaps._oldSegmentCandidates = candidates;
}
// Instances taken out of service for the whole replica group are
excluded above, so measuring against
// the ideal state count is what makes the replica health metrics
reflect a group-wide knockout.
putOldSegment(segment, candidates, idealStateInstanceStateMap);
}
- for (Map.Entry<String, Set<String>> entry :
newSegmentToOnlineInstancesMap.entrySet()) {
+ for (Map.Entry<String, InstanceStateMaps> entry :
newSegmentToInstanceStateMaps.entrySet()) {
String segment = entry.getKey();
- Set<String> onlineInstances = entry.getValue();
+ Set<String> onlineInstances = entry.getValue()._onlineInstances;
Map<String, String> idealStateInstanceStateMap =
idealStateAssignment.get(segment);
- Map<String, String> sortedIdealStateInstanceStateMap =
convertToSortedMap(idealStateInstanceStateMap);
+ Map<String, String> sortedIdealStateInstanceStateMap =
sortByKeyIfNeeded(idealStateInstanceStateMap);
Set<String> unavailableInstances =
unavailableInstancesMap.getOrDefault(idealStateInstanceStateMap.keySet(),
Set.of());
List<SegmentInstanceCandidate> candidates = new
ArrayList<>(idealStateInstanceStateMap.size());
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 64bed8992ea..421f3fa9288 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
@@ -29,12 +29,14 @@ import java.util.Arrays;
import java.util.Collections;
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.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.helix.model.ExternalView;
@@ -81,7 +83,9 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotSame;
import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
@@ -2406,4 +2410,189 @@ public class InstanceSelectorTest {
assertEquals(first.getSegmentToInstanceMap(), expectedInstances);
assertEquals(first.getOptionalSegmentToInstanceMap(), Map.of("optional",
"instance128"));
}
+
+ @Test
+ public void testSortByKeyIfNeeded() {
+ Map<String, String> sorted = new LinkedHashMap<>();
+ sorted.put("a", ONLINE);
+ sorted.put("b", ONLINE);
+ sorted.put("c", ONLINE);
+ assertSame(BaseInstanceSelector.sortByKeyIfNeeded(sorted), sorted);
+
+ Map<String, String> treeMap = new TreeMap<>(sorted);
+ assertSame(BaseInstanceSelector.sortByKeyIfNeeded(treeMap), treeMap);
+
+ Map<String, String> unsorted = new LinkedHashMap<>();
+ unsorted.put("c", ONLINE);
+ unsorted.put("a", ONLINE);
+ unsorted.put("b", ONLINE);
+ Map<String, String> sortedCopy =
BaseInstanceSelector.sortByKeyIfNeeded(unsorted);
+ assertNotSame(sortedCopy, unsorted);
+ assertEquals(new ArrayList<>(sortedCopy.keySet()), List.of("a", "b", "c"));
+ assertEquals(sortedCopy, unsorted);
+ }
+
+ /// Segments with equal ideal state and external view instance state maps
share one candidate list within a
+ /// rebuild. Checks that the sharing does not change any routing outcome, by
comparing a selector that shares with
+ /// one that builds a list per segment, and that the candidates keep the
sorted instance order when the instance
+ /// state maps are not in sorted order.
+ @Test(dataProvider = "selectorType")
+ public void testSharedCandidateListsMatchPerSegmentLists(String
selectorType) {
+ // Partition 0: the ideal state lists the instances out of order. One of
its old segments has a replica OFFLINE in
+ // the external view.
+ // Partition 1: one of its instances is disabled.
+ String[] partition0Instances = {"instance_c", "instance_a", "instance_b"};
+ String[] partition1Instances = {"instance_d", "instance_e", "instance_f"};
+ Set<String> enabledInstances =
+ Set.of("instance_a", "instance_b", "instance_c", "instance_d",
"instance_e", "instance_g");
+
+ Map<String, Map<String, String>> idealStateMaps = new LinkedHashMap<>();
+ Map<String, Map<String, String>> externalViewMaps = new LinkedHashMap<>();
+ for (int i = 0; i < 4; i++) {
+ idealStateMaps.put("p0_" + i, instanceStateMap(partition0Instances,
ONLINE, ONLINE, ONLINE));
+ externalViewMaps.put("p0_" + i, i == 3 ?
instanceStateMap(partition0Instances, ONLINE, ONLINE, OFFLINE)
+ : instanceStateMap(partition0Instances, ONLINE, ONLINE, ONLINE));
+ }
+ for (int i = 0; i < 3; i++) {
+ String state = i == 2 ? CONSUMING : ONLINE;
+ idealStateMaps.put("p1_" + i, instanceStateMap(partition1Instances,
state, state, state));
+ externalViewMaps.put("p1_" + i, instanceStateMap(partition1Instances,
state, state, state));
+ }
+ // Old segment missing from the external view: unavailable
+ idealStateMaps.put("p2_0", instanceStateMap(new String[]{"instance_g"},
ONLINE));
+ IdealState initialIdealState =
createInsertionOrderedIdealState(idealStateMaps);
+ ExternalView initialExternalView =
createInsertionOrderedExternalView(externalViewMaps);
+ Set<String> initialOnlineSegments = new HashSet<>(idealStateMaps.keySet());
+
+ // New segments, first seen by the assignment change: one missing from the
external view, one partially online
+ idealStateMaps.put("new_0", instanceStateMap(partition0Instances, ONLINE,
ONLINE, ONLINE));
+ idealStateMaps.put("new_1", instanceStateMap(partition0Instances, ONLINE,
ONLINE, ONLINE));
+ externalViewMaps.put("new_1", Map.of("instance_a", ONLINE));
+ IdealState idealState = createInsertionOrderedIdealState(idealStateMaps);
+ ExternalView externalView =
createInsertionOrderedExternalView(externalViewMaps);
+ Set<String> onlineSegments = new HashSet<>(idealStateMaps.keySet());
+
+ BaseInstanceSelector sharingSelector = (BaseInstanceSelector)
createTestInstanceSelector(selectorType,
+ enabledInstances, initialIdealState, initialExternalView,
initialOnlineSegments);
+ sharingSelector.onAssignmentChange(idealState, externalView,
onlineSegments);
+ BaseInstanceSelector perSegmentSelector = (BaseInstanceSelector)
createTestInstanceSelector(selectorType,
+ enabledInstances, initialIdealState, initialExternalView,
initialOnlineSegments);
+ perSegmentSelector._maxSharedCandidateLists = 0;
+ perSegmentSelector.onAssignmentChange(idealState, externalView,
onlineSegments);
+ // Shares the lists of one pair of instance state maps only, so that one
rebuild mixes shared and own lists
+ BaseInstanceSelector boundedSelector = (BaseInstanceSelector)
createTestInstanceSelector(selectorType,
+ enabledInstances, initialIdealState, initialExternalView,
initialOnlineSegments);
+ boundedSelector._maxSharedCandidateLists = 1;
+ boundedSelector.onAssignmentChange(idealState, externalView,
onlineSegments);
+
+ // Same candidates for every segment
+ SegmentStates sharedStates = sharingSelector._segmentStates;
+ SegmentStates perSegmentStates = perSegmentSelector._segmentStates;
+ SegmentStates boundedStates = boundedSelector._segmentStates;
+ for (String segment : onlineSegments) {
+ assertEquals(describe(sharedStates.getCandidates(segment)),
describe(perSegmentStates.getCandidates(segment)),
+ segment);
+ assertEquals(describe(boundedStates.getCandidates(segment)),
describe(perSegmentStates.getCandidates(segment)),
+ segment);
+ }
+ assertEquals(boundedStates.getUnavailableSegments(),
perSegmentStates.getUnavailableSegments());
+ assertEquals(boundedStates.getServingInstances(),
perSegmentStates.getServingInstances());
+ assertEquals(sharedStates.getUnavailableSegments(), Set.of("p2_0"));
+ assertEquals(sharedStates.getUnavailableSegments(),
perSegmentStates.getUnavailableSegments());
+ assertEquals(sharedStates.getServingInstances(),
perSegmentStates.getServingInstances());
+ TableReplicaHealth sharedHealth = sharingSelector.getReplicaHealth();
+ TableReplicaHealth perSegmentHealth =
perSegmentSelector.getReplicaHealth();
+ assertEquals(sharedHealth.getMinPercentOfReplicas(),
perSegmentHealth.getMinPercentOfReplicas());
+ assertEquals(sharedHealth.getNumSegmentsAtMinPercentOfReplicas(),
+ perSegmentHealth.getNumSegmentsAtMinPercentOfReplicas());
+ assertEquals(sharedHealth.getNumUnavailableSegments(),
perSegmentHealth.getNumUnavailableSegments());
+
+ // Candidates are in sorted instance order, with the replica id of the
sorted ideal state, whatever the order of
+ // the instance state maps
+ boolean strict =
selectorType.equals(STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE);
+ // Strict replica-group takes instance_b out of service for the whole
partition, because it is OFFLINE for p0_3
+ assertEquals(describe(sharedStates.getCandidates("p0_0")),
+ strict ? List.of("instance_a/0/true", "instance_c/2/true")
+ : List.of("instance_a/0/true", "instance_b/1/true",
"instance_c/2/true"));
+ assertEquals(describe(sharedStates.getCandidates("p0_3")),
List.of("instance_a/0/true", "instance_c/2/true"));
+ // instance_f is disabled
+ assertEquals(describe(sharedStates.getCandidates("p1_0")),
List.of("instance_d/0/true", "instance_e/1/true"));
+ assertEquals(describe(sharedStates.getCandidates("new_0")),
+ strict ? List.of("instance_a/0/false", "instance_c/2/false")
+ : List.of("instance_a/0/false", "instance_b/1/false",
"instance_c/2/false"));
+ assertEquals(describe(sharedStates.getCandidates("new_1")),
+ strict ? List.of("instance_a/0/true", "instance_c/2/false")
+ : List.of("instance_a/0/true", "instance_b/1/false",
"instance_c/2/false"));
+
+ // The sharing selector shares the lists of segments with equal instance
state maps, the other one does not
+ assertSame(sharedStates.getCandidates("p0_0"),
sharedStates.getCandidates("p0_1"));
+ assertSame(sharedStates.getCandidates("p0_0"),
sharedStates.getCandidates("p0_2"));
+ assertNotSame(perSegmentStates.getCandidates("p0_0"),
perSegmentStates.getCandidates("p0_1"));
+
+ // Same selection for every request
+ List<String> segments = new ArrayList<>(idealStateMaps.keySet());
+ for (long requestId = 0; requestId < 6; requestId++) {
+ InstanceSelector.SelectionResult shared =
sharingSelector.select(_brokerRequest, segments, requestId);
+ InstanceSelector.SelectionResult perSegment =
perSegmentSelector.select(_brokerRequest, segments, requestId);
+ InstanceSelector.SelectionResult bounded =
boundedSelector.select(_brokerRequest, segments, requestId);
+ assertEquals(shared.getSegmentToInstanceMap(),
perSegment.getSegmentToInstanceMap());
+ assertEquals(shared.getOptionalSegmentToInstanceMap(),
perSegment.getOptionalSegmentToInstanceMap());
+ assertEquals(shared.getUnavailableSegments(),
perSegment.getUnavailableSegments());
+ assertEquals(shared.getUnavailableSegments(), List.of("p2_0"));
+ assertEquals(bounded.getSegmentToInstanceMap(),
perSegment.getSegmentToInstanceMap());
+ assertEquals(bounded.getOptionalSegmentToInstanceMap(),
perSegment.getOptionalSegmentToInstanceMap());
+ }
+
+ // Enabling the disabled instance keeps both selectors in step
+ Set<String> allInstances = new HashSet<>(enabledInstances);
+ allInstances.add("instance_f");
+ sharingSelector.onInstancesChange(allInstances, List.of("instance_f"));
+ perSegmentSelector.onInstancesChange(allInstances, List.of("instance_f"));
+
assertEquals(describe(sharingSelector._segmentStates.getCandidates("p1_0")),
+ List.of("instance_d/0/true", "instance_e/1/true",
"instance_f/2/true"));
+ assertSame(sharingSelector._segmentStates.getCandidates("p1_0"),
+ sharingSelector._segmentStates.getCandidates("p1_1"));
+ for (long requestId = 0; requestId < 6; requestId++) {
+ InstanceSelector.SelectionResult shared =
sharingSelector.select(_brokerRequest, segments, requestId);
+ InstanceSelector.SelectionResult perSegment =
perSegmentSelector.select(_brokerRequest, segments, requestId);
+ assertEquals(shared.getSegmentToInstanceMap(),
perSegment.getSegmentToInstanceMap());
+ assertEquals(shared.getOptionalSegmentToInstanceMap(),
perSegment.getOptionalSegmentToInstanceMap());
+ }
+ }
+
+ /// Returns an insertion-ordered instance state map, keeping the order of
the given instances.
+ private static Map<String, String> instanceStateMap(String[] instances,
String... states) {
+ Map<String, String> instanceStateMap = new LinkedHashMap<>();
+ for (int i = 0; i < instances.length; i++) {
+ instanceStateMap.put(instances[i], states[i]);
+ }
+ return instanceStateMap;
+ }
+
+ private static IdealState createInsertionOrderedIdealState(Map<String,
Map<String, String>> instanceStateMaps) {
+ IdealState idealState = new IdealState(TABLE_NAME);
+ for (Map.Entry<String, Map<String, String>> entry :
instanceStateMaps.entrySet()) {
+ idealState.getRecord().setMapField(entry.getKey(), new
LinkedHashMap<>(entry.getValue()));
+ }
+ return idealState;
+ }
+
+ private static ExternalView createInsertionOrderedExternalView(Map<String,
Map<String, String>> instanceStateMaps) {
+ ExternalView externalView = new ExternalView(TABLE_NAME);
+ for (Map.Entry<String, Map<String, String>> entry :
instanceStateMaps.entrySet()) {
+ externalView.getRecord().setMapField(entry.getKey(), new
LinkedHashMap<>(entry.getValue()));
+ }
+ return externalView;
+ }
+
+ private static List<String> describe(@Nullable
List<SegmentInstanceCandidate> candidates) {
+ if (candidates == null) {
+ return null;
+ }
+ List<String> descriptions = new ArrayList<>(candidates.size());
+ for (SegmentInstanceCandidate candidate : candidates) {
+ descriptions.add(candidate.getInstance() + "/" +
candidate.getIdealStateReplicaId() + "/" + candidate.isOnline());
+ }
+ return descriptions;
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
