Repository navigation
Upstream/fail open sse - #19908
Upstream/fail open sse#19908kpatel006 wants to merge 3 commits into
Conversation
xiangfu0
left a comment
There was a problem hiding this comment.
Thorough review across correctness, mixed-version compatibility, concurrency/distributed state, and security. This is well-structured work — the all-or-nothing coverage, the healthy-mapping-never-overridden guarantee, the admin-disabled exclusion, and keeping recovery owned by the failure detector are all sound, and the test coverage is genuinely thorough.
I couldn't find any correctness, compatibility, or concurrency defect: the global write lock serializes the exclude/include/instance-config paths so the fail-open-eligible snapshots are consistent, the new InstanceSelector default methods preserve behavior for the MSE/strict selectors, old constructors are all retained, and nothing touches a wire or serialized format. The snapshot-vs-live-unhealthy-set window degrades safely (falls back to reporting unavailable).
Leaving this as a comment since the PR is still a draft. Only a couple of minor nits inline — nothing blocking.
| public static final BrokerMeter ROUTING_TABLE_UPDATE_FAILURE = | ||
| create("ROUTING_TABLE_UPDATE_FAILURE", "failures", false); | ||
|
|
||
| // Panic routing metrics |
There was a problem hiding this comment.
Small naming nit: this comment says "Panic routing metrics", but the feature, config key, and meter names all use "fail-open" everywhere else. Looks like leftover naming from an earlier iteration — worth renaming to "Fail-open routing metrics" so operators grepping for the feature find it.
There was a problem hiding this comment.
Its based off another PR, which has this removed: #19891. Once its merged, I'll rebase and it should go away
| for (String segment : unavailableSegments) { | ||
| List<SegmentInstanceCandidate> candidates = failOpenSegmentStates.getCandidates(segment); | ||
| if (candidates != null && !candidates.isEmpty()) { | ||
| SegmentInstanceCandidate candidate = candidates.get(requestId++ % candidates.size()); |
There was a problem hiding this comment.
Minor defensive nit, not a live bug: this uses plain % on requestId, while ReplicaGroupInstanceSelector.selectServers uses Math.floorMod(...) for the same index. The request id generator is always non-negative today so % is safe, but matching floorMod here (and at the pre-existing line 95) would make it robust if that ever changes.
PR flow
Successful fail-open routing flow when all unavailable segments have eligible unhealthy replicas.
AI-generated · Green: added · Yellow: modified · Red: removed · Gray: existing
Partial evidence: 0 file patches omitted; 1 truncated.
Diff evidence
Background
The broker has two separate ideas of which servers it can use:
_enabledServerInstanceMap): servers Helix marks as enabled, i.e. not disabled by an operator._routableServerInstanceMap): enabled servers minus the ones theFailureDetectorhasexcluded after connection or query failures.
Instance selectors only ever see the routable set. If every replica of a segment is excluded, the segment counts as
unavailable. The query then returns partial results and an unavailable-segments error.
Motivation
The failure detector can be wrong or too aggressive: a network blip, a slow GC, or a burst of timeouts across replicas.
When that hits every replica of a segment, the broker drops data it could probably have served, which is a
self-inflicted partial outage. Fail-open routing makes the broker try an unhealthy replica as a last resort instead
of skipping the segment.
It never routes to a server an operator has disabled. It also never overrides a healthy routing choice.
Summary of changes
an operator hasn't disabled.
FailureDetector.also find servers that exist only in a remote cluster.
Follow-up
A separate PR will add fail-open support for multi-stage (MSE) queries,
StrictReplicaGroupInstanceSelector, andMultiStageReplicaGroupSelector.Diagrams
Request-time selection flow
flowchart TD A["Query arrives<br/>BaseInstanceSelector.select()"] --> B["Normal select()<br/>over routable servers"] B --> C{"Any requested segments<br/>unavailable?"} C -- No --> R1["Return normal mapping"] C -- Yes --> D{"Any of them have<br/>enabled-but-excluded replicas?"} D -- No --> R2["Return mapping +<br/>unavailable segments"] D -- Yes --> M["Emit FAIL_OPEN_ELIGIBLE_REQUESTS"] M --> E{"FailOpenPolicy enabled<br/>AND not multi-stage?"} E -- No --> R2 E -- Yes --> F["For each unavailable segment:<br/>enabled replicas ∩ detector.getUnhealthyServers()"] F --> G{"Every segment has<br/>≥1 candidate?"} G -- No --> R2 G -- Yes --> H["selectForFailOpen()<br/>Balanced: round-robin by requestId<br/>ReplicaGroup: fixed-replica off"] H --> I{"Full coverage?"} I -- No --> R2 I -- Yes --> J["Merge: healthy mapping + fail-open picks<br/>record POOL_SEG_QUERIES"] J --> R3["Return complete mapping,<br/>no unavailable segments"]Component wiring and state propagation
flowchart LR subgraph Config BC["Broker conf<br/>pinot.broker.enable.fail.open"] CC["Cluster config override<br/>(dynamic)"] end BC --> FOP["FailOpenPolicy<br/>volatile enabled flag"] CC -- "onChange()<br/>delete reverts to broker value" --> FOP FD["FailureDetector"] -- "exclude/include server" --> BS["BaseBrokerStarter /<br/>MultiClusterHelixBrokerStarter"] BS --> LRM["BrokerRoutingManager (local)"] BS -- "new: forwarded" --> RRM["RemoteClusterBrokerRoutingManager(s)"] FOP --> LRM FD --> LRM FOP --> RRM FD --> RRM LRM -- "onInstancesChange(routable,<br/>failOpenEligible = all enabled, changed)" --> IS["InstanceSelector<br/>(+ sampler selectors)"] RRM --> IS IS -- "rebuild SegmentStates:<br/>failOpenCandidatesMap,<br/>segmentsWithoutCandidates" --> SS["SegmentStates"] SS -- "FAIL_OPEN_ELIGIBLE_SEGMENTS" --> MET["BrokerMetrics"] RH["Request handlers<br/>(gRPC / SingleConn / MSE)"] -- "retryUnhealthyServer →<br/>getServerInstanceForRetry<br/>local, then multi-cluster" --> FDTests added
InstanceSelectorTest: about 17 tests run per selector type. They cover the policy being disabled, themulti-stage gate, the all-or-nothing rule, rejection of disabled or unassigned servers, keeping healthy picks,
deterministic choice by
requestId, the replica-group behaviors, and pool metrics.BrokerRoutingManagerTest: fail-open picks become ordinary routing table entries, for both the main and samplerselectors, and excluded servers stay eligible after an instance config change.
BaseBrokerRequestHandlerTest: covers the fallback order ingetServerInstanceForRetry.MultiClusterHelixBrokerStarterTest: checks that detector notifications reach the remote routing managers,including notifications that arrive before those managers exist.