xiangfu0 commented on code in PR #20023:
URL: https://github.com/apache/pinot/pull/20023#discussion_r4368781019
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java:
##########
@@ -150,16 +167,80 @@ protected void channelRead0(ChannelHandlerContext ctx,
ByteBuf msg) {
queryRequest.getTimerContext().startNewPhaseTimer(ServerQueryPhase.REQUEST_DESERIALIZATION,
queryArrivalTimeMs)
.stopAndRecord();
try {
+ if (rejectIfExpired(queryRequest, ctx, queryArrivalTimeMs)) {
+ return;
+ }
submitQuery(queryRequest, ctx, queryArrivalTimeMs);
} catch (Exception e) {
long requestId = queryRequest.getRequestId();
String tableNameWithType = queryRequest.getTableNameWithType();
- LOGGER.error("Caught exception while submitting query request: {}",
requestId, e);
+ LOGGER.error("Caught exception while handling query request: {}",
requestId, e);
sendErrorResponse(ctx, requestId, tableNameWithType, queryArrivalTimeMs,
DataTableBuilderFactory.getEmptyDataTable(), e);
}
}
+ /// Answers a request that reached this server after the broker's deadline
for it with a timeout error, without
+ /// executing it, and returns `true`; returns `false` when the request is to
be executed as usual.
+ ///
+ /// The broker sends its deadline as the `requestDeadlineMs` query option,
so a request from an old broker (no
+ /// option) is always executed. The deadline is compared against this
server's clock, which is why it has to be more
+ /// than [Server#CONFIG_OF_REQUEST_DEADLINE_CLOCK_SKEW_MS] in the past
before the request is dropped. Without this
+ /// check a request that spent minutes in the socket buffers of a stalled
connection, and is delivered when the
+ /// connection recovers, is executed with a full timeout budget measured
from its arrival, for a broker that gave up
+ /// on it long ago. The response is still sent: it costs a few bytes and
tells the broker side that the channel is
+ /// alive, even though the broker no longer has a query waiting for it. A
malformed option propagates to the caller,
+ /// which answers with the parse error under the right request id.
+ private boolean rejectIfExpired(ServerQueryRequest queryRequest,
ChannelHandlerContext ctx,
+ long queryArrivalTimeMs) {
+ if (_requestDeadlineClockSkewMs < 0) {
+ return false;
+ }
+ Long requestDeadlineMs =
QueryOptionsUtils.getRequestDeadlineMs(queryRequest.getQueryContext().getQueryOptions());
Review Comment:
Deliberate. It is how every other numeric option behaves (`timeoutMs`,
`maxServerResponseSizeBytes` and the rest go through the same
`checkedParseLong` and fail the query on garbage), it is only reachable when a
user sets the option through an old broker, and in that case failing loudly is
the better outcome: executing the query would hide a typo nobody would
otherwise notice, while the error names the option and the value. New brokers
overwrite the value, so the path is gone once brokers are upgraded.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java:
##########
@@ -1741,6 +1752,17 @@ public enum AndRestrictionPushdownMode {
public static final String DEFAULT_SERVER_QUERY_REGEX_CLASS = "JAVA_UTIL";
public static final String CONFIG_OF_ENABLE_QUERY_CANCELLATION =
"pinot.server.enable.query.cancellation";
public static final boolean DEFAULT_ENABLE_QUERY_CANCELLATION = true;
+ /// Allowance for clock skew between a broker and this server when judging
whether a request arrived after the
+ /// broker's deadline for it
([Broker.Request.QueryOptionKey#REQUEST_DEADLINE_MS]). Such a request is
answered
+ /// with a timeout error instead of being executed, but only when it
arrives more than this many milliseconds
+ /// after the deadline. The stamp approximates the broker's remaining wait
budget; the response wait starts later
+ /// during dispatch. A dispatch delay or a server clock ahead of the
broker's by more than this allowance can
+ /// therefore reject a request while the broker is still waiting. A server
clock behind the broker's only makes
+ /// the check more lenient. Rejections show up as
`QuerySchedulingTimeoutError` responses carrying the arrival time
+ /// and the deadline, and in the `QUERIES_EXPIRED_ON_ARRIVAL` meter. A
negative value disables the check.
+ public static final String CONFIG_OF_REQUEST_DEADLINE_CLOCK_SKEW_MS =
+ "pinot.server.request.deadline.clock.skew.ms";
+ public static final long DEFAULT_REQUEST_DEADLINE_CLOCK_SKEW_MS = 1_000L;
Review Comment:
Agreed on watching `QUERIES_EXPIRED_ON_ARRIVAL` after rollout. The window is
narrower than the gap suggests, though: the channel-lock wait and the write
happen after `AsyncQueryResponse` has started the broker's timer in
`QueryRouter#submitQuery`, so they do not widen the stamp-to-timer gap. That
gap is only the few statements between `setRequestDeadline` and `submitQuery`
(response-size options, the materialized-view split decision), so it takes a
pause of more than the allowance right there, and even then the server only
drops a request that arrives when the broker has less than (pause minus
allowance) of its budget left, i.e. one that spent its whole timeout in flight.
A request the server would have executed in that sliver would have timed out at
the broker anyway. If the meter shows otherwise the allowance is a config key.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java:
##########
@@ -150,16 +167,80 @@ protected void channelRead0(ChannelHandlerContext ctx,
ByteBuf msg) {
queryRequest.getTimerContext().startNewPhaseTimer(ServerQueryPhase.REQUEST_DESERIALIZATION,
queryArrivalTimeMs)
.stopAndRecord();
try {
+ if (rejectIfExpired(queryRequest, ctx, queryArrivalTimeMs)) {
+ return;
+ }
submitQuery(queryRequest, ctx, queryArrivalTimeMs);
} catch (Exception e) {
long requestId = queryRequest.getRequestId();
String tableNameWithType = queryRequest.getTableNameWithType();
- LOGGER.error("Caught exception while submitting query request: {}",
requestId, e);
+ LOGGER.error("Caught exception while handling query request: {}",
requestId, e);
sendErrorResponse(ctx, requestId, tableNameWithType, queryArrivalTimeMs,
DataTableBuilderFactory.getEmptyDataTable(), e);
}
}
+ /// Answers a request that reached this server after the broker's deadline
for it with a timeout error, without
+ /// executing it, and returns `true`; returns `false` when the request is to
be executed as usual.
+ ///
+ /// The broker sends its deadline as the `requestDeadlineMs` query option,
so a request from an old broker (no
+ /// option) is always executed. The deadline is compared against this
server's clock, which is why it has to be more
+ /// than [Server#CONFIG_OF_REQUEST_DEADLINE_CLOCK_SKEW_MS] in the past
before the request is dropped. Without this
+ /// check a request that spent minutes in the socket buffers of a stalled
connection, and is delivered when the
+ /// connection recovers, is executed with a full timeout budget measured
from its arrival, for a broker that gave up
+ /// on it long ago. The response is still sent: it costs a few bytes and
tells the broker side that the channel is
+ /// alive, even though the broker no longer has a query waiting for it. A
malformed option propagates to the caller,
+ /// which answers with the parse error under the right request id.
+ private boolean rejectIfExpired(ServerQueryRequest queryRequest,
ChannelHandlerContext ctx,
+ long queryArrivalTimeMs) {
+ if (_requestDeadlineClockSkewMs < 0) {
+ return false;
+ }
+ Long requestDeadlineMs =
QueryOptionsUtils.getRequestDeadlineMs(queryRequest.getQueryContext().getQueryOptions());
+ if (requestDeadlineMs == null) {
+ return false;
+ }
+ // Subtract rather than add, so that a huge deadline cannot overflow into
a rejection.
+ long lateByMs = queryArrivalTimeMs - requestDeadlineMs;
+ if (lateByMs <= _requestDeadlineClockSkewMs) {
Review Comment:
Thanks, that is the intent: a huge deadline from an old broker must not
overflow into a rejection, and both short-circuits keep old brokers and the
disabled config on the old path.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java:
##########
@@ -892,6 +892,7 @@ protected BrokerResponse doHandleRequest(long requestId,
String query, SqlNodeAn
errorMsgs.add(new
QueryProcessingException(QueryErrorCode.BROKER_TIMEOUT, errorMessage));
return BrokerResponseNative.fromBrokerErrors(errorMsgs);
}
+ setRequestDeadline(remainingTimeMs, offlineBrokerRequest,
realtimeBrokerRequest, null);
Review Comment:
Right, the `TimeoutException` path above is what keeps a zero or negative
deadline from ever being stamped, and the shared longer deadline is covered by
`testRequestDeadlineOnHybridTableFollowsTheLongerTimeout`.
--
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]
