xiangfu0 commented on code in PR #20023:
URL: https://github.com/apache/pinot/pull/20023#discussion_r4366269412
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java:
##########
@@ -2401,6 +2402,23 @@ private long setQueryTimeout(String tableNameWithType,
Long logicalTableQueryTim
return remainingTimeMs;
}
+ /// Sets the time at which this broker stops waiting for the servers (now
plus the remaining time, for a hybrid
+ /// table the longer of the two table types' remaining times) into the query
options of every table type, see
+ /// [QueryOptionKey#REQUEST_DEADLINE_MS]. The absolute deadline lets a
server tell a request that sat in a stalled
+ /// connection from a live one, instead of granting it a fresh timeout
budget from its arrival. Always overwritten:
+ /// it is this broker's deadline, whatever the user set.
+ private static void setRequestDeadline(long remainingTimeMs, @Nullable
BrokerRequest offlineBrokerRequest,
+ @Nullable BrokerRequest realtimeBrokerRequest) {
+ String requestDeadlineMs = Long.toString(System.currentTimeMillis() +
remainingTimeMs);
Review Comment:
Intentional. Both readings hold, depending on the clock you compare against:
relative to compile-start + timeout the stamp is a hair late (the wall-clock
between the routing measurement and this line), but the clock that matters is
the broker's actual wait, and `AsyncQueryResponse` measures the same
`remainingTimeMs` from its own start in `QueryRouter#submitQuery`, which runs a
little after this line. Against that the stamp is a hair early (sub-millisecond
to a few ms, well inside the 1 s allowance), so a server can only ever drop a
request the broker has already given up on. Anchoring on the request arrival
time would move it earlier still, by the whole compile and routing time. Added
a sentence to the `setRequestDeadline` Javadoc in ad3caf843c so the next reader
does not have to redo the arithmetic.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java:
##########
@@ -150,16 +167,78 @@ 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) {
+ 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,
+ String.format("Request arrived on: %s at %d, %dms after the broker
deadline %d, and was not executed",
+ _instanceName, queryArrivalTimeMs, lateByMs, requestDeadlineMs));
+ sendResponse(ctx, requestId, tableNameWithType, queryArrivalTimeMs,
dataTable.toBytes());
+ } catch (Exception e) {
+ LOGGER.error("Exception while sending the expired response for request:
{}", requestId, e);
+ }
+ return true;
+ }
+
+ /// Logs at most one line per [#EXPIRED_REQUEST_LOG_INTERVAL_MS], carrying
the number of expired requests since
+ /// the previous line and the details of the latest one. The handler is
shared by every channel, hence the atomics.
+ private void logExpiredRequest(ServerQueryRequest queryRequest, long
queryArrivalTimeMs, long requestDeadlineMs,
+ long lateByMs) {
+ _expiredRequestsSinceLastLog.incrementAndGet();
+ long nextLogTimeMs = _nextExpiredRequestLogTimeMs.get();
+ if (queryArrivalTimeMs < nextLogTimeMs ||
!_nextExpiredRequestLogTimeMs.compareAndSet(nextLogTimeMs,
Review Comment:
Fair point. Switched the gate to `System.currentTimeMillis()` in ad3caf843c.
Same behaviour in practice (the arrival stamp is taken on the same thread
microseconds earlier), but the one-line-per-second intent no longer depends on
that observation.
--
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]
