raghavyadav01 commented on code in PR #20023:
URL: https://github.com/apache/pinot/pull/20023#discussion_r4448542185
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java:
##########
@@ -158,11 +175,14 @@ protected void channelRead0(ChannelHandlerContext ctx,
ByteBuf msg) {
queryRequest.getTimerContext().startNewPhaseTimer(ServerQueryPhase.REQUEST_DESERIALIZATION,
queryArrivalTimeMs)
.stopAndRecord();
try {
+ if (rejectIfExpired(queryRequest, ctx, queryArrivalTimeMs)) {
Review Comment:
By this point `ServerQueryRequest` has already compiled the query on the
event loop, so a recovered channel's whole backlog still pays deserialize +
compile per request before being dropped. Could the check move ahead of the
compile, reading the option from
`instanceRequest.getQuery().getPinotQuery().getQueryOptions()` and the table
from `getTableNameWithType(instanceRequest)`?
##########
pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java:
##########
@@ -1770,6 +1781,21 @@ 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 broker stamps the deadline a few statements
before it starts its own wait for the
+ /// servers, and network or channel-lock delays come after that wait has
started, so with synchronized clocks a
+ /// request is rejected only once the broker has stopped waiting for it
for longer than this allowance; neither a
+ /// slow hop nor a short `timeoutMs` changes that, since the query's
timeout is not part of the comparison. The one
+ /// exposure is a server clock running ahead of the broker's by more than
this allowance: it rejects requests that
+ /// arrive within the excess of their deadline, i.e. every request of a
query whose remaining timeout is shorter
+ /// than the excess. 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:
With 1s, a server whose clock drifts ~1.5s ahead rejects every request with
less than ~500ms remaining, so low-latency tables would see partial results on
every query to that server. The response still counts as alive and the code
isn't a client error, so nothing marks the server down. Since the stale
backlogs this targets are minutes old, could we default this to something like
10-30s?
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java:
##########
@@ -183,6 +203,68 @@ private void sendPingResponse(ChannelHandlerContext ctx,
long requestId) {
}
}
+ /// 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) {
+ return false;
+ }
+ long requestId = queryRequest.getRequestId();
+ String tableNameWithType = queryRequest.getTableNameWithType();
+ _serverMetrics.addMeteredTableValue(tableNameWithType,
ServerMeter.QUERIES_EXPIRED_ON_ARRIVAL, 1);
Review Comment:
This runs once per rejected request on the event loop during the burst, and
`addMeteredTableValue` without a reused meter does a name build and registry
lookup each time. Could we cache the meter per table, or pass a reused meter
here?
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java:
##########
@@ -183,6 +203,68 @@ private void sendPingResponse(ChannelHandlerContext ctx,
long requestId) {
}
}
+ /// 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) {
+ return false;
+ }
+ long requestId = queryRequest.getRequestId();
+ String tableNameWithType = queryRequest.getTableNameWithType();
+ _serverMetrics.addMeteredTableValue(tableNameWithType,
ServerMeter.QUERIES_EXPIRED_ON_ARRIVAL, 1);
+ logExpiredRequest(queryRequest, queryArrivalTimeMs, requestDeadlineMs,
lateByMs);
+ try {
+ DataTable dataTable = DataTableBuilderFactory.getEmptyDataTable();
+ dataTable.getMetadata().put(MetadataKey.REQUEST_ID.getName(),
Long.toString(requestId));
+ dataTable.addException(QueryErrorCode.QUERY_SCHEDULING_TIMEOUT,
Review Comment:
Reusing QUERY_SCHEDULING_TIMEOUT means that on the broker a clock-skew
rejection looks exactly like a scheduler queue timeout, and operators would go
chasing server capacity. Could we at least give the message a stable prefix to
grep for, and mention in the release notes that these show up under 240?
--
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]
