[kpatel006]

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

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)"]

Loading

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