xiangfu0 commented on code in PR #20023:
URL: https://github.com/apache/pinot/pull/20023#discussion_r4431233974
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java:
##########
@@ -154,16 +171,81 @@ 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;
Review Comment:
Confirmed. Both operands are positive by the time the subtraction runs
(`getRequestDeadlineMs` returns `null` for anything non-positive and arrival is
epoch millis), so `arrival - deadline` cannot overflow, and a huge deadline
just comes out as a large negative lateness. `testHugeDeadlineIsExecuted` now
pins it with `Long.MAX_VALUE` and `Long.MAX_VALUE - 1`: both requests are
scheduled and answered normally.
(292f7053aa)
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java:
##########
@@ -154,16 +171,81 @@ 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:
The allowance covers this, because the query's timeout never enters the
comparison. The server drops a request only when it arrives more than the
allowance after the stamp, and the broker stops waiting `remainingTimeMs` after
it creates the `AsyncQueryResponse` in `QueryRouter#submitQuery`, which is a
few statements after the stamp. Network time and the channel-lock wait come
after that wait has started, so a request that lands more than the allowance
late has already been given up on by the broker, whether its timeout was 200 ms
or 30 s. A short timeout is only exposed to clock skew: a server clock ahead by
more than the allowance rejects every request of a query whose remaining
timeout is shorter than the excess.
Checking this turned up one real gap. A materialized-view split attempt that
fails falls back to the standard dispatch, and that dispatch waits the full
budget again, but the fallback requests kept the deadline stamped before the
attempt. The fallback now re-stamps, and
`testMaterializedViewSplitDispatchesFreshDeadlineAndSkipsExplain` gained a row
where the split fails; that row fails without the re-stamp. I also rewrote the
clock-skew config Javadoc to state exactly what the allowance covers, replacing
the earlier "a dispatch delay can reject" wording, which was broader than the
code.
(292f7053aa)
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java:
##########
@@ -307,6 +307,16 @@ public static Long getTimeoutMs(Map<String, String>
queryOptions) {
return checkedParseLongPositive(QueryOptionKey.TIMEOUT_MS,
timeoutMsString);
}
+ /// The broker's deadline for the request
([QueryOptionKey#REQUEST_DEADLINE_MS]) in epoch milliseconds, or `null`
+ /// when the option is absent or non-positive, in which case the request is
not checked against it.
+ @Nullable
+ public static Long getRequestDeadlineMs(Map<String, String> queryOptions) {
+ String requestDeadlineMsString =
queryOptions.get(QueryOptionKey.REQUEST_DEADLINE_MS);
+ Long requestDeadlineMs =
+ checkedParseLong(QueryOptionKey.REQUEST_DEADLINE_MS,
requestDeadlineMsString, Long.MIN_VALUE);
Review Comment:
Intended: a non-positive value can never be a real epoch deadline, so
treating it as absent keeps the old behaviour rather than rejecting or failing
the query. Added a one-line comment saying so next to the check.
(292f7053aa)
--
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]
