xiangfu0 opened a new pull request, #20023: URL: https://github.com/apache/pinot/pull/20023
Third and last PR of the series that started with #20021 (detect and close half-open broker-to-server channels) and continued in #20022 (make the failure detector probe the channels that carry queries). Independent of both: it applies to master on its own and touches none of their code. ## Problem In the path-stall incident written up in #20021, three broker-to-server connections stalled for 37 minutes. The queries routed to those servers timed out at the broker after 3 s, but their requests stayed in Netty's outbound queue on the stalled channels, and when the path recovered the whole backlog was delivered to the servers and **executed as fresh requests**. The servers had no way to know those requests were stale: `InstanceRequestHandler` records `queryArrivalTimeMs` when it reads a request off the socket, and `ServerQueryRequest#toExecutionContext` derives the execution deadline from that arrival time plus the `timeoutMs` query option, so a request that had spent 37 minutes in a socket buffer still got its full 3 s budget on arrival, for a broker that had dropped the query long before. Two things stop the broker from fixing this by itself: - **The broker's deadline is not on the wire.** `InstanceRequest` carries `timeoutMs` as a relative query option and nothing else about time, and its Thrift definition must not change (mixed-version clusters). - **Netty gives no way to pull one message back out of a channel's outbound buffer.** `ChannelOutboundBuffer#addFlush` marks every entry uncancellable (`promise.setUncancellable()`) and only discards entries whose promise was cancelled *before* that. `writeAndFlush` runs the flush on the event loop right after the write task, microseconds after the broker thread submitted the request, so during a stall every queued request is already flushed and waiting for the kernel to accept it. Writing with a promise and cancelling it when the query times out would therefore not skip anything, and I found no cheap, correct way to skip the flush of a queued request whose query already timed out. The only lever that drops what is queued is closing the channel, which is what the write timeout in #20021 does; the watermark there bounds how much can queue up in the first place. `ServerChannels` now documents this. ## Fix Carry the broker's deadline as a query option, and let the server refuse requests that arrive after it. **Broker** (`BaseSingleStageBrokerRequestHandler#setRequestDeadline`): next to `timeoutMs`, the broker puts `requestDeadlineMs` (`QueryOptionKey.REQUEST_DEADLINE_MS`), the epoch time in milliseconds at which it stops waiting, into the server query options of both table types. For a hybrid table whose two sides have different timeouts that is now plus the longer of the two remaining times, the same value on both sides, because that is how long the broker waits (the side with the shorter timeout keeps its own `timeoutMs`). A user-supplied value is overwritten: it is this broker's deadline. **Server** (`InstanceRequestHandler#rejectIfExpired`): after the request is deserialised and compiled and before it is scheduled, the arrival time is compared with the deadline. A request that arrived more than the clock-skew allowance after its deadline is answered right away with a DataTable carrying `QUERY_SCHEDULING_TIMEOUT` (the same code the scheduler uses when a query waited past its budget; the message carries the arrival time, the deadline and the lateness) and never reaches the `QueryScheduler`. A missing or non-positive deadline means the old behaviour; a malformed value (only an old broker forwards user input unchanged) is answered with the parse error under the request's own id. The response is still sent: it is a few bytes, keeps one response per request, and tells the broker side that the channel is alive (the unanswered-query judgment in #20022 counts any response). Only the Netty query transport checks the deadline; the gRPC server and the multi-stage leaf path bu ild `ServerQueryRequest` on their own and are unchanged. - `pinot.server.request.deadline.clock.skew.ms`, default 1000. The comparison runs on the server's clock, so this allowance has to live on the server rather than the broker, which is the one knob of this series that is not a broker config key. A false drop needs the server's clock to run ahead of the broker's by more than the allowance *plus the query's remaining time*; a server clock behind the broker's only makes the check more lenient. A negative value disables the check, so it can be switched off per server without touching brokers. - New table-level meter `QUERIES_EXPIRED_ON_ARRIVAL`, plus a WARN line at most once a second with the number of requests dropped since the previous line and the latest one's request id, query id, table, arrival time and lateness. The check runs on the Netty event loop and a recovered channel delivers its whole backlog at once, so per-request logging would stall every channel on that loop for the length of the burst. Not done on purpose: capping the server's execution deadline at the broker's deadline (min of arrival + timeout and `requestDeadlineMs`). That would make every query sensitive to a skew above the allowance, whereas the arrival check only fires once a whole timeout has elapsed by the server's clock. It can be a follow-up if wanted. ### How the incident plays out with this change On recovery, the stale backlog is still delivered (nothing on the broker can take it back), but each request is answered in microseconds with a timeout error that the broker ignores (its query is long gone), instead of being executed with a 3 s budget next to live traffic. With #20021 in place as well, the backlog is bounded by the watermark and the channel is closed by the write timeout, so there is little left to deliver; this PR covers whatever still gets through, and the case where a stall is shorter than the write timeout. ## Compatibility No wire format or serialization change: the deadline travels in the existing query options string map on `PinotQuery`. Old servers ignore the option, old brokers do not send it, and a server that gets no option behaves exactly as before. The new config key has a default. The new meter is exported like every other table-level server meter (`ServerPrometheusMetricsTest` covers all `ServerMeter` values). ## Tests - `InstanceRequestHandlerTest`: a request 5 s past its deadline is answered with `QUERY_SCHEDULING_TIMEOUT` and never scheduled; one past its deadline by less than the allowance is executed; no option, `0` and `-1` are executed; a malformed value is answered with the parse error under the right request id and not scheduled; a negative allowance switches the check off and a zero allowance drops any late request. - `QueryRoutingTest#testRequestPastDeadlineIsAnsweredWithoutExecution`: over real Netty from `QueryRouter` to `QueryServer`, an expired request comes back as `COMPLETED` with the timeout error and the scheduler mock is never called; a request with its deadline ahead on the same channel is executed. - `BaseSingleStageBrokerRequestHandlerTest#testRequestDeadlineSetOnServerQueryOptions`: the broker sets `requestDeadlineMs` within `[before + timeoutMs, after + timeoutMs]` of the request, overwriting the user's `SET requestDeadlineMs = 1`; `#testRequestDeadlineOnHybridTableFollowsTheLongerTimeout`: with a 5 s offline and a 60 s realtime timeout, both sides carry the same deadline, derived from the 60 s side. - `QueryOptionsUtilsTest#testGetRequestDeadlineMs`: absent and non-positive values mean no deadline, garbage is rejected. Verified locally: the above plus `YammerServerPrometheusMetricsTest` (exports every `ServerMeter`, including the new one), checkstyle and license checks on `pinot-spi`, `pinot-common`, `pinot-core`, `pinot-broker`. An independent review pass found no critical issue; its major finding (per-request logging on the event loop) and minors (hybrid deadline, malformed option, gRPC scope of the Javadoc, a timing assumption in a test) are addressed here. -- 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]
