jasperjiaguo commented on code in PR #13952:
URL: https://github.com/apache/pinot/pull/13952#discussion_r1755231339
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/ReplicaGroupInstanceSelector.java:
##########
@@ -135,37 +137,30 @@ private Pair<Map<String, String>, Map<String, String>>
selectServersUsingRoundRo
}
private Pair<Map<String, String>, Map<String, String>>
selectServersUsingAdaptiveServerSelector(List<String> segments,
- int requestId, SegmentStates segmentStates, List<String> serverRankList)
{
+ int requestId, SegmentStates segmentStates, Map<String, Integer>
serverRankMap) {
Map<String, String> 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.
Map<String, String> optionalSegmentToInstanceMap = new HashMap<>();
- for (String segment : segments) {
+ segments.forEach(segment -> {
// NOTE: candidates can be null when there is no enabled instances for
the segment, or the instance selector has
// not been updated (we update all components for routing in sequence)
List<SegmentInstanceCandidate> candidates =
segmentStates.getCandidates(segment);
if (candidates == null) {
- continue;
+ return; // continue to the next iteration
}
- // Round Robin.
- int numCandidates = candidates.size();
- int instanceIdx = requestId % numCandidates;
- SegmentInstanceCandidate selectedInstance = candidates.get(instanceIdx);
- // Adaptive Server Selection
- // TODO: Support numReplicaGroupsToQuery with Adaptive Server Selection.
- if (!serverRankList.isEmpty()) {
- int minIdx = Integer.MAX_VALUE;
- for (SegmentInstanceCandidate candidate : candidates) {
- int idx = serverRankList.indexOf(candidate.getInstance());
- if (idx == -1) {
- // Let's use the round-robin approach until stats for all servers
are populated.
- selectedInstance = candidates.get(instanceIdx);
- break;
- }
- if (idx < minIdx) {
- minIdx = idx;
- selectedInstance = candidate;
- }
- }
+
+ // Round Robin selection
+ int roundRobinInstanceIdx = requestId % candidates.size();
+ SegmentInstanceCandidate selectedInstance =
candidates.get(roundRobinInstanceIdx);
+
+ // Adaptive Server Selection logic
+ if (!serverRankMap.isEmpty()) {
+ // Use instance with the best rank if all servers have stats
populated, if not use round-robin selected instance
+ selectedInstance = candidates.stream()
+ .filter(candidate ->
serverRankMap.containsKey(candidate.getInstance()))
+ .min(Comparator.comparingInt(candidate ->
+ serverRankMap.getOrDefault(candidate.getInstance(),
Integer.MAX_VALUE)))
+ .orElse(candidates.get(roundRobinInstanceIdx));
Review Comment:
```suggestion
selectedInstance = candidates.stream().anyMatch(candidate ->
!serverRankMap.containsKey(candidate.getInstance())) ?
candidates.get(roundRobinInstanceIdx) :
candidates.stream().min(Comparator.comparingInt(candidate ->
serverRankMap.get(candidate.getInstance()))).orElse(candidates.get(roundRobinInstanceIdx));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
