yashmayya opened a new pull request, #19873:
URL: https://github.com/apache/pinot/pull/19873
## Problem
Multi-stage (MSE) queries can time out for hours after server pods restart
together, when every server uses the same mailbox port (for example 8101).
Two server pods can swap IP addresses during a batch restart. A sender whose
gRPC channel to server A still uses A's old IP then reaches server B. B accepts
the stream as its own and puts the blocks in a receiving mailbox that nothing
reads. A waits for the end of stream until the query times out (error 250). The
sender gets normal acks, so nothing logs an error. The only trace is `Evicting
dangling receiving mailbox` on B, 300 s later.
Broker dispatch (query server port, for example 8100) has the same exposure:
B can run a stage plan meant for A.
With random ports this is rare. A restarted server gets a new port, so
senders use a new channel, and a stale connection hits a closed port.
## Why a channel keeps the old address
grpc-java 1.81 and 1.84 have the same code here:
- The default load balancer is the old `PickFirstLoadBalancer`.
- When a connection drops, the subchannel goes to IDLE and resolves the
hostname once, at that moment. A terminating pod can still be in DNS (for
example with `publishNotReadyAddresses=true`). `DnsNameResolver` also skips a
refresh less than 30 s after the last one.
- The next RPC, maybe much later, connects to that address list. It does not
resolve again.
- If another server now listens on that IP and port, the connection works. A
working connection is never resolved again.
## Fix
1. **Receivers check who a request is for.**
- Requests carry a 64-bit hash of the target's instance id in the
`pinot-target-instance-id-hash` header. The broker adds it to dispatch requests
(stage plans, cancel, explain). For mailbox streams, plans carry it for each
receiver (`MailboxInfo.instanceIdHash`, a new proto field), and senders pass it
on.
- The query server and the mailbox servers (servers and brokers) reject a
request for another instance with `FAILED_PRECONDITION` and a
`pinot-misrouted-to` trailer. A gRPC interceptor does the check after auth,
before a receiving mailbox is created or a plan runs.
2. **Senders react to a rejection.**
- The channel to that address resolves again before the next RPC
(`ManagedChannel.enterIdle()`). This happens at most once a second per address,
because a stale connection fails many streams at once.
- The sending mailbox fails its next data send with
`SERVER_NOT_RESPONDING` and a clear message.
- It also sends one error to the intended receiver over a new stream, so
that the receiver fails now instead of at the deadline. This is best effort: if
DNS still returns the old address, the receiver waits until the deadline, as
before.
- The broker fails the query with `SERVER_NOT_RESPONDING`. It does not
mark the server unhealthy, because the server is fine.
3. **Channels resolve again when a server restarts.** This prevents most
cases before any query fails.
- Servers refresh the mailbox channel to a peer when the peer finishes
startup, also when the channel is READY. `ServerGrpcChannelBackoffResetHandler`
is renamed to `ServerMailboxChannelRefreshHandler`. Before, it only reset the
backoff of a channel in TRANSIENT_FAILURE.
- The broker refreshes the dispatch channel to a server each time the
server becomes routable. Before, it only reset the backoff, and only for
servers that the failure detector had excluded.
4. **Channels to old addresses are closed.** When a server moves to another
address (for example a new random port), servers shut down their mailbox
channel to the old address, and brokers their dispatch channel, unless another
server now uses that address. Before, each restart left one unused channel per
sender.
5. **Observability.**
- Three new meters, on servers (`ServerMeter`) and on brokers
(`BrokerMeter`): `MULTI_STAGE_MISROUTED_REQUESTS_REJECTED`,
`MULTI_STAGE_MISROUTED_REQUESTS_REJECTED_BY_PEER` and
`MULTI_STAGE_ORPHANED_RECEIVING_MAILBOXES` (receiving mailboxes that nothing
read).
- A receive timeout names the senders that did not finish, for example:
`Timed out on stage 2 waiting for data from child stage 3 (no end of stream
from 1 of 46 senders: worker 17 on server-0-1:8101)`.
## Design notes
- **Instance id, not a per-run id.** I first tried a per-run id
(incarnation) that each server publishes in its instance config. But brokers
read it from ZK, and that copy can be stale (unconditional read-modify-write
updates of instance configs, a ZK restore). Then a healthy server rejects all
its queries until it restarts. The instance id never changes, so stale metadata
cannot cause this.
- **A hash, not the id.** A plan repeats the receivers of a stage for each
sender worker. An instance id is about as long as a hostname, so it would
almost double the mailbox part of the plan. The hash adds 9 bytes per receiver
entry. Sender entries do not carry it. Headers also carry ASCII only, and the
hex text of a hash is ASCII for any instance id. The hash is FarmHash
Fingerprint64, which Guava keeps stable across versions.
- **Refresh, not a new channel.** `enterIdle()` keeps the channel object
that concurrent senders hold. Streams in flight finish on the old connection.
It is marked `@ExperimentalApi` in grpc-java.
- **Not an `:authority` check.** A receiver does not always know the
hostname that brokers use for it. A false mismatch would reject all traffic.
- **Keep-alive does not help.** The wrong server answers pings.
- **TLS.** Hostname verification can block the wrong connection, but not
when servers share a certificate or use a wildcard one.
- **One PR.** Closing channels to old addresses and the timeout message are
not needed for the check. They are here because they come from the same
incident: the first stops channels from piling up with random ports, and the
second makes the next lost stream easy to find. I can split them out if
reviewers prefer.
## Compatibility
- Receivers accept requests without the new header (older senders). Older
receivers ignore the header and the new proto field, which is additive. A
rolling upgrade is safe in any order. Protection starts when the broker, the
sender and the receiver all run this version.
- `pinot.query.runner.reject.misrouted.requests` (default `true`) turns off
the check. Only receivers read it, at startup, so set it on brokers and on
servers.
- API changes:
- `BaseBrokerRoutingManager.setServerReenableCallback` is now
`setServerEnabledCallback`. It runs for every newly enabled server, not only
for servers that the failure detector excluded.
- `QueryDispatcher.resetClientConnectionBackoff`,
`MailboxService.resetConnectBackoff` and `ChannelManager.resetConnectBackoff`
are removed. A refresh also clears the backoff.
- `MailboxService.shutdown()` also shuts down its client channels.
## Tests
- `MailboxMisdeliveryTest` and `DispatchMisrouteTest` reproduce the incident
with real gRPC servers. A test name resolver plays DNS, and a TCP relay plays
the IP address that moves from A to B.
- Without the check (older senders, or check off): the block lands on B as
an orphan and A gets nothing. B runs A's stage plan.
- With the check: B rejects, A gets the error at once, the meters count it
once, and the next stream or plan reaches A. Cancel, explain and streaming
submit are rejected too.
- Unit tests: `ReceiverIdentity` (pinned hash, a negative hash and a
non-ASCII id through the header), plan serialization, mailbox routing, mailbox
assignment for every exchange kind in both planners, the refresh debounce, the
timeout message, the orphan meter, the server refresh handler and the broker
routing callback.
- `QueryRunnerTest` and `ResourceBasedQueriesTest` now run with the check
on. `MultiStageEngineIntegrationTest` checks that a real cluster rejects a
wrong id and accepts its own. With wrong ids in the test servers, these tests
fail, so the check is live.
- The full `pinot-query-runtime`, `pinot-query-planner`, `pinot-broker` and
`pinot-server` suites pass.
## Not in this PR
- **Single-stage queries.** The broker resolves the hostname at each
reconnect, so the risk is lower. A wrong server reports missing segments
(partial results, not a hang). A fix needs the target instance in the thrift
`InstanceRequest`.
- **Time series.** `TimeSeriesQueryDispatcher` has its own channels with the
same exposure.
- **Proactive refresh for brokers and remote clusters.** Channels to a
broker's mailbox, and to servers of other clusters, are not refreshed on
restart. The receiver check still applies to them.
--
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]
