Copilot commented on code in PR #19221:
URL: https://github.com/apache/pinot/pull/19221#discussion_r3770261551
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/BaseInstanceSelector.java:
##########
@@ -368,6 +464,30 @@ void refreshSegmentStates() {
}
_segmentStates = new SegmentStates(instanceCandidatesMap,
servingInstances, unavailableSegments);
+ _replicaHealth =
+ new SegmentReplicaHealth(minPercentOfReplicas,
numSegmentsWithoutRedundancy, unavailableSegments.size());
+ emitReplicaHealthMetrics(_replicaHealth);
Review Comment:
The grace-period classification is only reconsidered during
`onAssignmentChange`; `onInstancesChange` merely refreshes these cached maps.
If a partially loaded segment receives no later IS/EV change, it remains in
`_newSegmentStateMap` after the configured expiration and all three gauges can
stay healthy indefinitely. Add a clock-driven refresh/promotion at expiration
(and a test that advances the clock without an assignment event).
##########
pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerGauge.java:
##########
@@ -107,7 +107,27 @@ 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
segments. `100` means every segment can be
+ /// served from every replica the ideal state assigns to it; `0` means at
least one segment cannot be served at all.
+ ///
+ /// Segments still classified new (see
+ ///
[org.apache.pinot.spi.utils.CommonConstants.Broker#CONFIG_OF_NEW_SEGMENT_EXPIRATION_SECONDS])
are
+ /// excluded because they are commonly not yet loaded everywhere
Review Comment:
The documented interpretation omits that single-replica segments are
excluded. The gauge can remain 100 even when such a segment is completely
unavailable, so saying 100 covers every segment and 0 covers any unavailable
segment is misleading for dashboard users.
##########
pinot-broker/src/test/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelectorTest.java:
##########
@@ -1927,4 +1934,511 @@ public void testReplicaGroupAdaptiveServerSelector() {
assertEquals(selectedResult.getLeft(), expectedSelection);
}
+
+ // Replica health metrics
+ //
+ // The scenarios below all use the same three instances and assert on the
SegmentReplicaHealth 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;
+ }
+
+ 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()));
+
+ SegmentReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 100);
+ assertEquals(replicaHealth.getNumUnavailableSegments(), 0);
+ assertEquals(replicaHealth.getNumSegmentsWithoutRedundancy(), 0);
+ // Nothing is degraded, so no expected replica count has to be remembered
+ assertTrue(selector._oldSegmentExpectedReplicasMap.isEmpty());
+ assertEquals(replicaHealth.getNumSegmentsWithoutRedundancy(), 0);
+ }
+
+ /// 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)));
+
+ SegmentReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 0);
+ assertEquals(replicaHealth.getNumSegmentsWithoutRedundancy(), 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()));
+
+ SegmentReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 100);
+ assertEquals(replicaHealth.getNumSegmentsWithoutRedundancy(), 0);
+ 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));
+ SegmentReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getNumSegmentsWithoutRedundancy(), 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))));
+
+ SegmentReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 100);
+ assertEquals(replicaHealth.getNumSegmentsWithoutRedundancy(), 0);
+ }
+
+
+
+
+
+
+
+
+ @Test(dataProvider = "selectorType")
+ public void testWithoutRedundancyIgnoresSingleReplicaSegments(String
selectorType) {
+ // A segment the ideal state assigns one replica has no redundancy to
lose, so it must never be counted -
+ // otherwise a table that is single-replica by design reads as permanently
at risk.
+ 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));
+ SegmentReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getNumSegmentsWithoutRedundancy(), 0);
+ }
+
+ @Test
+ public void testWithoutRedundancyCountsLastReplicaNotMerelyShort() {
+ // The whole point of a percentage: 2 of 3 is 66% and passes, 1 of 3 is
33% and does not. Pinned because
+ // this is the line the controller-side alert drew, and moving it silently
would change what pages.
+ // Balanced routing only: under strict replica groups segment1's gaps
would exclude those groups for
+ // segment0 as well, taking it to 1 of 3 and hiding the threshold 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));
+
assertEquals(selector.getReplicaHealth().getNumSegmentsWithoutRedundancy(), 1);
+ }
+
+
+
+ @Test(dataProvider = "selectorType")
+ public void
testWithoutRedundancyCountsTwoReplicaSegmentOnItsLastReplica(String
selectorType) {
+ // This is where the count parts company with the alert's percentage. 1 of
2 is 50%, above the threshold,
+ // so the percentage leaves it alone - but the segment genuinely has no
redundancy, and nothing alerts on
+ // the count, so it is reported. Both segments count: segment0 at 1 of 2
and segment1 at 1 of 3.
+ 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));
+ SegmentReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getNumSegmentsWithoutRedundancy(), 2);
+ // The percentage still reports the worst of the two, which is the 1-of-3
segment
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 33);
+ }
+
+ @Test(dataProvider = "selectorType")
+ public void
testWithoutRedundancyCountsTwoReplicaSegmentWithNoReplicaLeft(String
selectorType) {
+ // Leaving 1 of 2 alone must not extend to 0 of 2: the data is gone, and
0% is below the threshold like
+ // any other total loss. A replica-count rule keyed on "three or more
assigned" would miss this.
+ 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));
+ SegmentReplicaHealth replicaHealth = selector.getReplicaHealth();
+ long nowMs = _mutableClock.millis();
+ assertEquals(replicaHealth.getNumSegmentsWithoutRedundancy(), 1);
+ }
+
+ @Test
+ public void testWithoutRedundancyWatchesReplicatedPartOfMixedTable() {
+ // 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 not stop that from happening.
+ 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));
+ SegmentReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getNumSegmentsWithoutRedundancy(), 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)));
+
+ SegmentReplicaHealth 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)));
+
+ SegmentReplicaHealth replicaHealth = selector.getReplicaHealth();
+ assertEquals(replicaHealth.getMinPercentOfReplicas(), 0);
+ assertEquals(replicaHealth.getNumUnavailableSegments(), 1);
+ // An unavailable segment is not also counted as under-replicated
Review Comment:
This comment contradicts the new metric contract and implementation: a
replicated segment with zero serving replicas is counted in
`SEGMENTS_WITHOUT_REDUNDANCY` as well as `UNAVAILABLE_SEGMENTS`. Assert that
value here instead, so the zero-replica behavior is covered.
This issue also appears in the following locations of the same file:
- line 2274
- line 2358
- line 2437
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java:
##########
@@ -693,6 +694,14 @@ public void buildRouting(String tableNameWithType) {
_globalLock.readLock().lock();
try {
buildRoutingInternal(tableNameWithType);
+ } catch (Exception e) {
+ // Creating the instance selector registers the table's replica health
gauges, which happens before the
+ // routing entry is stored. If the build failed in between there is no
routing entry to clean them up
+ // later, so they would keep being exported frozen at a value that no
longer describes the table
+ if (!_routingEntryMap.containsKey(tableNameWithType)) {
+ BaseInstanceSelector.removeReplicaHealthMetrics(_brokerMetrics,
tableNameWithType);
+ }
Review Comment:
This check does not handle failed rebuilds. The old routing entry keeps
`containsKey` true, but the newly created selector has already overwritten the
table gauges; if any later build step fails, queries continue using the old
routing while the gauges describe the discarded selector. Publish metrics only
after the new entry is installed, or explicitly restore the old selector's
values and clean up the failed candidate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
