This is an automated email from the ASF dual-hosted git repository.
yashmayya 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 ca9e7a3d6ce Add broker-side percent of replica metrics (#19221)
ca9e7a3d6ce is described below
commit ca9e7a3d6ceb4b8035e88d8491c41552329769f3
Author: Jhow <[email protected]>
AuthorDate: Tue Aug 18 03:16:35 2026 +0800
Add broker-side percent of replica metrics (#19221)
---
.../instanceselector/BaseInstanceSelector.java | 78 +++-
.../routing/instanceselector/InstanceSelector.java | 16 +
.../ReplicaGroupInstanceSelector.java | 5 +-
.../instanceselector/TableReplicaHealth.java | 84 ++++
.../routing/manager/BaseBrokerRoutingManager.java | 84 +++-
.../instanceselector/InstanceSelectorTest.java | 467 +++++++++++++++++++++
.../routing/manager/BrokerRoutingManagerTest.java | 271 +++++++++++-
.../apache/pinot/common/metrics/BrokerGauge.java | 25 +-
8 files changed, 1020 insertions(+), 10 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 6fe60b5411b..f322325519d 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
@@ -80,6 +80,15 @@ import static
org.apache.pinot.spi.utils.CommonConstants.Broker.FALLBACK_POOL_ID
/// 2) When there is no state update from helix, new segments won't be retired
because of the time passing (those with
/// creation time more than 5 minutes ago).
/// TODO: refresh new/old segment state where there is no update from helix
for long time.
+///
+/// Alongside the selection state, this class maintains a [TableReplicaHealth]
describing how well the
+/// table's segments are replicated across the instances it can route to,
exposed through
+/// [#getReplicaHealth()]. Because it is derived from what routing can
actually use, it accounts for both
+/// external view divergence and disabled instances, and for the replica
groups that
+/// [ReplicaGroupInstanceSelector] takes out of service wholesale. Measuring
it here and reporting it in
+/// [org.apache.pinot.broker.routing.manager.BaseBrokerRoutingManager] is
deliberate: a selector cannot
+/// tell whether it covers the whole table or a sampled subset of it, and only
the routing manager knows
+/// which one owns the table's gauges.
public abstract class BaseInstanceSelector implements InstanceSelector {
private static final Logger LOGGER =
LoggerFactory.getLogger(BaseInstanceSelector.class);
// To prevent int overflow, reset the request id once it reaches this value
@@ -104,9 +113,19 @@ public abstract class BaseInstanceSelector implements
InstanceSelector {
// Reduce this map to reduce garbage
protected final Map<String, List<SegmentInstanceCandidate>>
_oldSegmentCandidatesMap = new HashMap<>();
protected Map<String, NewSegmentState> _newSegmentStateMap;
+ /// Number of ONLINE/CONSUMING instances in the ideal state, for old
segments that have fewer
+ /// candidates than that. Only used for metrics
+ ///
+ /// Kept sparse deliberately: an absent entry means "as many candidates as
the ideal state assigns",
+ /// so a healthy table stores nothing here. Read it through
[#getExpectedReplicas].
+ protected final Map<String, Integer> _oldSegmentExpectedReplicasMap = new
HashMap<>();
// _segmentStates is needed for instance selection (multi-threaded), so it
is made volatile.
protected volatile SegmentStates _segmentStates;
+ // Published together with _segmentStates and read back by the routing
manager to report the table's
+ // gauges. Volatile so that a reader on another thread cannot see it lagging
the segment states it was
+ // computed alongside.
+ protected volatile TableReplicaHealth _replicaHealth;
protected Map<String, ServerInstance> _enabledServerStore;
@Override
@@ -143,6 +162,43 @@ public abstract class BaseInstanceSelector implements
InstanceSelector {
refreshSegmentStates();
}
+ /// Returns how well the table's segments are currently replicated across
the instances this selector
+ /// can route to. Never null once [#init] has run.
+ @Override
+ public TableReplicaHealth getReplicaHealth() {
+ return _replicaHealth;
+ }
+
+ /// Returns the number of instances the ideal state assigns to the given old
segment in an
+ /// ONLINE/CONSUMING state. See [#_oldSegmentExpectedReplicasMap] for why
this is stored sparsely.
+ private int getExpectedReplicas(String segment, int numCandidates) {
+ return Math.max(1, _oldSegmentExpectedReplicasMap.getOrDefault(segment,
numCandidates));
+ }
+
+ /// Records an old segment's candidates, along with the ideal-state replica
count the replica health needs
+ /// that the candidate list does not preserve. The single entry point for
both structures, so that they
+ /// cannot fall out of step.
+ protected void putOldSegment(String segment, List<SegmentInstanceCandidate>
candidates,
+ Map<String, String> idealStateInstanceStateMap) {
+ _oldSegmentCandidatesMap.put(segment, candidates);
+ // Stored only when it differs from the candidate count, see
_oldSegmentExpectedReplicasMap
+ int numIdealStateReplicas =
getNumInstancesOnlineForRouting(idealStateInstanceStateMap);
+ if (numIdealStateReplicas > candidates.size()) {
+ _oldSegmentExpectedReplicasMap.put(segment, numIdealStateReplicas);
+ }
+ }
+
+ /// Returns the number of instances in the given ideal state assignment that
are ONLINE/CONSUMING.
+ protected static int getNumInstancesOnlineForRouting(Map<String, String>
idealStateInstanceStateMap) {
+ int numOnlineForRouting = 0;
+ for (String state : idealStateInstanceStateMap.values()) {
+ if (isOnlineForRouting(state)) {
+ numOnlineForRouting++;
+ }
+ }
+ return numOnlineForRouting;
+ }
+
/// Returns whether the instance state is online for routing purpose
(ONLINE/CONSUMING).
static boolean isOnlineForRouting(@Nullable String state) {
return SegmentStateModel.ONLINE.equals(state) ||
SegmentStateModel.CONSUMING.equals(state);
@@ -241,6 +297,7 @@ public abstract class BaseInstanceSelector implements
InstanceSelector {
void updateSegmentMaps(IdealState idealState, ExternalView externalView,
Set<String> onlineSegments,
Map<String, Long> newSegmentCreationTimeMap) {
_oldSegmentCandidatesMap.clear();
+ _oldSegmentExpectedReplicasMap.clear();
_newSegmentStateMap = new
HashMap<>(HashUtil.getHashMapCapacity(newSegmentCreationTimeMap.size()));
Map<String, Map<String, String>> idealStateAssignment =
idealState.getRecord().getMapFields();
@@ -269,7 +326,7 @@ public abstract class BaseInstanceSelector implements
InstanceSelector {
_newSegmentStateMap.put(segment, new
NewSegmentState(newSegmentCreationTimeMs, candidates));
} else {
// Old segment
- _oldSegmentCandidatesMap.put(segment, List.of());
+ putOldSegment(segment, List.of(), idealStateInstanceStateMap);
}
} else {
TreeSet<String> onlineInstances =
getOnlineInstances(idealStateInstanceStateMap, externalViewInstanceStateMap);
@@ -297,7 +354,7 @@ public abstract class BaseInstanceSelector implements
InstanceSelector {
}
idealStateReplicaId++;
}
- _oldSegmentCandidatesMap.put(segment, candidates);
+ putOldSegment(segment, candidates, idealStateInstanceStateMap);
}
}
if (_emitSinglePoolSegmentsMetric) {
@@ -326,12 +383,27 @@ public abstract class BaseInstanceSelector implements
InstanceSelector {
new
HashMap<>(HashUtil.getHashMapCapacity(_oldSegmentCandidatesMap.size() +
_newSegmentStateMap.size()));
Set<String> servingInstances = new HashSet<>();
Set<String> unavailableSegments = new HashSet<>();
+ int minPercentOfReplicas = TableReplicaHealth.FULLY_REPLICATED_PERCENT;
+ // Segments seen at exactly minPercentOfReplicas so far. Reset whenever a
lower percentage displaces the
+ // minimum, so that the pair published below always describes the same
population.
+ int numSegmentsAtMinPercentOfReplicas = 0;
for (Map.Entry<String, List<SegmentInstanceCandidate>> entry :
_oldSegmentCandidatesMap.entrySet()) {
String segment = entry.getKey();
List<SegmentInstanceCandidate> candidates = entry.getValue();
List<SegmentInstanceCandidate> enabledCandidates =
getEnabledCandidatesAndAddToServingInstances(candidates,
servingInstances);
+ int expectedReplicas = getExpectedReplicas(segment, candidates.size());
+ int servingReplicas = enabledCandidates.size();
+ if (TableReplicaHealth.shouldMeasure(expectedReplicas)) {
+ int percentOfReplicas = TableReplicaHealth.toPercent(servingReplicas,
expectedReplicas);
+ if (percentOfReplicas < minPercentOfReplicas) {
+ minPercentOfReplicas = percentOfReplicas;
+ numSegmentsAtMinPercentOfReplicas = 1;
+ } else if (percentOfReplicas == minPercentOfReplicas) {
+ numSegmentsAtMinPercentOfReplicas++;
+ }
+ }
if (!enabledCandidates.isEmpty()) {
instanceCandidatesMap.put(segment, enabledCandidates);
} else {
@@ -368,6 +440,8 @@ public abstract class BaseInstanceSelector implements
InstanceSelector {
}
_segmentStates = new SegmentStates(instanceCandidatesMap,
servingInstances, unavailableSegments);
+ _replicaHealth = new TableReplicaHealth(minPercentOfReplicas,
numSegmentsAtMinPercentOfReplicas,
+ unavailableSegments.size());
}
private List<SegmentInstanceCandidate>
getEnabledCandidatesAndAddToServingInstances(
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 b9b1434d223..06491ebd90d 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
@@ -72,6 +72,22 @@ public interface InstanceSelector {
/// Returns the enabled server instances currently serving the table.
Set<String> getServingInstances();
+ /// Returns how well the table's segments are replicated across the
instances this selector can route to,
+ /// or `null` if this selector does not measure it. The routing manager
reads this after each assignment
+ /// and instance change and publishes it as the table's replica health
gauges.
+ ///
+ /// Implementations must:
+ /// - Measure the table's whole segment set, not a subset. The values are
published under table level
+ /// gauges, so numbers derived from a subset would be read as the table's
own.
+ /// - Return an immutable snapshot, recomputed in [#onAssignmentChange] and
[#onInstancesChange] and
+ /// published so that the manager's subsequent read observes it, rather
than a live view.
+ /// - Return `null` consistently rather than only sometimes: a selector that
stops measuring has its
+ /// gauges dropped, so alternating would make the series flicker.
+ @Nullable
+ default TableReplicaHealth getReplicaHealth() {
+ return null;
+ }
+
class SelectionResult {
private final Pair<Map<String, String>, Map<String, String>/*optional
segments*/> _segmentToInstanceMap;
private final List<String> _unavailableSegments;
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 de15a5006c0..debc04a1912 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
@@ -191,6 +191,7 @@ public class ReplicaGroupInstanceSelector extends
BaseInstanceSelector {
void updateSegmentMapsForUpsertTable(IdealState idealState, ExternalView
externalView, Set<String> onlineSegments,
Map<String, Long> newSegmentCreationTimeMap) {
_oldSegmentCandidatesMap.clear();
+ _oldSegmentExpectedReplicasMap.clear();
int newSegmentMapCapacity =
HashUtil.getHashMapCapacity(newSegmentCreationTimeMap.size());
_newSegmentStateMap = new HashMap<>(newSegmentMapCapacity);
@@ -256,7 +257,9 @@ public class ReplicaGroupInstanceSelector extends
BaseInstanceSelector {
}
idealStateReplicaId++;
}
- _oldSegmentCandidatesMap.put(segment, 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()) {
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/TableReplicaHealth.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/TableReplicaHealth.java
new file mode 100644
index 00000000000..0fff10df054
--- /dev/null
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/TableReplicaHealth.java
@@ -0,0 +1,84 @@
+/**
+ * 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 javax.annotation.concurrent.Immutable;
+
+
+/// Table-level view of how well a table's segments are replicated across the
servers this broker can
+/// actually route to: one set of numbers per table, aggregated over its
segments. Computed alongside
+/// [SegmentStates] and published as a whole so the values stay mutually
consistent. The percentage is
+/// `servingReplicas * 100 / expectedReplicas` evaluated per segment and then
minimised, so that a table
+/// whose segments are not uniformly replicated still reads correctly, and
counted so the minimum comes with
+/// the number of segments behind it.
+///
+/// A snapshot, not a history: a segment is reported the moment it drops, and
debouncing belongs in the
+/// alert. Only two populations are left out - single-replica segments, and
segments routing still calls new
+/// (it does not serve those either, so a stuck push is invisible here by
design). Consuming segments count.
+@Immutable
+public class TableReplicaHealth {
+ /// Value reported when a table has no segments to measure.
+ public static final int FULLY_REPLICATED_PERCENT = 100;
+
+ /// Assigned replicas below which a segment is not measured: a
single-replica segment has no redundancy even
+ /// when healthy, so measuring it would report the replication level rather
than an incident.
+ private static final int MIN_MEASURED_REPLICAS = 2;
+
+ private final int _minPercentOfReplicas;
+ private final int _numSegmentsAtMinPercentOfReplicas;
+ private final int _numUnavailableSegments;
+
+ public TableReplicaHealth(int minPercentOfReplicas, int
numSegmentsAtMinPercentOfReplicas,
+ int numUnavailableSegments) {
+ _minPercentOfReplicas = minPercentOfReplicas;
+ _numSegmentsAtMinPercentOfReplicas = numSegmentsAtMinPercentOfReplicas;
+ _numUnavailableSegments = numUnavailableSegments;
+ }
+
+ /// Returns the worst measured segment's replica percentage, or
[#FULLY_REPLICATED_PERCENT] if there are
+ /// none. The minimum, so one unservable segment is not diluted by a large
healthy table.
+ public int getMinPercentOfReplicas() {
+ return _minPercentOfReplicas;
+ }
+
+ /// Returns how many measured segments sit at [#getMinPercentOfReplicas] -
the blast radius behind it, which
+ /// the minimum alone cannot distinguish: one straggler and a whole table
down read the same. `0` only when
+ /// there is nothing to measure, since otherwise some segment is always the
worst one.
+ public int getNumSegmentsAtMinPercentOfReplicas() {
+ return _numSegmentsAtMinPercentOfReplicas;
+ }
+
+ /// Returns how many segments cannot be routed anywhere, whatever their
replication. Matches
+ /// [SegmentStates#getUnavailableSegments()], i.e. what the query path
refuses to serve, so unlike the two
+ /// above it includes single-replica segments. When
[#getMinPercentOfReplicas] is `0`, this is how much of
+ /// the data is already gone rather than merely degraded.
+ public int getNumUnavailableSegments() {
+ return _numUnavailableSegments;
+ }
+
+ /// Returns whether a segment's replication is high enough to measure at all.
+ static boolean shouldMeasure(int expectedReplicas) {
+ return expectedReplicas >= MIN_MEASURED_REPLICAS;
+ }
+
+ /// Returns the percentage of assigned replicas that are serving, truncated
and capped at 100.
+ static int toPercent(int servingReplicas, int expectedReplicas) {
+ return Math.min(FULLY_REPLICATED_PERCENT, servingReplicas *
FULLY_REPLICATED_PERCENT / expectedReplicas);
+ }
+}
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 e0ccad18e36..631b67f63cc 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
@@ -60,6 +60,7 @@ import
org.apache.pinot.broker.routing.adaptiveserverselector.AdaptiveServerSele
import
org.apache.pinot.broker.routing.adaptiveserverselector.AdaptiveServerSelectorFactory;
import org.apache.pinot.broker.routing.instanceselector.InstanceSelector;
import
org.apache.pinot.broker.routing.instanceselector.InstanceSelectorFactory;
+import org.apache.pinot.broker.routing.instanceselector.TableReplicaHealth;
import
org.apache.pinot.broker.routing.segmentmetadata.SegmentZkMetadataFetchListener;
import
org.apache.pinot.broker.routing.segmentmetadata.SegmentZkMetadataFetcher;
import
org.apache.pinot.broker.routing.segmentpartition.SegmentPartitionMetadataManager;
@@ -73,6 +74,7 @@ import
org.apache.pinot.broker.routing.tablesampler.TableSampler;
import org.apache.pinot.broker.routing.tablesampler.TableSamplerFactory;
import org.apache.pinot.broker.routing.timeboundary.TimeBoundaryManager;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metrics.BrokerGauge;
import org.apache.pinot.common.metrics.BrokerMeter;
import org.apache.pinot.common.metrics.BrokerMetrics;
import org.apache.pinot.common.request.BrokerRequest;
@@ -381,6 +383,7 @@ public abstract class BaseBrokerRoutingManager implements
RoutingManager, Cluste
LOGGER.error("Caught unexpected exception while updating routing entry
on segment assignment change for "
+ "table: {}", tableNameWithType, e);
}
+ updateReplicaHealthMetrics(routingEntry);
return true;
}
return false;
@@ -478,7 +481,7 @@ public abstract class BaseBrokerRoutingManager implements
RoutingManager, Cluste
try {
Object tableLock = getRoutingTableBuildLock(tableNameWithType);
synchronized (tableLock) {
- routingEntry.onInstancesChange(_routableServerInstanceMap.keySet(),
changedServers);
+ updateRoutingEntryOnInstancesChange(routingEntry,
_routableServerInstanceMap.keySet(), changedServers);
}
} catch (Exception e) {
LOGGER.error("Caught unexpected exception while updating routing entry
on instances change for table: {}",
@@ -553,7 +556,7 @@ public abstract class BaseBrokerRoutingManager implements
RoutingManager, Cluste
try {
Object tableLock = getRoutingTableBuildLock(tableNameWithType);
synchronized (tableLock) {
- routingEntry.onInstancesChange(_routableServerInstanceMap.keySet(),
changedServers);
+ updateRoutingEntryOnInstancesChange(routingEntry,
_routableServerInstanceMap.keySet(), changedServers);
}
} catch (Exception e) {
LOGGER.error("Caught unexpected exception while updating routing entry
when excluding server: {} for table: {}",
@@ -600,7 +603,7 @@ public abstract class BaseBrokerRoutingManager implements
RoutingManager, Cluste
try {
Object tableLock = getRoutingTableBuildLock(tableNameWithType);
synchronized (tableLock) {
- routingEntry.onInstancesChange(_routableServerInstanceMap.keySet(),
changedServers);
+ updateRoutingEntryOnInstancesChange(routingEntry,
_routableServerInstanceMap.keySet(), changedServers);
}
} catch (Exception e) {
LOGGER.error("Caught unexpected exception while updating routing entry
when including server: {} for table: {}",
@@ -698,6 +701,68 @@ public abstract class BaseBrokerRoutingManager implements
RoutingManager, Cluste
}
}
+ /// Applies an instance change to the routing entry and re-reports its
replica health gauges.
+ ///
+ /// The single entry point for the instance change paths, so that none of
them can apply the change without
+ /// refreshing the gauges it moves. Callers must hold the table's routing
build lock.
+ private void updateRoutingEntryOnInstancesChange(RoutingEntry routingEntry,
Set<String> enabledInstances,
+ List<String> changedInstances) {
+ try {
+ routingEntry.onInstancesChange(enabledInstances, changedInstances);
+ } finally {
+ // See processAssignmentChangeForTable for why the report is in a
finally block. It cannot be folded into
+ // the callers' own try/catch the way it is there: theirs wraps the
synchronized block rather than sitting
+ // inside it, and this must run under the table lock.
+ updateReplicaHealthMetrics(routingEntry);
+ }
+ }
+
+ /// Reports the table's replica health gauges from what its own instance
selector currently measures.
+ /// Called after every change that can move those numbers, so the gauges
track the routing rather than a
+ /// sampling interval.
+ ///
+ /// Only the routing entry's own selector is asked. The per-sampler
selectors see a subset of the table's
+ /// segments and share its name, so reporting from them would overwrite the
table's real values with
+ /// numbers measured over that subset. They still measure their own subset -
the wasted work is a counter
+ /// per sampled segment, cheap enough not to be worth a switch that could be
set wrong.
+ ///
+ /// Callers must hold the table's routing build lock: that is what makes the
`_disabled` read below see the
+ /// value written by the assignment change, and what keeps two changes from
interleaving a report with a
+ /// removal.
+ private void updateReplicaHealthMetrics(RoutingEntry routingEntry) {
+ String tableNameWithType = routingEntry.getTableNameWithType();
+ if (routingEntry.isDisabled()) {
+ // A disabled table has every replica driven OFFLINE on purpose, so
reporting it as unavailable would
+ // be a false alarm. Drop the gauges instead, so the series disappears
rather than reading as an
+ // outage, and reappears when the table is enabled again.
+ removeReplicaHealthMetrics(tableNameWithType);
+ return;
+ }
+ TableReplicaHealth replicaHealth =
routingEntry._instanceSelector.getReplicaHealth();
+ if (replicaHealth == null) {
+ // A custom instance selector that does not measure replica health. Drop
rather than leave whatever a
+ // previous selector reported, so that swapping the table to such a
selector ends the series instead of
+ // freezing it.
+ removeReplicaHealthMetrics(tableNameWithType);
+ return;
+ }
+ // All three are plain values: nothing here depends on the clock, so a
snapshot is the whole truth
+ _brokerMetrics.setValueOfTableGauge(tableNameWithType,
BrokerGauge.PERCENT_OF_REPLICAS,
+ replicaHealth.getMinPercentOfReplicas());
+ _brokerMetrics.setValueOfTableGauge(tableNameWithType,
BrokerGauge.SEGMENTS_AT_MIN_PERCENT_OF_REPLICAS,
+ replicaHealth.getNumSegmentsAtMinPercentOfReplicas());
+ _brokerMetrics.setValueOfTableGauge(tableNameWithType,
BrokerGauge.UNAVAILABLE_SEGMENTS,
+ replicaHealth.getNumUnavailableSegments());
+ }
+
+ /// Stops reporting the table's replica health gauges, so that they do not
keep being exported frozen at a
+ /// value that no longer describes the table.
+ private void removeReplicaHealthMetrics(String tableNameWithType) {
+ _brokerMetrics.removeTableGauge(tableNameWithType,
BrokerGauge.PERCENT_OF_REPLICAS);
+ _brokerMetrics.removeTableGauge(tableNameWithType,
BrokerGauge.UNAVAILABLE_SEGMENTS);
+ _brokerMetrics.removeTableGauge(tableNameWithType,
BrokerGauge.SEGMENTS_AT_MIN_PERCENT_OF_REPLICAS);
+ }
+
private void buildRoutingInternal(String tableNameWithType) {
long buildStartTimeMs = System.currentTimeMillis();
Object tableLock = getRoutingTableBuildLock(tableNameWithType);
@@ -892,6 +957,10 @@ public abstract class BaseBrokerRoutingManager implements
RoutingManager, Cluste
} else {
LOGGER.info("Rebuilt routing for table: {}", tableNameWithType);
}
+ // Reported only once the entry is stored, so that a build that failed
earlier cannot leave gauges
+ // behind with no routing entry to ever clean them up. The IS / EV
re-check below reports again if it
+ // ends up updating the entry.
+ updateReplicaHealthMetrics(routingEntry);
// Check for updates to the IS / EV after adding the routing entry, as
it is possible that the
// processSegmentAssignmentChange() may have run and missed updating
this newly added entry. Only update
@@ -973,6 +1042,10 @@ public abstract class BaseBrokerRoutingManager implements
RoutingManager, Cluste
if (_routingEntryMap.remove(tableNameWithType) != null) {
LOGGER.info("Removed routing for table: {}", tableNameWithType);
+ // Stop reporting the table level gauges owned by the routing,
otherwise they keep being exported
+ // for a table this broker no longer serves
+ removeReplicaHealthMetrics(tableNameWithType);
+
// Remove time boundary manager for the offline part routing if the
removed routing is the real-time part of a
// hybrid table
if (TableNameBuilder.isRealtimeTableResource(tableNameWithType)) {
@@ -1404,6 +1477,10 @@ public abstract class BaseBrokerRoutingManager
implements RoutingManager, Cluste
// inconsistency between components, which is fine because the
inconsistency only exists for the newly changed
// segments and only lasts for a very short time.
void onAssignmentChange(IdealState idealState, ExternalView externalView) {
+ // Derived purely from the ideal state, so it is set up front: a
component failing partway through the
+ // update must not leave this reading the previous ideal state, since
the replica health reporting
+ // decides from it whether the table's gauges are reported or dropped
+ _disabled = !idealState.isEnabled();
Set<String> onlineSegments = getOnlineSegments(idealState);
Set<String> preSelectedOnlineSegments =
_segmentPreSelector.preSelect(onlineSegments);
_segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
preSelectedOnlineSegments);
@@ -1415,7 +1492,6 @@ public abstract class BaseBrokerRoutingManager implements
RoutingManager, Cluste
updateSamplerInfos(idealState, externalView, preSelectedOnlineSegments);
_lastUpdateIdealStateVersion = idealState.getStat().getVersion();
_lastUpdateExternalViewVersion = externalView.getStat().getVersion();
- _disabled = !idealState.isEnabled();
}
void onInstancesChange(Set<String> enabledInstances, List<String>
changedInstances) {
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 03e35d251da..796ee992b1d 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
@@ -33,6 +33,7 @@ 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 org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
@@ -43,6 +44,7 @@ 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.BrokerGauge;
import org.apache.pinot.common.metrics.BrokerMetrics;
import org.apache.pinot.common.request.BrokerRequest;
import org.apache.pinot.common.request.PinotQuery;
@@ -68,9 +70,12 @@ import static
org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.Segmen
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.clearInvocations;
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.assertTrue;
@@ -1927,4 +1932,466 @@ public class InstanceSelectorTest {
assertEquals(selectedResult.getLeft(), expectedSelection);
}
+
+ // Replica health metrics
+ //
+ // The scenarios below all use the same three instances and assert on the
TableReplicaHealth the
+ // selector derives, since that is what the gauges are emitted from. Each
segment's percentage is
+ // measured against the replicas its own ideal state assigns, so segments do
not have to be uniformly
+ // replicated for the numbers to make sense.
+
+ private static final String REPLICA_INSTANCE_0 = "instance0";
+ private static final String REPLICA_INSTANCE_1 = "instance1";
+ private static final String REPLICA_INSTANCE_2 = "instance2";
+ private static final Set<String> REPLICA_INSTANCES =
+ ImmutableSet.of(REPLICA_INSTANCE_0, REPLICA_INSTANCE_1,
REPLICA_INSTANCE_2);
+
+ /// Returns the ideal state assignment placing the segment on all three
instances as ONLINE.
+ private static List<Pair<String, String>> allOnline() {
+ return List.of(new ImmutablePair<>(REPLICA_INSTANCE_0, ONLINE), new
ImmutablePair<>(REPLICA_INSTANCE_1, ONLINE),
+ new ImmutablePair<>(REPLICA_INSTANCE_2, ONLINE));
+ }
+
+ /// Returns an assignment placing the segment on two of the three instances
as ONLINE.
+ private static List<Pair<String, String>> twoReplicas() {
+ return List.of(new ImmutablePair<>(REPLICA_INSTANCE_0, ONLINE), new
ImmutablePair<>(REPLICA_INSTANCE_1, ONLINE));
+ }
+
+ /// Returns an external view assignment where the first `numOnline` of the
three instances are ONLINE
+ /// and the rest are OFFLINE, so the segment looks partially loaded.
+ private static List<Pair<String, String>> partiallyOnline(int numOnline) {
+ List<String> instances = List.of(REPLICA_INSTANCE_0, REPLICA_INSTANCE_1,
REPLICA_INSTANCE_2);
+ List<Pair<String, String>> assignment = new ArrayList<>(instances.size());
+ for (int i = 0; i < instances.size(); i++) {
+ assignment.add(new ImmutablePair<>(instances.get(i), i < numOnline ?
ONLINE : OFFLINE));
+ }
+ return assignment;
+ }
+
+ /// Returns an external view assignment where every instance but
`offlineInstance` is ONLINE, so that
+ /// different segments can be made to lose different replicas.
+ private static List<Pair<String, String>> onlineExcept(String
offlineInstance) {
+ List<Pair<String, String>> assignment = new
ArrayList<>(REPLICA_INSTANCES.size());
+ for (String instance : List.of(REPLICA_INSTANCE_0, REPLICA_INSTANCE_1,
REPLICA_INSTANCE_2)) {
+ assignment.add(new ImmutablePair<>(instance,
instance.equals(offlineInstance) ? OFFLINE : ONLINE));
+ }
+ return assignment;
+ }
+
+ private BaseInstanceSelector createReplicaHealthSelector(String
selectorType, Set<String> enabledInstances,
+ Map<String, List<Pair<String, String>>> idealStateAssignment,
+ Map<String, List<Pair<String, String>>> externalViewAssignment) {
+ // Sorted so that the order the selector looks up segment metadata in is
deterministic, which is what
+ // the stub set up by createSegmentCreationTimes matches on
+ return (BaseInstanceSelector) createTestInstanceSelector(selectorType,
enabledInstances,
+ createIdealState(idealStateAssignment),
createExternalView(externalViewAssignment),
+ new TreeSet<>(externalViewAssignment.keySet()));
+ }
+
+ /// Stubs the segment metadata lookup with the given creation times, in the
order the selector reads them.
+ private void createSegmentCreationTimes(Map<String, Long>
creationTimeMsBySegment) {
+ List<Pair<String, Long>> creationTimes = new
ArrayList<>(creationTimeMsBySegment.size());
+ for (String segment : new TreeSet<>(creationTimeMsBySegment.keySet())) {
+ creationTimes.add(new ImmutablePair<>(segment,
creationTimeMsBySegment.get(segment)));
+ }
+ createSegments(creationTimes);
+ }
+
+ /// Marks the given segments as created long enough ago that they are no
longer treated as new, so that
+ /// they count towards the replica health even though their external view
has not converged.
+ private void createOldSegments(List<String> segments) {
+ long creationTimeMs = _mutableClock.millis() -
NEW_SEGMENT_EXPIRATION_MILLIS - 1;
+ Map<String, Long> creationTimes = new HashMap<>();
+ for (String segment : segments) {
+ creationTimes.put(segment, creationTimeMs);
+ }
+ createSegmentCreationTimes(creationTimes);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void testReplicaHealthFullyReplicated(String selectorType) {
+ // Every segment is ONLINE everywhere the ideal state assigns it
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ Map.of("segment0", allOnline(), "segment1", allOnline()),
+ Map.of("segment0", allOnline(), "segment1", allOnline()));
+
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 100);
+ assertEquals(replicaHealth.getNumUnavailableSegments(), 0);
+ // Every measured segment sits at the minimum when the minimum is 100, so
the count is the measured
+ // population rather than 0 - it counts what the percentage speaks for,
not what is wrong
+ assertEquals(replicaHealth.getNumSegmentsAtMinPercentOfReplicas(), 2);
+ // Nothing is degraded, so no expected replica count has to be remembered
+ assertTrue(selector._oldSegmentExpectedReplicasMap.isEmpty());
+ }
+
+ /// Returns an external view assignment with the first `numOnline` of the
three instances ONLINE and the
+ /// rest in ERROR, i.e. replicas that failed their state transition rather
than merely being offline.
+ private static List<Pair<String, String>> partiallyOnlineRestInError(int
numOnline) {
+ List<String> instances = List.of(REPLICA_INSTANCE_0, REPLICA_INSTANCE_1,
REPLICA_INSTANCE_2);
+ List<Pair<String, String>> assignment = new ArrayList<>(instances.size());
+ for (int i = 0; i < instances.size(); i++) {
+ assignment.add(new ImmutablePair<>(instances.get(i), i < numOnline ?
ONLINE : ERROR));
+ }
+ return assignment;
+ }
+
+ private static List<Pair<String, String>> singleReplica(String state) {
+ return List.of(new ImmutablePair<>(REPLICA_INSTANCE_0, state));
+ }
+
+ /// Returns an ideal state assignment placing the segment on all three
instances as CONSUMING, i.e. a
+ /// segment the controller still considers in progress.
+ private static List<Pair<String, String>> allConsuming() {
+ return List.of(new ImmutablePair<>(REPLICA_INSTANCE_0, CONSUMING),
+ new ImmutablePair<>(REPLICA_INSTANCE_1, CONSUMING), new
ImmutablePair<>(REPLICA_INSTANCE_2, CONSUMING));
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void testConsumingSegmentMeasuredLikeAnyOther(String selectorType) {
+ // Consuming segments are deliberately not special-cased. A partition
whose replicas are all gone will also
+ // raise an ingestion alert, and reconciling that overlap belongs in the
alerting pipeline rather than in a
+ // metric that would otherwise stop meaning what its name says.
+ createOldSegments(List.of("segment0"));
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ Map.of("segment0", allConsuming()), Map.of("segment0",
partiallyOnline(0)));
+
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 0);
+ assertEquals(replicaHealth.getNumSegmentsAtMinPercentOfReplicas(), 1);
+ assertEquals(replicaHealth.getNumUnavailableSegments(), 1);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void testHealthyConsumingSegmentReportsFullyReplicated(String
selectorType) {
+ // The flip side of not excluding them: a partition consuming normally on
every replica must read 100%, or
+ // every real-time table would look permanently degraded. CONSUMING counts
as serving for routing.
+ createOldSegments(List.of("segment0"));
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ Map.of("segment0", allConsuming()), Map.of("segment0",
allConsuming()));
+
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 100);
+ assertEquals(replicaHealth.getNumSegmentsAtMinPercentOfReplicas(), 1);
+ assertEquals(replicaHealth.getNumUnavailableSegments(), 0);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void testCommittedSegmentCountedWhilePeersStillDownloading(String
selectorType) {
+ // The exclusion has to end at the commit, not when the last replica
finishes downloading. The ideal state
+ // turns ONLINE at commit while peers still report CONSUMING, and that
segment is an ordinary immutable
+ // one whose replicas are genuinely missing.
+ createOldSegments(List.of("segment0"));
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ // Committed: ideal state ONLINE everywhere
+ Map.of("segment0", allOnline()),
+ // Only the committer has it; the peers have dropped out rather than
reporting CONSUMING
+ Map.of("segment0", partiallyOnline(1)));
+
+ _mutableClock.fastForward(Duration.ofMillis(NEW_SEGMENT_EXPIRATION_MILLIS
+ 1));
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 33);
+ assertEquals(replicaHealth.getNumSegmentsAtMinPercentOfReplicas(), 1);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void
testCommittingSegmentStillRoutableFromPeersReportingConsuming(String
selectorType) {
+ // The normal commit window: ideal state ONLINE, peers still CONSUMING in
the external view. They remain
+ // routable, so nothing is short of replicas and no clock starts.
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ Map.of("segment0", allOnline()),
+ Map.of("segment0", List.of(new ImmutablePair<>(REPLICA_INSTANCE_0,
ONLINE),
+ new ImmutablePair<>(REPLICA_INSTANCE_1, CONSUMING), new
ImmutablePair<>(REPLICA_INSTANCE_2, CONSUMING))));
+
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 100);
+ assertEquals(replicaHealth.getNumSegmentsAtMinPercentOfReplicas(), 1);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void testSegmentsAtMinPercentIgnoresSingleReplicaSegments(String
selectorType) {
+ // The count is drawn from the same population as the percentage, so a
segment the ideal state assigns one
+ // replica is left out of it too - otherwise a table that is
single-replica by design would report every
+ // one of its segments as sitting at the worst level.
+ createOldSegments(List.of("segment0"));
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ Map.of("segment0", singleReplica(ONLINE)), Map.of("segment0",
singleReplica(OFFLINE)));
+
+ _mutableClock.fastForward(Duration.ofMillis(NEW_SEGMENT_EXPIRATION_MILLIS
* 2));
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 100);
+ // Nothing measured at all, which is the only way the count reaches 0
+ assertEquals(replicaHealth.getNumSegmentsAtMinPercentOfReplicas(), 0);
+ }
+
+ @Test
+ public void testSegmentsAtMinPercentCountsOnlyTheWorstLevel() {
+ // segment0 is at 2 of 3 and segment1 at 1 of 3. The count belongs to the
percentage that is reported, so
+ // only segment1 is in it - a segment that is degraded but better off than
the worst must not inflate the
+ // blast radius the minimum is describing.
+ // Balanced routing only: under strict replica groups segment1's gaps
would exclude those groups for
+ // segment0 as well, taking both segments to 1 of 3 and hiding the
distinction this asserts on.
+ createOldSegments(List.of("segment0", "segment1"));
+ BaseInstanceSelector selector =
createReplicaHealthSelector(BALANCED_INSTANCE_SELECTOR, REPLICA_INSTANCES,
+ Map.of("segment0", allOnline(), "segment1", allOnline()),
+ Map.of("segment0", partiallyOnline(2), "segment1",
partiallyOnline(1)));
+
+ _mutableClock.fastForward(Duration.ofMillis(NEW_SEGMENT_EXPIRATION_MILLIS
* 2));
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 33);
+ assertEquals(replicaHealth.getNumSegmentsAtMinPercentOfReplicas(), 1);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void
testSegmentsAtMinPercentComparesUnevenlyReplicatedSegments(String selectorType)
{
+ // segment0 is down to 1 of its 2 assigned replicas and segment1 to 1 of
its 3. Both have lost all but one
+ // replica, yet the percentages differ - 50 against 33 - and it is the
percentage that decides membership,
+ // so only segment1 is counted. Measuring each segment against its own
assignment is what makes the two
+ // comparable in the first place.
+ createOldSegments(List.of("segment0", "segment1"));
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ Map.of("segment0", twoReplicas(), "segment1", allOnline()),
+ Map.of("segment0", List.of(new ImmutablePair<>(REPLICA_INSTANCE_0,
ONLINE),
+ new ImmutablePair<>(REPLICA_INSTANCE_1, OFFLINE)), "segment1",
partiallyOnline(1)));
+
+ _mutableClock.fastForward(Duration.ofMillis(NEW_SEGMENT_EXPIRATION_MILLIS
* 2));
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ // The percentage reports the worse of the two, which is the 1-of-3 segment
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 33);
+ assertEquals(replicaHealth.getNumSegmentsAtMinPercentOfReplicas(), 1);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void
testSegmentsAtMinPercentCountsTwoReplicaSegmentWithNoReplicaLeft(String
selectorType) {
+ // A two-replica segment is measured like any other, so losing both takes
it to 0% and counts it. A rule
+ // keyed on "three or more assigned" would miss a total loss on a
two-replica table entirely.
+ createOldSegments(List.of("segment0"));
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ Map.of("segment0", twoReplicas()),
+ Map.of("segment0", List.of(new ImmutablePair<>(REPLICA_INSTANCE_0,
OFFLINE),
+ new ImmutablePair<>(REPLICA_INSTANCE_1, OFFLINE))));
+
+ _mutableClock.fastForward(Duration.ofMillis(NEW_SEGMENT_EXPIRATION_MILLIS
* 2));
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 0);
+ assertEquals(replicaHealth.getNumSegmentsAtMinPercentOfReplicas(), 1);
+ }
+
+ @Test
+ public void testSegmentsAtMinPercentWatchesReplicatedPartOfMixedTable() {
+ // A real-time table whose consuming segment lives on one replica while
its completed segments live on
+ // three. Losing two replicas of the completed segment has to be reported,
and the thinly replicated
+ // consuming segment must neither stop that from happening nor be counted
alongside it.
+ createOldSegments(List.of("completed"));
+ BaseInstanceSelector selector =
createReplicaHealthSelector(BALANCED_INSTANCE_SELECTOR, REPLICA_INSTANCES,
+ Map.of("consuming", List.of(new ImmutablePair<>(REPLICA_INSTANCE_0,
CONSUMING)), "completed", allOnline()),
+ Map.of("consuming", List.of(new ImmutablePair<>(REPLICA_INSTANCE_0,
CONSUMING)), "completed",
+ partiallyOnline(1)));
+
+ _mutableClock.fastForward(Duration.ofMillis(NEW_SEGMENT_EXPIRATION_MILLIS
* 2));
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 33);
+ assertEquals(replicaHealth.getNumSegmentsAtMinPercentOfReplicas(), 1);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void testReplicaHealthPartiallyReplicated(String selectorType) {
+ // segment0 is only loaded on 1 of its 3 replicas, which is the threshold
the low replica alert fires on
+ createOldSegments(List.of("segment0"));
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ Map.of("segment0", allOnline()), Map.of("segment0",
partiallyOnline(1)));
+
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 33);
+ assertEquals(replicaHealth.getNumUnavailableSegments(), 0);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void testReplicaHealthUnavailableSegment(String selectorType) {
+ // segment0 is loaded nowhere
+ createOldSegments(List.of("segment0"));
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ Map.of("segment0", allOnline()), Map.of("segment0",
partiallyOnline(0)));
+
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 0);
+ assertEquals(replicaHealth.getNumUnavailableSegments(), 1);
+ // A replicated segment that is unavailable is at 0%, which is as low as
the minimum goes, so it is
+ // counted by both gauges rather than moving from one to the other
+ assertEquals(replicaHealth.getNumSegmentsAtMinPercentOfReplicas(), 1);
+ }
+
+ @Test
+ public void testReplicaHealthReportsWorstSegmentNotAverage() {
+ // segment0 is loaded nowhere while the other two are fully loaded.
Reporting the minimum is what keeps
+ // a single unservable segment from being diluted by a large healthy
table, and the count says how much
+ // of the table that minimum is speaking for - here one segment out of the
three measured.
+ createOldSegments(List.of("segment0"));
+ BaseInstanceSelector selector =
createReplicaHealthSelector(BALANCED_INSTANCE_SELECTOR, REPLICA_INSTANCES,
+ Map.of("segment0", allOnline(), "segment1", allOnline(), "segment2",
allOnline()),
+ Map.of("segment0", partiallyOnline(0), "segment1", allOnline(),
"segment2", allOnline()));
+
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 0);
+ assertEquals(replicaHealth.getNumSegmentsAtMinPercentOfReplicas(), 1);
+ assertEquals(replicaHealth.getNumUnavailableSegments(), 1);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void testReplicaHealthIgnoresNewSegments(String selectorType) {
+ // segment0 was just created and is only loaded on 1 replica. That is
expected right after a push, so
+ // it must not drag the percentage down.
+ createSegmentCreationTimes(Map.of("segment0", _mutableClock.millis()));
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ Map.of("segment0", allOnline()), Map.of("segment0",
partiallyOnline(1)));
+
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 100);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void testReplicaHealthAccountsForDisabledInstance(String
selectorType) {
+ // A disabled instance reduces the replicas the broker can route to just
as much as a missing external
+ // view entry does, and it arrives through onInstancesChange rather than
an assignment change.
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ Map.of("segment0", allOnline()), Map.of("segment0", allOnline()));
+ assertEquals(selector.getReplicaHealth().getMinPercentOfReplicas(), 100);
+
+ selector.onInstancesChange(ImmutableSet.of(REPLICA_INSTANCE_0,
REPLICA_INSTANCE_1),
+ List.of(REPLICA_INSTANCE_2));
+
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 66);
+ assertEquals(replicaHealth.getNumUnavailableSegments(), 0);
+ }
+
+ @Test
+ public void testReplicaHealthReflectsStrictReplicaGroupKnockout() {
+ // The two segments share an ideal state assignment but are each missing a
different instance: segment0
+ // is not loaded on instance2, segment1 is not loaded on instance1. Strict
replica group routing takes
+ // both instances out of service for every segment in the group, so each
segment is left with instance0
+ // alone. The external view on its own only ever shows two of three
replicas missing per segment, so the
+ // group-wide knockout is degradation a metric computed from the external
view cannot see.
+ createOldSegments(List.of("segment0", "segment1"));
+ Map<String, List<Pair<String, String>>> idealStateAssignment =
+ Map.of("segment0", allOnline(), "segment1", allOnline());
+ Map<String, List<Pair<String, String>>> externalViewAssignment =
+ Map.of("segment0", onlineExcept(REPLICA_INSTANCE_2), "segment1",
onlineExcept(REPLICA_INSTANCE_1));
+
+ BaseInstanceSelector strictSelector =
+
createReplicaHealthSelector(STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE,
REPLICA_INSTANCES,
+ idealStateAssignment, externalViewAssignment);
+ TableReplicaHealth strictReplicaHealth = strictSelector.getReplicaHealth();
+ // 1 of 3, not the 2 of 3 the external view would suggest, and both
segments are down to a single replica
+ assertEquals(strictReplicaHealth.getMinPercentOfReplicas(), 33);
+ assertEquals(strictReplicaHealth.getNumSegmentsAtMinPercentOfReplicas(),
2);
+
+ // Without the strict guarantee each segment only loses the replica its
own external view is missing
+ BaseInstanceSelector balancedSelector =
+ createReplicaHealthSelector(BALANCED_INSTANCE_SELECTOR,
REPLICA_INSTANCES, idealStateAssignment,
+ externalViewAssignment);
+ TableReplicaHealth balancedReplicaHealth =
balancedSelector.getReplicaHealth();
+ assertEquals(balancedReplicaHealth.getMinPercentOfReplicas(), 66);
+ // Both segments tie at the milder minimum, so the count stays 2 while the
percentage improves: it tracks
+ // whatever level is currently worst, and is only ever read together with
that level
+ assertEquals(balancedReplicaHealth.getNumSegmentsAtMinPercentOfReplicas(),
2);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void testReplicaHealthWithUnevenlyReplicatedSegments(String
selectorType) {
+ // segment0 is deliberately assigned a single replica, as happens when
only part of a table's data sits
+ // on a tier with fewer replicas. Measuring it against its own ideal state
keeps it at 100 instead of
+ // permanently reporting it as under-replicated.
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ Map.of("segment0", List.of(new ImmutablePair<>(REPLICA_INSTANCE_0,
ONLINE)), "segment1", allOnline()),
+ Map.of("segment0", List.of(new ImmutablePair<>(REPLICA_INSTANCE_0,
ONLINE)), "segment1", allOnline()));
+
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 100);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void testReplicaHealthIgnoresIdealStateOfflineReplicas(String
selectorType) {
+ // An instance the ideal state marks OFFLINE is not expected to serve the
segment, so it must not count
+ // against the segment either.
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ Map.of("segment0",
+ List.of(new ImmutablePair<>(REPLICA_INSTANCE_0, ONLINE), new
ImmutablePair<>(REPLICA_INSTANCE_1, ONLINE),
+ new ImmutablePair<>(REPLICA_INSTANCE_2, OFFLINE))),
+ Map.of("segment0", allOnline()));
+
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 100);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void testReplicaHealthWithNoOldSegments(String selectorType) {
+ // The state right after a table is created: every segment is new, so
there is nothing to measure. It
+ // has to read as fully replicated, otherwise it would trip the very alert
the gauge exists for.
+ createSegmentCreationTimes(Map.of("segment0", _mutableClock.millis()));
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ Map.of("segment0", allOnline()), Map.of("segment0",
partiallyOnline(0)));
+
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 100);
+ // The 100 above is the "nothing to measure" default rather than a
measurement, and the count is what
+ // tells the two apart
+ assertEquals(replicaHealth.getNumSegmentsAtMinPercentOfReplicas(), 0);
+ assertEquals(replicaHealth.getNumUnavailableSegments(), 0);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void testReplicaHealthRecoversOnAssignmentChange(String selectorType)
{
+ // The expected replica counts are cached until the next assignment change
rebuilds them, so a segment
+ // that converges has to stop being reported as degraded.
+ createOldSegments(List.of("segment0"));
+ Map<String, List<Pair<String, String>>> idealStateAssignment =
Map.of("segment0", allOnline());
+ BaseInstanceSelector selector = createReplicaHealthSelector(selectorType,
REPLICA_INSTANCES,
+ idealStateAssignment, Map.of("segment0", partiallyOnline(1)));
+ assertEquals(selector.getReplicaHealth().getMinPercentOfReplicas(), 33);
+
+ selector.onAssignmentChange(createIdealState(idealStateAssignment),
+ createExternalView(Map.of("segment0", allOnline())),
Set.of("segment0"));
+
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 100);
+ // The cached counts have to be dropped on rebuild, or the segment stays
degraded forever
+ assertEquals(selector._oldSegmentExpectedReplicasMap, Map.of());
+ }
+
+ @Test
+ public void testReplicaHealthWithUpsertTableAppliesStrictReplicaGroupRules()
{
+ // An upsert table gets the strict replica-group treatment even under the
plain replica-group selector,
+ // so the group-wide knockout has to be reflected there too.
+ when(_tableConfig.isUpsertEnabled()).thenReturn(true);
+ createOldSegments(List.of("segment0", "segment1"));
+ // Same setup as testReplicaHealthReflectsStrictReplicaGroupKnockout: a
different instance missing per
+ // segment, so 33 can only come from the group-wide knockout and not from
either segment's own view
+ BaseInstanceSelector selector =
+ createReplicaHealthSelector(REPLICA_GROUP_INSTANCE_SELECTOR_TYPE,
REPLICA_INSTANCES,
+ Map.of("segment0", allOnline(), "segment1", allOnline()),
+ Map.of("segment0", onlineExcept(REPLICA_INSTANCE_2), "segment1",
onlineExcept(REPLICA_INSTANCE_1)));
+
+ TableReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 33);
+ assertEquals(replicaHealth.getNumSegmentsAtMinPercentOfReplicas(), 2);
+ }
+
+ @Test
+ public void testReplicaHealthMeasuredForDisabledTable() {
+ // The selector measures a disabled table like any other - it cannot tell
the difference, and it is the
+ // routing manager that decides a disabled table's gauges should be
dropped rather than reported.
+ createOldSegments(List.of("segment0"));
+ IdealState disabledIdealState = createIdealState(Map.of("segment0",
allOnline()));
+ disabledIdealState.enable(false);
+ BalancedInstanceSelector selector = new BalancedInstanceSelector();
+ selector.init(_tableConfig, _propertyStore, _brokerMetrics, null,
_mutableClock, INSTANCE_SELECTOR_CONFIG,
+ REPLICA_INSTANCES, EMPTY_SERVER_MAP, disabledIdealState,
+ createExternalView(Map.of("segment0", partiallyOnline(0))),
Set.of("segment0"));
+
+ assertEquals(selector.getReplicaHealth().getNumUnavailableSegments(), 1);
+ // The selector never touches the replica health gauges itself
+ verify(_brokerMetrics, never()).setValueOfTableGauge(eq(TABLE_NAME),
any(BrokerGauge.class), anyLong());
+ verify(_brokerMetrics, never()).removeTableGauge(eq(TABLE_NAME),
any(BrokerGauge.class));
+ }
}
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
index bc4134ebb61..32be5f024cf 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
@@ -29,23 +29,34 @@ import org.apache.helix.HelixConstants.ChangeType;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
import org.apache.helix.PropertyKey;
+import org.apache.helix.model.ExternalView;
+import org.apache.helix.model.IdealState;
import org.apache.helix.model.InstanceConfig;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.pinot.broker.routing.instanceselector.InstanceSelector;
+import org.apache.pinot.broker.routing.instanceselector.TableReplicaHealth;
import
org.apache.pinot.broker.routing.segmentmetadata.SegmentZkMetadataFetcher;
import
org.apache.pinot.broker.routing.segmentpartition.SegmentPartitionMetadataManager;
import org.apache.pinot.broker.routing.segmentpreselector.SegmentPreSelector;
import org.apache.pinot.broker.routing.segmentpruner.SegmentPruner;
import org.apache.pinot.broker.routing.segmentselector.SegmentSelector;
+import org.apache.pinot.broker.routing.tablesampler.TableSampler;
import org.apache.pinot.broker.routing.timeboundary.TimeBoundaryManager;
+import org.apache.pinot.common.metrics.BrokerGauge;
import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.common.utils.config.TableConfigSerDeUtils;
import org.apache.pinot.core.routing.TablePartitionInfo;
import org.apache.pinot.core.routing.TablePartitionReplicatedServersInfo;
import org.apache.pinot.core.routing.timeboundary.TimeBoundaryInfo;
import org.apache.pinot.core.transport.ServerInstance;
import
org.apache.pinot.core.transport.server.routing.stats.ServerRoutingStatsManager;
+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.builder.TableConfigBuilder;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.apache.zookeeper.data.Stat;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
@@ -55,7 +66,10 @@ import org.testng.annotations.Test;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
@@ -72,6 +86,9 @@ public class BrokerRoutingManagerTest {
private static final int SERVER_PORT = 8000;
private static final String INSTANCE_CONFIGS_PATH = "/CONFIGS/PARTICIPANT";
private static final String TEST_TABLE = "testTable_OFFLINE";
+ private static final List<BrokerGauge> REPLICA_HEALTH_GAUGES =
+ List.of(BrokerGauge.PERCENT_OF_REPLICAS,
BrokerGauge.UNAVAILABLE_SEGMENTS,
+ BrokerGauge.SEGMENTS_AT_MIN_PERCENT_OF_REPLICAS);
private AutoCloseable _mocks;
@@ -244,6 +261,14 @@ public class BrokerRoutingManagerTest {
private static Object createRoutingEntry(String tableNameWithType,
TimeBoundaryManager timeBoundaryManager,
SegmentPartitionMetadataManager partitionMetadataManager, Map<String, ?>
samplerInfos)
throws Exception {
+ return createRoutingEntry(tableNameWithType, timeBoundaryManager,
partitionMetadataManager, samplerInfos,
+ mock(InstanceSelector.class), false);
+ }
+
+ private static Object createRoutingEntry(String tableNameWithType,
TimeBoundaryManager timeBoundaryManager,
+ SegmentPartitionMetadataManager partitionMetadataManager, Map<String, ?>
samplerInfos,
+ InstanceSelector instanceSelector, boolean disabled)
+ throws Exception {
Class<?> routingEntryClass =
Class.forName(BaseBrokerRoutingManager.class.getName() + "$RoutingEntry");
Constructor<?> constructor =
routingEntryClass.getDeclaredConstructor(String.class, String.class,
String.class,
SegmentPreSelector.class, SegmentSelector.class, List.class,
InstanceSelector.class, int.class, int.class,
@@ -252,9 +277,251 @@ public class BrokerRoutingManagerTest {
constructor.setAccessible(true);
return constructor.newInstance(tableNameWithType, "/IDEALSTATES/" +
tableNameWithType,
"/EXTERNALVIEW/" + tableNameWithType, mock(SegmentPreSelector.class),
mock(SegmentSelector.class),
- Collections.<SegmentPruner>emptyList(), mock(InstanceSelector.class),
1, 1,
+ Collections.<SegmentPruner>emptyList(), instanceSelector, 1, 1,
mock(SegmentZkMetadataFetcher.class), timeBoundaryManager,
partitionMetadataManager, null, samplerInfos,
- false);
+ disabled);
+ }
+
+ private static Object createSamplerInfo(InstanceSelector instanceSelector)
+ throws Exception {
+ Class<?> samplerInfoClass =
Class.forName(BaseBrokerRoutingManager.class.getName() + "$SamplerInfo");
+ Constructor<?> constructor =
+ samplerInfoClass.getDeclaredConstructor(TableSampler.class,
SegmentSelector.class, InstanceSelector.class);
+ constructor.setAccessible(true);
+ return constructor.newInstance(mock(TableSampler.class),
mock(SegmentSelector.class), instanceSelector);
+ }
+
+ /// Registers the test server as routable, so that the exclude/include paths
actually walk the routing
+ /// entries instead of short-circuiting.
+ private void enableTestServer() {
+ when(_zkDataAccessor.getChildren(eq(INSTANCE_CONFIGS_PATH), any(),
eq(AccessOption.PERSISTENT), anyInt(), anyInt()))
+ .thenReturn(List.of(createEnabledServerZNRecord(SERVER_INSTANCE_ID)));
+ _routingManager.processClusterChange(ChangeType.INSTANCE_CONFIG);
+ }
+
+ private void verifyReplicaHealthGaugesRemoved() {
+ for (BrokerGauge gauge : REPLICA_HEALTH_GAUGES) {
+ verify(_brokerMetrics).removeTableGauge(TEST_TABLE, gauge);
+ }
+ }
+
+ private void verifyNoReplicaHealthGaugesReported() {
+ for (BrokerGauge gauge : REPLICA_HEALTH_GAUGES) {
+ verify(_brokerMetrics, never()).setValueOfTableGauge(eq(TEST_TABLE),
eq(gauge), anyLong());
+ }
+ }
+
+ @Test
+ public void testRemoveRoutingRemovesReplicaHealthMetrics()
+ throws Exception {
+ // The routing owns the table's replica health gauges, so tearing it down
has to stop them - otherwise
+ // they keep being exported for a table this broker no longer serves
+ putRoutingEntry(TEST_TABLE, createRoutingEntry(TEST_TABLE, null, null,
Map.of()));
+
+ _routingManager.removeRouting(TEST_TABLE);
+
+ verifyReplicaHealthGaugesRemoved();
+ }
+
+ @Test
+ public void testInstanceChangeReportsReplicaHealthMetrics()
+ throws Exception {
+ // The gauges have to track the routing, so an instance change re-reports
what the selector now measures
+ enableTestServer();
+ InstanceSelector instanceSelector = mock(InstanceSelector.class);
+ when(instanceSelector.getReplicaHealth()).thenReturn(new
TableReplicaHealth(33, 2, 1));
+ putRoutingEntry(TEST_TABLE, createRoutingEntry(TEST_TABLE, null, null,
Map.of(), instanceSelector, false));
+
+ _routingManager.excludeServerFromRouting(SERVER_INSTANCE_ID);
+
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.PERCENT_OF_REPLICAS, 33);
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.SEGMENTS_AT_MIN_PERCENT_OF_REPLICAS, 2);
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.UNAVAILABLE_SEGMENTS, 1);
+ }
+
+ @Test
+ public void testReplicaHealthMetricsDroppedForDisabledTable()
+ throws Exception {
+ // Disabling a table drives every replica OFFLINE on purpose, so its
segments look unavailable. Reporting
+ // that would fire the alert for a table nobody expects to be queryable.
Drop the gauges instead, so the
+ // series disappears rather than reading as an outage.
+ enableTestServer();
+ InstanceSelector instanceSelector = mock(InstanceSelector.class);
+ when(instanceSelector.getReplicaHealth()).thenReturn(new
TableReplicaHealth(0, 1, 1));
+ putRoutingEntry(TEST_TABLE, createRoutingEntry(TEST_TABLE, null, null,
Map.of(), instanceSelector, true));
+
+ _routingManager.excludeServerFromRouting(SERVER_INSTANCE_ID);
+
+ verifyReplicaHealthGaugesRemoved();
+ verifyNoReplicaHealthGaugesReported();
+ }
+
+ /// Stubs the ZK reads that `processSegmentAssignmentChangeInternal` makes
for the test table, so that the
+ /// real assignment change path runs against the given ideal state.
+ private void stubSegmentAssignmentChange(IdealState idealState) {
+ String idealStatePath = "/IDEALSTATES/" + TEST_TABLE;
+ String externalViewPath = "/EXTERNALVIEW/" + TEST_TABLE;
+ // Any version other than the one the routing entry currently holds, so
that the change is not skipped
+ Stat changedStat = new Stat();
+ changedStat.setVersion(2);
+ when(_zkDataAccessor.getStats(eq(List.of(idealStatePath)),
eq(AccessOption.PERSISTENT)))
+ .thenReturn(new Stat[]{changedStat});
+ when(_zkDataAccessor.getStats(eq(List.of(externalViewPath)),
eq(AccessOption.PERSISTENT)))
+ .thenReturn(new Stat[]{changedStat});
+ when(_zkDataAccessor.get(eq(idealStatePath), any(),
eq(AccessOption.PERSISTENT)))
+ .thenReturn(idealState.getRecord());
+ when(_zkDataAccessor.get(eq(externalViewPath), any(),
eq(AccessOption.PERSISTENT)))
+ .thenReturn(new ExternalView(TEST_TABLE).getRecord());
+ }
+
+ private static IdealState createIdealState(boolean enabled) {
+ IdealState idealState = new IdealState(TEST_TABLE);
+ idealState.enable(enabled);
+ return idealState;
+ }
+
+ @Test
+ public void testBuildRoutingReportsReplicaHealthGauges()
+ throws Exception {
+ // Building the routing is the only path by which a newly served table's
gauges first appear. Without a
+ // report here they would only materialize at the next unrelated cluster
change, and a table that was
+ // never rebuilt would stay missing from the dashboard entirely.
+ enableTestServer();
+ TableConfig tableConfig =
+ new
TableConfigBuilder(TableType.OFFLINE).setTableName(TableNameBuilder.extractRawTableName(TEST_TABLE))
+ .build();
+ when(_propertyStore.get(eq("/CONFIGS/TABLE/" + TEST_TABLE), any(),
eq(AccessOption.PERSISTENT)))
+ .thenReturn(TableConfigSerDeUtils.toZNRecord(tableConfig));
+ when(_zkDataAccessor.get(eq("/IDEALSTATES/" + TEST_TABLE), any(),
eq(AccessOption.PERSISTENT)))
+ .thenReturn(createIdealState(true).getRecord());
+ when(_zkDataAccessor.get(eq("/EXTERNALVIEW/" + TEST_TABLE), any(),
eq(AccessOption.PERSISTENT)))
+ .thenReturn(new ExternalView(TEST_TABLE).getRecord());
+
+ _routingManager.buildRouting(TEST_TABLE);
+
+ assertTrue(_routingManager.routingExists(TEST_TABLE));
+ // A table with no segments has nothing to measure, so it reads as fully
replicated rather than as an
+ // outage - see TableReplicaHealth
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.PERCENT_OF_REPLICAS, 100);
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.SEGMENTS_AT_MIN_PERCENT_OF_REPLICAS, 0);
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.UNAVAILABLE_SEGMENTS, 0);
+ }
+
+ @Test
+ public void
testAssignmentChangeDropsThenRestoresGaugesAcrossDisableAndEnable()
+ throws Exception {
+ // Drives the real assignment change path rather than injecting the
disabled flag, since that is the only
+ // thing that ever flips it. Both directions matter: dropping the gauges
of a disabled table is only
+ // correct if enabling it brings them back, otherwise the table stays
invisible for good.
+ InstanceSelector instanceSelector = mock(InstanceSelector.class);
+ when(instanceSelector.getReplicaHealth()).thenReturn(new
TableReplicaHealth(100, 0, 0));
+ putRoutingEntry(TEST_TABLE, createRoutingEntry(TEST_TABLE, null, null,
Map.of(), instanceSelector, false));
+
+ // Disabling a table drives every replica OFFLINE on purpose, so reporting
it would be a false alarm
+ stubSegmentAssignmentChange(createIdealState(false));
+ _routingManager.processSegmentAssignmentChangeInternal();
+
+ verifyReplicaHealthGaugesRemoved();
+ verifyNoReplicaHealthGaugesReported();
+
+ clearInvocations(_brokerMetrics);
+ stubSegmentAssignmentChange(createIdealState(true));
+ _routingManager.processSegmentAssignmentChangeInternal();
+
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.PERCENT_OF_REPLICAS, 100);
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.SEGMENTS_AT_MIN_PERCENT_OF_REPLICAS, 0);
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.UNAVAILABLE_SEGMENTS, 0);
+ }
+
+ @Test
+ public void testSamplerInstanceSelectorNeverReportsReplicaHealth()
+ throws Exception {
+ // A sampler's selector sees only a sampled subset of the table's segments
and shares the table name with
+ // the selector covering the whole table, so reporting from it would
overwrite the table's real values.
+ // Only the routing entry's own selector is ever asked.
+ enableTestServer();
+ InstanceSelector samplerInstanceSelector = mock(InstanceSelector.class);
+ when(samplerInstanceSelector.getReplicaHealth()).thenReturn(new
TableReplicaHealth(0, 5, 5));
+ InstanceSelector instanceSelector = mock(InstanceSelector.class);
+ when(instanceSelector.getReplicaHealth()).thenReturn(new
TableReplicaHealth(100, 0, 0));
+ putRoutingEntry(TEST_TABLE,
+ createRoutingEntry(TEST_TABLE, null, null, Map.of("sampler",
createSamplerInfo(samplerInstanceSelector)),
+ instanceSelector, false));
+
+ _routingManager.excludeServerFromRouting(SERVER_INSTANCE_ID);
+
+ // The sampler's selector is driven by the change like any other...
+ verify(samplerInstanceSelector).onInstancesChange(any(), any());
+ // ...but is never a source of the table's gauges
+ verify(samplerInstanceSelector, never()).getReplicaHealth();
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.PERCENT_OF_REPLICAS, 100);
+ }
+
+ @Test
+ public void testSelectorThatDoesNotMeasureReplicaHealthDropsTheGauges()
+ throws Exception {
+ // A custom instance selector that does not extend BaseInstanceSelector
returns null replica health. Its
+ // gauges are dropped rather than left at whatever a previous selector
reported, so swapping a table to
+ // such a selector ends the series instead of freezing it.
+ enableTestServer();
+ InstanceSelector instanceSelector = mock(InstanceSelector.class);
+ when(instanceSelector.getReplicaHealth()).thenReturn(null);
+ putRoutingEntry(TEST_TABLE, createRoutingEntry(TEST_TABLE, null, null,
Map.of(), instanceSelector, false));
+
+ _routingManager.excludeServerFromRouting(SERVER_INSTANCE_ID);
+
+ // Asserted positively first, so the never() checks below cannot pass just
because the reporting path
+ // was never reached at all
+ verify(instanceSelector).getReplicaHealth();
+ verifyReplicaHealthGaugesRemoved();
+ verifyNoReplicaHealthGaugesReported();
+ }
+
+ @Test
+ public void testReplicaHealthStillReportedWhenTheAssignmentChangeThrows()
+ throws Exception {
+ // RoutingEntry.onAssignmentChange updates the instance selector before
the time boundary manager, so by
+ // the time this throws the replica health is already fresh but
unreported. That is the window the
+ // finally block exists for. Real callers reach it through
MultiStageReplicaGroupSelector, whose own
+ // onAssignmentChange recomputes the health via super and then throws from
getInstancePartitions() when
+ // the instance partitions ZNode is missing.
+ InstanceSelector instanceSelector = mock(InstanceSelector.class);
+ when(instanceSelector.getReplicaHealth()).thenReturn(new
TableReplicaHealth(0, 4, 4));
+ TimeBoundaryManager timeBoundaryManager = mock(TimeBoundaryManager.class);
+ doThrow(new RuntimeException("simulated time boundary manager
failure")).when(timeBoundaryManager)
+ .onAssignmentChange(any(), any(), any());
+ putRoutingEntry(TEST_TABLE,
+ createRoutingEntry(TEST_TABLE, timeBoundaryManager, null, Map.of(),
instanceSelector, false));
+ stubSegmentAssignmentChange(createIdealState(true));
+
+ _routingManager.processSegmentAssignmentChangeInternal();
+
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.PERCENT_OF_REPLICAS, 0);
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.SEGMENTS_AT_MIN_PERCENT_OF_REPLICAS, 4);
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.UNAVAILABLE_SEGMENTS, 4);
+ }
+
+ @Test
+ public void testReplicaHealthStillReportedWhenTheInstanceChangeThrows()
+ throws Exception {
+ // Same window on the instance change path: RoutingEntry.onInstancesChange
updates the table's own
+ // selector before fanning out to the samplers', and unlike
updateSamplerInfos it does not catch per
+ // sampler, so a failing sampler selector aborts the update after the
health is already fresh.
+ enableTestServer();
+ InstanceSelector instanceSelector = mock(InstanceSelector.class);
+ when(instanceSelector.getReplicaHealth()).thenReturn(new
TableReplicaHealth(0, 3, 3));
+ InstanceSelector samplerInstanceSelector = mock(InstanceSelector.class);
+ doThrow(new RuntimeException("simulated instance selector
failure")).when(samplerInstanceSelector)
+ .onInstancesChange(any(), any());
+ putRoutingEntry(TEST_TABLE,
+ createRoutingEntry(TEST_TABLE, null, null, Map.of("sampler",
createSamplerInfo(samplerInstanceSelector)),
+ instanceSelector, false));
+
+ _routingManager.excludeServerFromRouting(SERVER_INSTANCE_ID);
+
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.PERCENT_OF_REPLICAS, 0);
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.SEGMENTS_AT_MIN_PERCENT_OF_REPLICAS, 3);
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.UNAVAILABLE_SEGMENTS, 3);
}
@SuppressWarnings({"rawtypes", "unchecked"})
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerGauge.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerGauge.java
index bd9374aacc9..9387ac13f8d 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerGauge.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerGauge.java
@@ -107,7 +107,30 @@ public enum BrokerGauge implements AbstractMetrics.Gauge {
/// signals a leak in the ZK listener / drop path.
MATERIALIZED_VIEW_CACHE_ENTRY_COUNT("materializedViewCacheEntries", true),
// Workload config fetch status: 1 = success, 0 = failure
- WORKLOAD_CONFIG_FETCH_STATUS("status", true);
+ WORKLOAD_CONFIG_FETCH_STATUS("status", true),
+
+ /// Replica availability of a table as observed by this broker's routing:
the smallest percentage of
+ /// assigned replicas that are actually routable, across all the table's
measured segments. `100` means every
+ /// measured segment can be served from every replica the ideal state
assigns to it; `0` means at least one measured
+ /// segment cannot be served at all.
+ ///
+ /// Two populations are left out of the measurement, so this gauge can read
`100` while such a segment is
+ /// unavailable - watch [#UNAVAILABLE_SEGMENTS] for those:
+ /// - Segments assigned a single replica, which have no redundancy to report
on and would otherwise pin a table of
+ /// mixed replication (e.g. a tier assigned fewer replicas) to `0` for the
length of any restart or rebalance.
+ /// - Segments still classified new (see
+ ///
[org.apache.pinot.spi.utils.CommonConstants.Broker#CONFIG_OF_NEW_SEGMENT_EXPIRATION_SECONDS]),
which are
+ /// commonly not yet loaded everywhere.
+ PERCENT_OF_REPLICAS("percent", false),
+
+ /// Number of the table's segments that this broker currently cannot route
to any server. Segments assigned a
+ /// single replica are included
+ UNAVAILABLE_SEGMENTS("segments", false),
+
+ /// Number of the table's measured segments that are replicated as poorly as
[#PERCENT_OF_REPLICAS] reports,
+ /// i.e. how many segments that percentage speaks for. The same populations
are excluded, so this reads `0`
+ /// exactly when the table has nothing to measure.
+ SEGMENTS_AT_MIN_PERCENT_OF_REPLICAS("segments", false);
private final String _brokerGaugeName;
private final String _unit;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
