kpatel006 opened a new pull request, #19980:
URL: https://github.com/apache/pinot/pull/19980
## Background
The earlier commits in this stack added a configurable fail-open routing
policy. When every enabled replica of a segment is unavailable, the broker can
route that segment to a replica that the failure detector has marked unhealthy,
instead of reporting the segment as unavailable. The single-stage commit wired
this into single-stage selection, with two limits:
- **MSE was gated off.** `BaseInstanceSelector.select()` turned fail-open
off when `useMultistageEngine` was set (`// TODO: Remove the multi-stage query
gate…`).
- **Fail-open only patched the unavailable segments.** It reselected just
those segments and merged the result into the normal mapping. That works for
balanced and replica-group selectors. It doesn't work for
`MultiStageReplicaGroupSelector`, which must keep every segment of an instance
partition on **one replica group** so colocated joins stay correct. A patch can
put segment A on RG0 and segment B on RG1. The broker picks this selector from
the table's routing config, not the query engine, so the problem affects both
single-stage and multi-stage queries on those tables.
## Summary
### Removing the MSE gate
- Removes the `isUseMultistageEngine` check from
`BaseInstanceSelector.select()`, so fail-open now applies to multi-stage
queries.
### Fail-open reselection changes (all selectors)
- **Full reselection.** `selectFailOpen()` now reruns selection over the
**full per-query segment list** in its original order, not just the unavailable
subset, and returns that mapping as is. Each segment's candidates:
- Healthy segments keep their normal candidates.
- Unavailable segments get their fail-open (unhealthy) candidates.
- Healthy segments that also have unhealthy replicas get those as
**fallback candidates** in a separate map. A selector may use them only as a
last resort.
- **New hooks for subclasses:**
- `shouldTrackFailOpenCandidatesForHealthySegments()`: default `false`;
`true` for Strict and `MultiStageReplicaGroupSelector`.
- `shouldFailOpenSegmentsWithoutHealthyCandidates()`: default `true`;
`false` for `MultiStageReplicaGroupSelector`.
- `canSkipSegmentForFailOpen()`: defaults to the new
`SegmentStates.isNewSegment()`; `MultiStageReplicaGroupSelector` uses
`isNewLLCSegment`.
- **`SegmentStates` tracks new segments** as part of each snapshot, so skip
decisions stay consistent if an assignment refresh races the query.
### Observability
- **Pool metrics in one place.** `POOL_SEG_QUERIES` is now recorded once per
query by `BaseInstanceSelector.recordPoolMetrics()`, from whichever mapping is
actually returned (normal or fail-open). The per-selector recording in
`BalancedInstanceSelector` and `ReplicaGroupInstanceSelector` is removed, which
fixes double counting during fail-open.
- **New meters:** `FAIL_OPEN_ROUTING_SUCCESSES` and
`FAIL_OPEN_ROUTING_FAILURES`.
- **Failure log:** a warning when fail-open fails, rate-limited to 1/s
across all tables and listing at most 5 sample segments.
### Selector changes
- **`BalancedInstanceSelector`:** fail-open now goes through the normal
selection code with adaptive selection turned off, instead of its own simple
round-robin loop. The round-robin sequence continues across fail-open.
- **`ReplicaGroupInstanceSelector`:** fail-open no longer forces
`useFixedReplica=false`, so sticky routing now carries through fail-open.
Adaptive ranking is still bypassed, and the `numReplicaGroupsToQuery` sequence
continues across fail-open.
- **`StrictReplicaGroupInstanceSelector`:** opts into tracking fail-open
candidates for healthy segments. New segments of an unavailable partition now
follow their older siblings onto the same fail-open instance instead of being
skipped.
- **`MultiStageReplicaGroupSelector`:**
- **Two passes in `assign()`.** It first tries the healthy-only map.
During fail-open reselection only, if no replica group can serve a partition,
it retries with the expanded map that includes fallback candidates. Within the
chosen replica group it still prefers healthy instances.
- **Defer instead of throw.** In the normal pass, a partition that can't
be assigned is deferred to fail-open instead of throwing, but only if it
contains an unavailable segment that has fail-open candidates. Otherwise it
throws as before.
- **Fail loudly.** A `select(BrokerRequest, …)` override throws if a
deferred segment ends up neither assigned nor reported unavailable, so a
partition is never silently served with segments missing.
- **Consistent snapshot.** A ThreadLocal hands the `InstancePartitions`
snapshot from normal selection to the fail-open pass, so both passes use the
same partitions. The ThreadLocals are cleared in `finally` on every exit path.
- **Sticky routing on both passes.** `resolvePreferredReplicaGroup()` is
shared, so fixed-replica/sticky routing is honored identically on the normal
and fail-open paths.
## Impact on single-stage queries
This PR is **not** MSE-only. It changes single-stage routing in these ways:
1. **Fail-open reselection works differently for every selector.**
Previously, `selectFailOpen()` reselected only the unavailable segments and
merged them into the normal mapping. Now it reselects the full per-query
segment list and returns that mapping as is.
- **Effect:** healthy segments in a single-stage query are routed again
during fail-open. Usually they land in the same place, but not always. With
adaptive selection, fail-open falls back to round-robin. With
`numReplicaGroupsToQuery`, the newly routed unavailable segments shift the
replica offset. Either case can move a healthy segment to a different *healthy*
server.
2. **Fixed replica is preserved.** `ReplicaGroupInstanceSelector` no longer
overrides `useFixedReplica=false` during fail-open. The test
`testReplicaGroupFailOpenOverridesFixedReplicaQueryOption` was replaced by
`testReplicaGroupFailOpenPreservesFixedReplicaQueryOption`.
3. **Balanced and Strict selectors behave differently during fail-open:**
- **Balanced:** `BalancedInstanceSelector` now uses its normal selection
code, with adaptive selection turned off.
- **Strict:** `StrictReplicaGroupInstanceSelector` routes new sibling
segments of an unavailable partition instead of skipping them.
4. **Pool metrics changed for all queries, including healthy ones.**
`POOL_SEG_QUERIES` is now recorded centrally on every path.
- **Healthy path:** counts should match the old ones.
- **Fail-open path:** the old code counted only the fail-open segments.
The new code counts the whole final mapping, which also fixes double counting.
5. **`MultiStageReplicaGroupSelector` changes also apply to single-stage
queries.** `InstanceSelectorFactory` picks it from the table's routing config
(`MULTI_STAGE_REPLICA_GROUP_SELECTOR_TYPE`), so single-stage queries on those
tables get the new behavior: deferring a partition to fail-open, the fallback
pass, and the new "dropped from the selection" exception.
**When the fail-open policy is disabled:** every reselection change above
sits behind `_failOpenPolicy.isEnabled()`. The only exception is the
pool-metrics refactor, which should be count-equivalent on the healthy path.
## Motivation
- **Close the MSE gap.** MSE queries are the ones that most need a
consistent replica group per partition, and they were the only ones without
fail-open.
- **Correctness first:**
- A partition is never split across replica groups.
- Healthy segments move onto unhealthy servers only when no healthy
assignment exists.
- The broker fails loudly rather than silently dropping segments.
- **Fail-open no longer changes healthy routing, for both engines.**
Reselecting over the full segment list in its original order keeps sticky
routing, round-robin position and pool metrics matching what the normal pass
would have done. The old patch-only approach changed the rotation, overrode
fixed-replica routing, and counted pool metrics twice.
- **Better operational signal:** success/failure meters and a rate-limited
failure log.
## Diagram
```mermaid
flowchart TD
A["BaseInstanceSelector.select(brokerRequest, segments)<br/>(SSE and
MSE; MSE gate removed)"] --> B["select(segments, ...,
failOpenEnabled)<br/>normal pass"]
B --> MSE{"MultiStageReplicaGroupSelector?<br/>(table routing config)"}
MSE -- yes --> M1["Save InstancePartitions snapshot in
ThreadLocal<br/>assign(requireComplete=true)"]
M1 --> M2{"Partition assignable<br/>with healthy servers?"}
M2 -- yes --> C
M2 -- no --> M3{"failOpenEnabled and partition has an<br/>unavailable
segment with fail-open candidates?"}
M3 -- no --> X1["throw: no replica group can serve partition"]
M3 -- yes --> M4["Defer partition's
segments<br/>(_segmentsDeferredToFailOpen)"] --> C
MSE -- "no (Balanced / ReplicaGroup / Strict)" --> C
C{"Unavailable segments<br/>in this request?"}
C -- no --> R1["recordPoolMetrics(normal)<br/>return normal mapping"]
C -- yes --> D{"Fail-open eligible and enabled?"}
D -- no --> R2["recordPoolMetrics(normal)<br/>return normal +
unavailable list"]
D -- yes --> E["selectFailOpen: rebuild states over FULL segment
list<br/>• unavailable → unhealthy candidates<br/>• healthy → normal candidates
+ fallback (unhealthy) list<br/>• new / candidate-less → skip per hooks"]
E --> F["selectForFailOpen()<br/>(adaptive off; sticky/fixed-replica
preserved)"]
F -->
G{"MultiStageReplicaGroupSelector:<br/>assign(requireComplete=false)<br/>healthy
map → fallback map as last resort"}
G --> H{"All required segments assigned?"}
H -- yes -->
S["FAIL_OPEN_ROUTING_SUCCESSES++<br/>recordPoolMetrics(fail-open)<br/>return
fail-open mapping"]
H -- no --> FL["FAIL_OPEN_ROUTING_FAILURES++<br/>rate-limited
WARN<br/>recordPoolMetrics(normal)<br/>return normal mapping + unavailable"]
S --> V
FL --> V{"MultiStageReplicaGroupSelector:<br/>any deferred segment
neither<br/>assigned nor unavailable?"}
V -- yes --> X2["throw: partition dropped,<br/>no consistent replica
group"]
V -- no --> OK["return result<br/>(finally: clear ThreadLocals)"]
```
## Tests added
### `MultiStageReplicaGroupSelectorTest` (22 new tests)
- **Basic fail-open and recovery:**
- `testFailOpenPreservesMultiStageReplicaGroupSelection`
- `testFailOpenReassignsCompletePartitionToFallbackReplicaGroup`
- `testFailOpenRebuildsReplicaGroupWhenHealthyCandidatesAreSplit`
- `testFailOpenRecoversPartitionWithUnavailableSegmentAndHealthySplit`
- **Last-resort fallback rules:**
- `testFailOpenOnlyMovesHealthySegmentsOntoUnhealthyServersAsLastResort`
- `testFailOpenLastResortPrefersHealthyReplicaWithinReplicaGroup`
- `testFailOpenRejectsCandidatesSplitAcrossReplicaGroups`
- **Defer vs. throw:**
- `testFailOpenStillThrowsWhenSplitPartitionHasNoUnavailableSegment`
- `testFailOpenThrowsWhenSplitPartitionFailOpenAlsoFails`
- `testFailedFailOpenReturnsPartialNormalAssignment`
- **New and old segments:**
- `testMseFailOpenSkipsCandidateLessNewLlcSegment`
- `testMseFailOpenDoesNotSkipCandidateLessOldSegment`
- `testFailOpenSkipsNewSegmentWithoutHealthyCandidates`
-
`testFailOpenNewSegmentWithoutHealthyCandidatesDoesNotMoveHealthyPartition`
- **Gating:**
- `testMseFailOpenDisabledForUnavailableSegment`
- `testMseFailOpenNotInvokedForUnavailableSegmentOutsideRequest`
- **Sticky routing:**
- `testFailOpenReselectionHonorsFixedReplicaStickyRouting`
- `testMseFailOpenPreservesFixedReplicaQueryOption`
- **Snapshot and ThreadLocal cleanup:**
- `testSelectForFailOpenUsesInstancePartitionsSnapshotFromPrecedingSelect`
- `testThreadLocalsClearedAfterSuccessfulFailOpen`
- `testThreadLocalClearedAfterHealthySelectionWithFailOpenDisabled`
- `testThreadLocalClearedAfterSelectionThrows`
### `InstanceSelectorTest`
- **Replaced (single-stage behavior changes):**
- `testFailOpenDisabledForMultiStageQuery` →
`testFailOpenEnabledForMultiStageQuery`
- `testReplicaGroupFailOpenOverridesFixedReplicaQueryOption` →
`testReplicaGroupFailOpenPreservesFixedReplicaQueryOption`
- **New, all selector types:**
- `testMultiStageFailOpenSkipsCandidateLessNewLLCSegment`
- `testFailOpenNotInvokedForUnavailableSegmentOutsideRequest`
- `testFailedFailOpenPreservesPartialNormalAssignment`
- **Balanced:**
- `testBalancedFailOpenContinuesRoundRobinSequence`
- `testBalancedFailOpenDoesNotDoubleCountPoolMetrics`
- `testBalancedFailOpenPreservesFixedReplicaQueryOption`
- **Replica group:**
- `testReplicaGroupFailOpenContinuesNumReplicaGroupsToQuerySequence`
- `testReplicaGroupFailOpenNumReplicaGroupsToQueryGreaterThanReplicas`
- **Strict replica group:**
- `testStrictReplicaGroupFailOpenNeverSplitsAcrossMultipleCandidates`
- `testStrictReplicaGroupFailOpenCoversSiblingNewSegment`
- `testStrictReplicaGroupFailOpenKeepsHealthySegmentOnHealthyServer`
-
`testStrictReplicaGroupFailOpenDoesNotMoveHealthyGroupOntoUnhealthyServer`
- `testStrictReplicaGroupFailOpenMatchesIdealWhenAllHealthy`
- `testStrictReplicaGroupFailOpenIsolatedToHealthyPartition`
- `testStrictReplicaGroupFailOpenPreservesFixedReplicaQueryOption`
- `testStrictReplicaGroupFailOpenWithNumReplicaGroupsToQuery`
### `ReplicaGroupSelectorTest`
- `testReplicaGroupFailOpenBypassesAdaptiveRanking`
- `testBalancedFailOpenBypassesAdaptiveRanking`
- `testStrictReplicaGroupFailOpenBypassesAdaptiveWithoutSplittingPartition`
--
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]
